From 06011bc799994b1fe98fdb5fa24507ebf9b73521 Mon Sep 17 00:00:00 2001 From: Alpha <2748109647@qq.com> Date: Sun, 25 Oct 2026 18:16:43 +0800 Subject: [PATCH 1/3] Add actor reentrancy support Signed-off-by: Alpha <2748109647@qq.com> --- .../java/io/dapr/examples/actors/README.md | 16 ++++ .../io/dapr/actors/client/DaprClientImpl.java | 21 +++-- .../internal/ActorReentrancyContext.java | 75 ++++++++++++++++ .../actors/runtime/ActorEventDispatcher.java | 30 +++++-- .../actors/runtime/ActorObjectSerializer.java | 15 ++++ .../actors/runtime/ActorReentrancyConfig.java | 70 +++++++++++++++ .../actors/runtime/ActorRuntimeConfig.java | 30 ++++++- .../dapr/actors/runtime/ActorTypeConfig.java | 26 +++++- .../actors/client/DaprGrpcClientTest.java | 22 +++++ .../runtime/ActorEventDispatcherTest.java | 53 ++++++++++- .../runtime/ActorObjectSerializerTest.java | 12 ++- .../io/dapr/springboot/DaprController.java | 26 ++++-- .../io/dapr/it/actors/ActorReentrancyIT.java | 89 +++++++++++++++++++ .../it/actors/app/MyActorGrpcService.java | 6 ++ .../io/dapr/it/actors/app/MyActorService.java | 6 ++ .../io/dapr/it/actors/app/ReentrantActor.java | 26 ++++++ .../it/actors/app/ReentrantActorImpl.java | 57 ++++++++++++ 17 files changed, 547 insertions(+), 33 deletions(-) create mode 100644 sdk-actors/src/main/java/io/dapr/actors/internal/ActorReentrancyContext.java create mode 100644 sdk-actors/src/main/java/io/dapr/actors/runtime/ActorReentrancyConfig.java create mode 100644 sdk-tests/src/test/java/io/dapr/it/actors/ActorReentrancyIT.java create mode 100644 sdk-tests/src/test/java/io/dapr/it/actors/app/ReentrantActor.java create mode 100644 sdk-tests/src/test/java/io/dapr/it/actors/app/ReentrantActorImpl.java diff --git a/examples/src/main/java/io/dapr/examples/actors/README.md b/examples/src/main/java/io/dapr/examples/actors/README.md index 4ed9e78c51..5c5edf0775 100644 --- a/examples/src/main/java/io/dapr/examples/actors/README.md +++ b/examples/src/main/java/io/dapr/examples/actors/README.md @@ -64,6 +64,22 @@ public class DemoActorService { ``` This application uses `ActorRuntime.getInstance().registerActor()` in order to register `DemoActorImpl` as an actor in the Dapr Actor runtime. Internally, it is using `DefaultObjectSerializer` for two properties: `objectSerializer` is for Dapr's sent and received objects, and `stateSerializer` is for objects to be persisted. + +### Enable actor reentrancy + +Reentrancy is disabled unless configured. It can be enabled for every actor type or overridden for a specific type. The same configuration is advertised by the HTTP actor host and the gRPC actor event stream: + +```java +ActorRuntime runtime = ActorRuntime.getInstance(); +runtime.getConfig().setReentrancy( + new ActorReentrancyConfig().setEnabled(true).setMaxStackDepth(5)); + +runtime.getConfig().addActorTypeConfig(new ActorTypeConfig() + .setActorTypeName("DemoActor") + .setReentrancy(new ActorReentrancyConfig().setEnabled(false))); +``` + +Actor methods that make nested actor calls should compose the returned `Mono` so the invocation context can carry Dapr's reentrancy id across scheduler changes and concurrent calls. `DaprApplication.start()` method will run the Spring Boot [DaprApplication](https://github.com/dapr/java-sdk/blob/master/sdk-tests/src/test/java/io/dapr/it/actors/services/springboot/DaprApplication.java), which registers the Dapr Spring Boot controller [DaprController](https://github.com/dapr/java-sdk/blob/master/sdk-springboot/src/main/java/io/dapr/springboot/DaprController.java). This controller contains all Actor methods implemented as endpoints. The Dapr's sidecar will call into the controller. diff --git a/sdk-actors/src/main/java/io/dapr/actors/client/DaprClientImpl.java b/sdk-actors/src/main/java/io/dapr/actors/client/DaprClientImpl.java index 4e0fdfac4d..96f7fb6606 100644 --- a/sdk-actors/src/main/java/io/dapr/actors/client/DaprClientImpl.java +++ b/sdk-actors/src/main/java/io/dapr/actors/client/DaprClientImpl.java @@ -14,6 +14,7 @@ package io.dapr.actors.client; import com.google.protobuf.ByteString; +import io.dapr.actors.internal.ActorReentrancyContext; import io.dapr.client.resiliency.ResiliencyOptions; import io.dapr.exceptions.DaprException; import io.dapr.internal.grpc.DaprClientGrpcInterceptors; @@ -80,19 +81,21 @@ class DaprClientImpl implements DaprClient { */ @Override public Mono invoke(String actorType, String actorId, String methodName, byte[] jsonPayload) { - DaprActorsProtos.InvokeActorRequest req = - DaprActorsProtos.InvokeActorRequest.newBuilder() + return Mono.deferContextual(context -> { + DaprActorsProtos.InvokeActorRequest.Builder requestBuilder = DaprActorsProtos.InvokeActorRequest.newBuilder() .setActorType(actorType) .setActorId(actorId) .setMethod(methodName) .putAllMetadata(this.metadata) - .setData(jsonPayload == null ? ByteString.EMPTY : ByteString.copyFrom(jsonPayload)) - .build(); - return Mono.deferContextual( - context -> this.createMono( - it -> this.grpcInterceptors.intercept(client, context).invokeActor(req, it) - ) - ).map(r -> r.getData().toByteArray()); + .setData(jsonPayload == null ? ByteString.EMPTY : ByteString.copyFrom(jsonPayload)); + String reentrancyId = ActorReentrancyContext.getReentrancyId(context); + if (reentrancyId != null) { + requestBuilder.putMetadata(ActorReentrancyContext.REENTRANCY_ID_HEADER, reentrancyId); + } + DaprActorsProtos.InvokeActorRequest request = requestBuilder.build(); + return this.createMono( + it -> this.grpcInterceptors.intercept(client, context).invokeActor(request, it)); + }).map(r -> r.getData().toByteArray()); } private Mono createMono(Consumer> consumer) { diff --git a/sdk-actors/src/main/java/io/dapr/actors/internal/ActorReentrancyContext.java b/sdk-actors/src/main/java/io/dapr/actors/internal/ActorReentrancyContext.java new file mode 100644 index 0000000000..dc9fee6688 --- /dev/null +++ b/sdk-actors/src/main/java/io/dapr/actors/internal/ActorReentrancyContext.java @@ -0,0 +1,75 @@ +/* + * Copyright 2026 The Dapr Authors + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * http://www.apache.org/licenses/LICENSE-2.0 + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and +limitations under the License. +*/ + +package io.dapr.actors.internal; + +import reactor.core.publisher.Mono; +import reactor.util.context.ContextView; + +import java.util.Map; + +/** + * Internal helpers for carrying an incoming actor reentrancy id through a Reactor invocation. + */ +public final class ActorReentrancyContext { + + /** Metadata key used by Dapr to identify an actor reentrant call chain. */ + public static final String REENTRANCY_ID_HEADER = "Dapr-Reentrancy-Id"; + + private static final String CONTEXT_KEY = ActorReentrancyContext.class.getName() + ".id"; + + private ActorReentrancyContext() { + } + + /** + * Adds the reentrancy id to the subscriber context for the lifetime of this publisher. + * + * @param publisher Publisher for the actor callback. + * @param reentrancyId Id received from Dapr, or {@code null} when this is not a reentrant call. + * @param Publisher result type. + * @return The publisher with the invocation-scoped context value. + */ + public static Mono withReentrancyId(Mono publisher, String reentrancyId) { + if (reentrancyId == null || reentrancyId.isEmpty()) { + return publisher; + } + return publisher.contextWrite(context -> context.put(CONTEXT_KEY, reentrancyId)); + } + + /** + * Reads the reentrancy id from a Reactor subscriber context. + * + * @param context Subscriber context. + * @return The id, or {@code null} when no actor invocation is active. + */ + public static String getReentrancyId(ContextView context) { + return context.getOrDefault(CONTEXT_KEY, null); + } + + /** + * Reads the reentrancy id from callback metadata, ignoring header-name casing. + * + * @param metadata Callback metadata. + * @return The id, or {@code null} when Dapr did not send one. + */ + public static String getReentrancyId(Map metadata) { + if (metadata == null) { + return null; + } + return metadata.entrySet().stream() + .filter(entry -> REENTRANCY_ID_HEADER.equalsIgnoreCase(entry.getKey())) + .map(Map.Entry::getValue) + .findFirst() + .orElse(null); + } +} diff --git a/sdk-actors/src/main/java/io/dapr/actors/runtime/ActorEventDispatcher.java b/sdk-actors/src/main/java/io/dapr/actors/runtime/ActorEventDispatcher.java index b0c6298c31..6179c9c482 100644 --- a/sdk-actors/src/main/java/io/dapr/actors/runtime/ActorEventDispatcher.java +++ b/sdk-actors/src/main/java/io/dapr/actors/runtime/ActorEventDispatcher.java @@ -19,7 +19,9 @@ import com.google.protobuf.ByteString; import com.google.protobuf.BytesValue; import com.google.protobuf.InvalidProtocolBufferException; +import io.dapr.actors.internal.ActorReentrancyContext; import io.dapr.v1.DaprActorsProtos.ActorEntityConfig; +import io.dapr.v1.DaprActorsProtos.ActorReentrancyConfig; import io.dapr.v1.DaprActorsProtos.SubscribeActorEventsRequestAlpha1; import io.dapr.v1.DaprActorsProtos.SubscribeActorEventsRequestDeactivateResponseAlpha1; import io.dapr.v1.DaprActorsProtos.SubscribeActorEventsRequestFailedAlpha1; @@ -70,6 +72,9 @@ static SubscribeActorEventsRequestAlpha1 buildInitialRequest(ActorRuntimeConfig if (config.getDrainBalancedActors() != null) { builder.setDrainRebalancedActors(config.getDrainBalancedActors()); } + if (config.getReentrancy() != null) { + builder.setReentrancy(toProto(config.getReentrancy())); + } for (ActorTypeConfig typeConfig : config.getActorTypeConfigs()) { ActorEntityConfig.Builder entityConfig = ActorEntityConfig.newBuilder() @@ -83,6 +88,9 @@ static SubscribeActorEventsRequestAlpha1 buildInitialRequest(ActorRuntimeConfig if (typeConfig.getDrainBalancedActors() != null) { entityConfig.setDrainRebalancedActors(typeConfig.getDrainBalancedActors()); } + if (typeConfig.getReentrancy() != null) { + entityConfig.setReentrancy(toProto(typeConfig.getReentrancy())); + } builder.addEntitiesConfig(entityConfig); } @@ -148,7 +156,9 @@ private static Mono invoke( byte[] payload = request.getData().isEmpty() ? null : request.getData().toByteArray(); String contentType = runtime.getObjectSerializer(request.getActorType()).getContentType(); - return runtime.invoke(request.getActorType(), request.getActorId(), request.getMethod(), payload) + return ActorReentrancyContext.withReentrancyId( + runtime.invoke(request.getActorType(), request.getActorId(), request.getMethod(), payload), + ActorReentrancyContext.getReentrancyId(request.getMetadataMap())) .map(ByteString::copyFrom) .defaultIfEmpty(ByteString.EMPTY) .map(data -> SubscribeActorEventsRequestAlpha1.newBuilder() @@ -167,9 +177,10 @@ private static Mono reminder( return Mono.just(notRegistered(id, request.getActorType())); } - return Mono.fromCallable(() -> buildCallbackBody( + return ActorReentrancyContext.withReentrancyId(Mono.fromCallable(() -> buildCallbackBody( null, request.getDueTime(), request.getPeriod(), request.hasData() ? request.getData() : null)) - .flatMap(body -> runtime.invokeReminder(request.getActorType(), request.getActorId(), request.getName(), body)) + .flatMap(body -> runtime.invokeReminder(request.getActorType(), request.getActorId(), request.getName(), body)), + ActorReentrancyContext.getReentrancyId(request.getMetadataMap())) .then(Mono.fromSupplier(() -> SubscribeActorEventsRequestAlpha1.newBuilder() .setReminderResponse(SubscribeActorEventsRequestReminderResponseAlpha1.newBuilder().setId(id)) .build())) @@ -183,10 +194,11 @@ private static Mono timer( return Mono.just(notRegistered(id, request.getActorType())); } - return Mono.fromCallable(() -> buildCallbackBody( + return ActorReentrancyContext.withReentrancyId(Mono.fromCallable(() -> buildCallbackBody( request.getCallback(), request.getDueTime(), request.getPeriod(), request.hasData() ? request.getData() : null)) - .flatMap(body -> runtime.invokeTimer(request.getActorType(), request.getActorId(), request.getName(), body)) + .flatMap(body -> runtime.invokeTimer(request.getActorType(), request.getActorId(), request.getName(), body)), + ActorReentrancyContext.getReentrancyId(request.getMetadataMap())) .then(Mono.fromSupplier(() -> SubscribeActorEventsRequestAlpha1.newBuilder() .setTimerResponse(SubscribeActorEventsRequestReminderResponseAlpha1.newBuilder().setId(id)) .build())) @@ -278,4 +290,12 @@ private static com.google.protobuf.Duration toProto(Duration duration) { .setNanos(duration.getNano()) .build(); } + + private static ActorReentrancyConfig toProto(io.dapr.actors.runtime.ActorReentrancyConfig config) { + ActorReentrancyConfig.Builder builder = ActorReentrancyConfig.newBuilder().setEnabled(config.isEnabled()); + if (config.getMaxStackDepth() != null) { + builder.setMaxStackDepth(config.getMaxStackDepth()); + } + return builder.build(); + } } diff --git a/sdk-actors/src/main/java/io/dapr/actors/runtime/ActorObjectSerializer.java b/sdk-actors/src/main/java/io/dapr/actors/runtime/ActorObjectSerializer.java index cf78ea6810..99f61c7b5f 100644 --- a/sdk-actors/src/main/java/io/dapr/actors/runtime/ActorObjectSerializer.java +++ b/sdk-actors/src/main/java/io/dapr/actors/runtime/ActorObjectSerializer.java @@ -160,6 +160,9 @@ private byte[] serialize(ActorRuntimeConfig config) throws IOException { if (config.getRemindersStoragePartitions() != null) { generator.writeNumberField("remindersStoragePartitions", config.getRemindersStoragePartitions()); } + if (config.getReentrancy() != null) { + writeReentrancy(generator, config.getReentrancy()); + } if (!config.getActorTypeConfigs().isEmpty()) { generator.writeArrayFieldStart("entitiesConfig"); @@ -188,6 +191,9 @@ private byte[] serialize(ActorRuntimeConfig config) throws IOException { if (actorTypeConfig.getRemindersStoragePartitions() != null) { generator.writeNumberField("remindersStoragePartitions", actorTypeConfig.getRemindersStoragePartitions()); } + if (actorTypeConfig.getReentrancy() != null) { + writeReentrancy(generator, actorTypeConfig.getReentrancy()); + } generator.writeEndObject(); } @@ -200,6 +206,15 @@ private byte[] serialize(ActorRuntimeConfig config) throws IOException { } } + private void writeReentrancy(JsonGenerator generator, ActorReentrancyConfig config) throws IOException { + generator.writeObjectFieldStart("reentrancy"); + generator.writeBooleanField("enabled", config.isEnabled()); + if (config.getMaxStackDepth() != null) { + generator.writeNumberField("maxStackDepth", config.getMaxStackDepth()); + } + generator.writeEndObject(); + } + /** * {@inheritDoc} */ diff --git a/sdk-actors/src/main/java/io/dapr/actors/runtime/ActorReentrancyConfig.java b/sdk-actors/src/main/java/io/dapr/actors/runtime/ActorReentrancyConfig.java new file mode 100644 index 0000000000..cff9bee683 --- /dev/null +++ b/sdk-actors/src/main/java/io/dapr/actors/runtime/ActorReentrancyConfig.java @@ -0,0 +1,70 @@ +/* + * Copyright 2026 The Dapr Authors + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * http://www.apache.org/licenses/LICENSE-2.0 + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and +limitations under the License. +*/ + +package io.dapr.actors.runtime; + +/** + * Configures reentrant actor calls. + */ +public class ActorReentrancyConfig { + + private boolean enabled; + + private Integer maxStackDepth; + + /** + * Creates an actor reentrancy configuration. Reentrancy is disabled by default. + */ + public ActorReentrancyConfig() { + } + + /** + * Gets whether actor reentrancy is enabled. + * + * @return {@code true} when reentrant calls are allowed. + */ + public boolean isEnabled() { + return enabled; + } + + /** + * Sets whether actor reentrancy is enabled. + * + * @param enabled Whether actor reentrancy is enabled. + * @return This instance. + */ + public ActorReentrancyConfig setEnabled(boolean enabled) { + this.enabled = enabled; + return this; + } + + /** + * Gets the maximum number of nested reentrant calls. + * + * @return Maximum reentrancy stack depth, or {@code null} to use Dapr's default. + */ + public Integer getMaxStackDepth() { + return maxStackDepth; + } + + /** + * Sets the maximum number of nested reentrant calls. + * + * @param maxStackDepth Maximum reentrancy stack depth, or {@code null} to use Dapr's default. + * @return This instance. + */ + public ActorReentrancyConfig setMaxStackDepth(Integer maxStackDepth) { + this.maxStackDepth = maxStackDepth; + return this; + } +} diff --git a/sdk-actors/src/main/java/io/dapr/actors/runtime/ActorRuntimeConfig.java b/sdk-actors/src/main/java/io/dapr/actors/runtime/ActorRuntimeConfig.java index 7e42d49ae6..944d54194a 100644 --- a/sdk-actors/src/main/java/io/dapr/actors/runtime/ActorRuntimeConfig.java +++ b/sdk-actors/src/main/java/io/dapr/actors/runtime/ActorRuntimeConfig.java @@ -36,6 +36,8 @@ public class ActorRuntimeConfig { private volatile Integer remindersStoragePartitions; + private volatile ActorReentrancyConfig reentrancy; + private volatile List actorTypeConfigs = new CopyOnWriteArrayList<>(); /** @@ -169,12 +171,32 @@ public ActorRuntimeConfig setRemindersStoragePartitions(Integer remindersStorage } /** - * Adds a actor type config to the list of actors type configs. - * - * @param actorTypeName Actor type that was registered. + * Gets the default reentrancy configuration for all actor types. + * + * @return Default reentrancy configuration, or {@code null} to use Dapr's default. + */ + public ActorReentrancyConfig getReentrancy() { + return reentrancy; + } + + /** + * Sets the default reentrancy configuration for all actor types. + * + * @param reentrancy Default reentrancy configuration, or {@code null} to use Dapr's default. + * @return This instance. + */ + public ActorRuntimeConfig setReentrancy(ActorReentrancyConfig reentrancy) { + this.reentrancy = reentrancy; + return this; + } + + /** + * Adds an actor type config to the list of actor type configs. + * + * @param config Actor type config to add. * @return This instance. */ - ActorRuntimeConfig addActorTypeConfig(ActorTypeConfig config) { + public ActorRuntimeConfig addActorTypeConfig(ActorTypeConfig config) { if (config == null) { throw new IllegalArgumentException("Add actor type config failed."); } diff --git a/sdk-actors/src/main/java/io/dapr/actors/runtime/ActorTypeConfig.java b/sdk-actors/src/main/java/io/dapr/actors/runtime/ActorTypeConfig.java index ce3d2d524e..da6bf1ae7f 100644 --- a/sdk-actors/src/main/java/io/dapr/actors/runtime/ActorTypeConfig.java +++ b/sdk-actors/src/main/java/io/dapr/actors/runtime/ActorTypeConfig.java @@ -32,10 +32,12 @@ public class ActorTypeConfig { private volatile Integer remindersStoragePartitions; + private volatile ActorReentrancyConfig reentrancy; + /** * Instantiates a new config for the Actor Runtime. */ - ActorTypeConfig() { + public ActorTypeConfig() { } /** @@ -44,7 +46,7 @@ public class ActorTypeConfig { * @param actorTypeName Actor type that was registered. * @return This instance. */ - ActorTypeConfig setActorTypeName(String actorTypeName) { + public ActorTypeConfig setActorTypeName(String actorTypeName) { if (actorTypeName == null) { throw new IllegalArgumentException("Registered actor must have a type name."); } @@ -161,4 +163,24 @@ public ActorTypeConfig setRemindersStoragePartitions(Integer remindersStoragePar this.remindersStoragePartitions = remindersStoragePartitions; return this; } + + /** + * Gets this actor type's reentrancy configuration. + * + * @return Reentrancy configuration, or {@code null} to inherit the runtime default. + */ + public ActorReentrancyConfig getReentrancy() { + return reentrancy; + } + + /** + * Sets this actor type's reentrancy configuration. + * + * @param reentrancy Reentrancy configuration, or {@code null} to inherit the runtime default. + * @return This instance. + */ + public ActorTypeConfig setReentrancy(ActorReentrancyConfig reentrancy) { + this.reentrancy = reentrancy; + return this; + } } diff --git a/sdk-actors/src/test/java/io/dapr/actors/client/DaprGrpcClientTest.java b/sdk-actors/src/test/java/io/dapr/actors/client/DaprGrpcClientTest.java index e2f75a2b41..b56babfeff 100644 --- a/sdk-actors/src/test/java/io/dapr/actors/client/DaprGrpcClientTest.java +++ b/sdk-actors/src/test/java/io/dapr/actors/client/DaprGrpcClientTest.java @@ -14,6 +14,7 @@ package io.dapr.actors.client; import com.google.protobuf.ByteString; +import io.dapr.actors.internal.ActorReentrancyContext; import io.dapr.v1.DaprActorsProtos; import io.dapr.v1.DaprGrpc; import io.grpc.ManagedChannel; @@ -29,6 +30,8 @@ import reactor.core.publisher.Mono; import java.io.IOException; +import java.util.Set; +import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.ExecutionException; import static io.dapr.actors.TestUtils.assertThrowsDaprException; @@ -53,6 +56,8 @@ public class DaprGrpcClientTest { private static final byte[] RESPONSE_PAYLOAD = "\"OK\"".getBytes(); + private final Set observedReentrancyIds = ConcurrentHashMap.newKeySet(); + @Rule public final GrpcCleanupRule grpcCleanup = new GrpcCleanupRule(); @@ -67,6 +72,11 @@ public void invokeActor(DaprActorsProtos.InvokeActorRequest request, switch (request.getActorId()) { case ACTOR_ID_OK: assertArrayEquals(REQUEST_PAYLOAD, request.getData().toByteArray()); + String reentrancyId = request.getMetadataMap() + .get(ActorReentrancyContext.REENTRANCY_ID_HEADER); + if (reentrancyId != null) { + observedReentrancyIds.add(reentrancyId); + } responseObserver.onNext( DaprActorsProtos.InvokeActorResponse.newBuilder().setData(ByteString.copyFrom(RESPONSE_PAYLOAD)) .build()); @@ -115,6 +125,18 @@ public void invoke() { assertArrayEquals(RESPONSE_PAYLOAD, result.block()); } + @Test + public void invokeUsesReentrancyIdFromEachSubscriberContext() { + Mono.when( + ActorReentrancyContext.withReentrancyId( + client.invoke(ACTOR_TYPE, ACTOR_ID_OK, METHOD_NAME, REQUEST_PAYLOAD), "chain-a"), + ActorReentrancyContext.withReentrancyId( + client.invoke(ACTOR_TYPE, ACTOR_ID_OK, METHOD_NAME, REQUEST_PAYLOAD), "chain-b")) + .block(); + + org.junit.jupiter.api.Assertions.assertEquals(Set.of("chain-a", "chain-b"), observedReentrancyIds); + } + @Test public void invokeNullPayload() { Mono result = client.invoke(ACTOR_TYPE, ACTOR_ID_NULL_INPUT, METHOD_NAME, null); diff --git a/sdk-actors/src/test/java/io/dapr/actors/runtime/ActorEventDispatcherTest.java b/sdk-actors/src/test/java/io/dapr/actors/runtime/ActorEventDispatcherTest.java index fa59799994..531aec439e 100644 --- a/sdk-actors/src/test/java/io/dapr/actors/runtime/ActorEventDispatcherTest.java +++ b/sdk-actors/src/test/java/io/dapr/actors/runtime/ActorEventDispatcherTest.java @@ -18,6 +18,7 @@ import com.google.protobuf.BytesValue; import io.dapr.actors.ActorId; import io.dapr.actors.ActorType; +import io.dapr.actors.internal.ActorReentrancyContext; import io.dapr.serializer.DefaultObjectSerializer; import io.dapr.utils.TypeRef; import io.dapr.v1.DaprActorsProtos.ActorEntityConfig; @@ -106,6 +107,24 @@ public void lookup() throws NoSuchMethodException { } } + public interface ReentrancyActor { + Mono currentReentrancyId(); + } + + @ActorType(name = "ReentrancyActor") + public static class ReentrancyActorImpl extends AbstractActor implements ReentrancyActor { + + public ReentrancyActorImpl(ActorRuntimeContext runtimeContext, ActorId id) { + super(runtimeContext, id); + } + + @Override + public Mono currentReentrancyId() { + return Mono.deferContextual(context -> + Mono.justOrEmpty(ActorReentrancyContext.getReentrancyId(context))); + } + } + private ActorRuntime runtime; static ActorRuntime newRuntime() throws Exception { @@ -128,6 +147,7 @@ public void setup() throws Exception { runtime.registerActor(ActorRuntimeTest.MyActorImpl.class); runtime.registerActor(MyRemindableActorImpl.class); runtime.registerActor(ThrowingActorImpl.class); + runtime.registerActor(ReentrancyActorImpl.class); } @Test @@ -136,9 +156,11 @@ public void initialRequestCarriesTypesAndConfig() { .setActorIdleTimeout(Duration.ofMinutes(2)) .setDrainOngoingCallTimeout(Duration.ofMillis(1500)) .setDrainBalancedActors(true) + .setReentrancy(new io.dapr.actors.runtime.ActorReentrancyConfig().setEnabled(true).setMaxStackDepth(5)) .addActorTypeConfig(new ActorTypeConfig() .setActorTypeName("MyGreatActor") - .setActorIdleTimeout(Duration.ofSeconds(30))) + .setActorIdleTimeout(Duration.ofSeconds(30)) + .setReentrancy(new io.dapr.actors.runtime.ActorReentrancyConfig().setEnabled(false).setMaxStackDepth(2))) .addActorTypeConfig(new ActorTypeConfig() .setActorTypeName("MyRemindableActor") .setDrainOngoingCallTimeout(Duration.ofSeconds(5)) @@ -147,15 +169,21 @@ public void initialRequestCarriesTypesAndConfig() { SubscribeActorEventsRequestInitialAlpha1 initial = ActorEventDispatcher.buildInitialRequest(runtime.getConfig()).getInitialRequest(); - assertEquals(Arrays.asList("MyGreatActor", "MyRemindableActor", "ThrowingActor"), initial.getEntitiesList()); + assertEquals(Arrays.asList("MyGreatActor", "MyRemindableActor", "ThrowingActor", "ReentrancyActor"), + initial.getEntitiesList()); assertEquals(120, initial.getActorIdleTimeout().getSeconds()); assertEquals(1, initial.getDrainOngoingCallTimeout().getSeconds()); assertEquals(500_000_000, initial.getDrainOngoingCallTimeout().getNanos()); assertTrue(initial.getDrainRebalancedActors()); - assertFalse(initial.hasReentrancy()); + assertTrue(initial.hasReentrancy()); + assertTrue(initial.getReentrancy().getEnabled()); + assertEquals(5, initial.getReentrancy().getMaxStackDepth()); ActorEntityConfig entityConfig = initial.getEntitiesConfig(0); assertEquals(Arrays.asList("MyGreatActor"), entityConfig.getEntitiesList()); assertEquals(30, entityConfig.getActorIdleTimeout().getSeconds()); + assertTrue(entityConfig.hasReentrancy()); + assertFalse(entityConfig.getReentrancy().getEnabled()); + assertEquals(2, entityConfig.getReentrancy().getMaxStackDepth()); assertFalse(entityConfig.hasDrainOngoingCallTimeout()); ActorEntityConfig secondConfig = initial.getEntitiesConfig(1); assertFalse(secondConfig.hasActorIdleTimeout()); @@ -190,6 +218,24 @@ public void invokePassesData() throws Exception { assertEquals("payload", LAST_TIMER_DATA.get()); } + @Test + public void invokeCarriesReentrancyIdInInvocationContext() throws Exception { + String reentrancyId = "chain-123"; + String actorId = UUID.randomUUID().toString(); + SubscribeActorEventsResponseAlpha1 request = SubscribeActorEventsResponseAlpha1.newBuilder() + .setInvokeRequest(SubscribeActorEventsResponseInvokeRequestAlpha1.newBuilder() + .setId("reentrant") + .setActorType("ReentrancyActor") + .setActorId(actorId) + .setMethod("currentReentrancyId") + .putMetadata(ActorReentrancyContext.REENTRANCY_ID_HEADER, reentrancyId)) + .build(); + + SubscribeActorEventsRequestAlpha1 response = dispatch(request); + + assertEquals(reentrancyId, SERIALIZER.deserialize(response.getInvokeResponse().getData().toByteArray(), String.class)); + } + @Test public void callbacksForUnknownActorTypeAreNotFound() { String actorId = UUID.randomUUID().toString(); @@ -217,6 +263,7 @@ public void initialRequestLeavesUnsetConfigUnset() { assertFalse(initial.hasActorIdleTimeout()); assertFalse(initial.hasDrainOngoingCallTimeout()); assertFalse(initial.hasDrainRebalancedActors()); + assertFalse(initial.hasReentrancy()); } @Test diff --git a/sdk-actors/src/test/java/io/dapr/actors/runtime/ActorObjectSerializerTest.java b/sdk-actors/src/test/java/io/dapr/actors/runtime/ActorObjectSerializerTest.java index a5178bf035..d7f3603c48 100644 --- a/sdk-actors/src/test/java/io/dapr/actors/runtime/ActorObjectSerializerTest.java +++ b/sdk-actors/src/test/java/io/dapr/actors/runtime/ActorObjectSerializerTest.java @@ -149,7 +149,8 @@ public void serializeActorRuntimeConfigWithAllAttributes() throws IOException { .setActorScanInterval(Duration.ofSeconds(30)) .setDrainOngoingCallTimeout(Duration.ofMinutes(1)) .setDrainBalancedActors(true) - .setRemindersStoragePartitions(7); + .setRemindersStoragePartitions(7) + .setReentrancy(new ActorReentrancyConfig().setEnabled(true).setMaxStackDepth(5)); byte[] serialized = SERIALIZER.serialize(config); @@ -158,7 +159,8 @@ public void serializeActorRuntimeConfigWithAllAttributes() throws IOException { + "\"actorScanInterval\":\"0h0m30s0ms\"," + "\"drainOngoingCallTimeout\":\"0h1m0s0ms\"," + "\"drainBalancedActors\":true," - + "\"remindersStoragePartitions\":7}"; + + "\"remindersStoragePartitions\":7," + + "\"reentrancy\":{\"enabled\":true,\"maxStackDepth\":5}}"; Assertions.assertEquals(OBJECT_MAPPER.readTree(expected), OBJECT_MAPPER.readTree(serialized)); } @@ -168,7 +170,8 @@ public void serializeActorRuntimeConfigWithActorTypeConfig() throws IOException .setActorTypeName("MyActor") .setActorIdleTimeout(Duration.ofHours(2)) .setDrainBalancedActors(false) - .setRemindersStoragePartitions(3); + .setRemindersStoragePartitions(3) + .setReentrancy(new ActorReentrancyConfig().setEnabled(false).setMaxStackDepth(2)); ActorRuntimeConfig config = new ActorRuntimeConfig() .addRegisteredActorType("MyActor") .addActorTypeConfig(actorTypeConfig); @@ -180,7 +183,8 @@ public void serializeActorRuntimeConfigWithActorTypeConfig() throws IOException + "\"entities\":[\"MyActor\"]," + "\"actorIdleTimeout\":\"2h0m0s0ms\"," + "\"drainBalancedActors\":false," - + "\"remindersStoragePartitions\":3}]}"; + + "\"remindersStoragePartitions\":3," + + "\"reentrancy\":{\"enabled\":false,\"maxStackDepth\":2}}]}"; Assertions.assertEquals(OBJECT_MAPPER.readTree(expected), OBJECT_MAPPER.readTree(serialized)); } diff --git a/sdk-springboot/src/main/java/io/dapr/springboot/DaprController.java b/sdk-springboot/src/main/java/io/dapr/springboot/DaprController.java index 35794e826d..c283585549 100644 --- a/sdk-springboot/src/main/java/io/dapr/springboot/DaprController.java +++ b/sdk-springboot/src/main/java/io/dapr/springboot/DaprController.java @@ -13,6 +13,7 @@ package io.dapr.springboot; +import io.dapr.actors.internal.ActorReentrancyContext; import io.dapr.actors.runtime.ActorRuntime; import io.dapr.serializer.DefaultObjectSerializer; import org.springframework.http.MediaType; @@ -21,6 +22,7 @@ import org.springframework.web.bind.annotation.PathVariable; import org.springframework.web.bind.annotation.PutMapping; import org.springframework.web.bind.annotation.RequestBody; +import org.springframework.web.bind.annotation.RequestHeader; import org.springframework.web.bind.annotation.RestController; import reactor.core.publisher.Mono; @@ -82,14 +84,18 @@ public Mono deactivateActor(@PathVariable("type") String type, * @param id Actor Id. * @param method Actor method. * @param body Raw request body. + * @param reentrancyId Dapr reentrancy id, if present. * @return Raw response body. */ @PutMapping(path = "/actors/{type}/{id}/method/{method}") public Mono invokeActorMethod(@PathVariable("type") String type, @PathVariable("id") String id, @PathVariable("method") String method, - @RequestBody(required = false) byte[] body) { - return ActorRuntime.getInstance().invoke(type, id, method, body); + @RequestBody(required = false) byte[] body, + @RequestHeader(value = ActorReentrancyContext.REENTRANCY_ID_HEADER, + required = false) String reentrancyId) { + return ActorReentrancyContext.withReentrancyId( + ActorRuntime.getInstance().invoke(type, id, method, body), reentrancyId); } /** @@ -98,14 +104,18 @@ public Mono invokeActorMethod(@PathVariable("type") String type, * @param id Actor Id. * @param timer Actor timer's name. * @param body Raw request's body. + * @param reentrancyId Dapr reentrancy id, if present. * @return Void. */ @PutMapping(path = "/actors/{type}/{id}/method/timer/{timer}") public Mono invokeActorTimer(@PathVariable("type") String type, @PathVariable("id") String id, @PathVariable("timer") String timer, - @RequestBody byte[] body) { - return ActorRuntime.getInstance().invokeTimer(type, id, timer, body); + @RequestBody byte[] body, + @RequestHeader(value = ActorReentrancyContext.REENTRANCY_ID_HEADER, + required = false) String reentrancyId) { + return ActorReentrancyContext.withReentrancyId( + ActorRuntime.getInstance().invokeTimer(type, id, timer, body), reentrancyId); } /** @@ -114,13 +124,17 @@ public Mono invokeActorTimer(@PathVariable("type") String type, * @param id Actor Id. * @param reminder Actor reminder's name. * @param body Raw request's body. + * @param reentrancyId Dapr reentrancy id, if present. * @return Void. */ @PutMapping(path = "/actors/{type}/{id}/method/remind/{reminder}") public Mono invokeActorReminder(@PathVariable("type") String type, @PathVariable("id") String id, @PathVariable("reminder") String reminder, - @RequestBody(required = false) byte[] body) { - return ActorRuntime.getInstance().invokeReminder(type, id, reminder, body); + @RequestBody(required = false) byte[] body, + @RequestHeader(value = ActorReentrancyContext.REENTRANCY_ID_HEADER, + required = false) String reentrancyId) { + return ActorReentrancyContext.withReentrancyId( + ActorRuntime.getInstance().invokeReminder(type, id, reminder, body), reentrancyId); } } diff --git a/sdk-tests/src/test/java/io/dapr/it/actors/ActorReentrancyIT.java b/sdk-tests/src/test/java/io/dapr/it/actors/ActorReentrancyIT.java new file mode 100644 index 0000000000..f9ffd7e6d7 --- /dev/null +++ b/sdk-tests/src/test/java/io/dapr/it/actors/ActorReentrancyIT.java @@ -0,0 +1,89 @@ +/* + * Copyright 2026 The Dapr Authors + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * http://www.apache.org/licenses/LICENSE-2.0 + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. +*/ + +package io.dapr.it.actors; + +import io.dapr.actors.ActorId; +import io.dapr.actors.client.ActorClient; +import io.dapr.actors.client.ActorProxyBuilder; +import io.dapr.it.AppRun; +import io.dapr.it.actors.app.MyActorGrpcService; +import io.dapr.it.actors.app.MyActorService; +import io.dapr.it.actors.app.ReentrantActor; +import io.dapr.it.actors.app.ReentrantActorImpl; +import io.dapr.it.containers.BaseContainerIT; +import io.dapr.testcontainers.DaprContainer; +import io.dapr.testcontainers.DaprProtocol; +import org.junit.jupiter.api.BeforeAll; +import org.junit.jupiter.api.Test; + +import java.util.UUID; + +import static org.junit.jupiter.api.Assertions.assertEquals; + +/** Verifies that an incoming reentrancy id is preserved for nested actor calls over both host protocols. */ +public class ActorReentrancyIT extends BaseContainerIT { + + private static DaprContainer httpDapr; + + private static DaprContainer grpcDapr; + + private static ActorClient httpActorClient; + + private static ActorClient grpcActorClient; + + @BeforeAll + public static void start() throws Exception { + var httpPair = startAppAndAttach( + "actor-reentrancy-http-it", + MyActorService.class, + AppRun.AppProtocol.HTTP, + appPort -> daprBuilder("actor-reentrancy-http-it") + .withAppPort(appPort) + .withAppChannelAddress("host.testcontainers.internal") + .withComponent(redisStateStore(STATE_STORE_NAME))); + httpDapr = httpPair.dapr(); + httpActorClient = newActorClient(httpDapr); + waitForActorsReady(httpDapr); + + var grpcPair = startActorHostAndAttach( + "actor-reentrancy-grpc-it", + MyActorGrpcService.class, + () -> daprBuilder("actor-reentrancy-grpc-it") + .withAppProtocol(DaprProtocol.GRPC) + .withComponent(redisStateStore(STATE_STORE_NAME))); + grpcDapr = grpcPair.dapr(); + grpcActorClient = newActorClient(grpcDapr); + waitForActorsReady(grpcDapr); + } + + @Test + public void reentersActorOverHttpHost() { + assertReentrantCallCompletes(httpActorClient); + } + + @Test + public void reentersActorOverGrpcHost() { + assertReentrantCallCompletes(grpcActorClient); + } + + private static void assertReentrantCallCompletes(ActorClient actorClient) { + ActorProxyBuilder builder = + new ActorProxyBuilder<>(ReentrantActorImpl.TYPE, ReentrantActor.class, actorClient); + String actorAId = UUID.randomUUID().toString(); + String actorBId = UUID.randomUUID().toString(); + ReentrantActor actorA = builder.build(new ActorId(actorAId)); + + assertEquals("reentrancy-ok", actorA.call(actorBId).block()); + } +} diff --git a/sdk-tests/src/test/java/io/dapr/it/actors/app/MyActorGrpcService.java b/sdk-tests/src/test/java/io/dapr/it/actors/app/MyActorGrpcService.java index cd17f07750..a097f96a49 100644 --- a/sdk-tests/src/test/java/io/dapr/it/actors/app/MyActorGrpcService.java +++ b/sdk-tests/src/test/java/io/dapr/it/actors/app/MyActorGrpcService.java @@ -15,6 +15,8 @@ import io.dapr.actors.runtime.ActorEventSubscription; import io.dapr.actors.runtime.ActorRuntime; +import io.dapr.actors.runtime.ActorReentrancyConfig; +import io.dapr.actors.runtime.ActorTypeConfig; import java.time.Duration; @@ -35,6 +37,10 @@ public static void main(String[] args) throws Exception { ActorRuntime.getInstance().registerActor(MyActorStringImpl.class); ActorRuntime.getInstance().registerActor(MyActorBinaryImpl.class); ActorRuntime.getInstance().registerActor(MyActorObjectImpl.class); + ActorRuntime.getInstance().getConfig().addActorTypeConfig(new ActorTypeConfig() + .setActorTypeName(ReentrantActorImpl.TYPE) + .setReentrancy(new ActorReentrancyConfig().setEnabled(true).setMaxStackDepth(5))); + ActorRuntime.getInstance().registerActor(ReentrantActorImpl.class); try (ActorEventSubscription subscription = ActorRuntime.getInstance().subscribeActorEvents()) { subscription.awaitTermination(); diff --git a/sdk-tests/src/test/java/io/dapr/it/actors/app/MyActorService.java b/sdk-tests/src/test/java/io/dapr/it/actors/app/MyActorService.java index 94f9959822..b1ed79f4ee 100644 --- a/sdk-tests/src/test/java/io/dapr/it/actors/app/MyActorService.java +++ b/sdk-tests/src/test/java/io/dapr/it/actors/app/MyActorService.java @@ -14,6 +14,8 @@ package io.dapr.it.actors.app; import io.dapr.actors.runtime.ActorRuntime; +import io.dapr.actors.runtime.ActorReentrancyConfig; +import io.dapr.actors.runtime.ActorTypeConfig; public class MyActorService { public static final String SUCCESS_MESSAGE = "dapr initialized. Status: Running"; @@ -30,6 +32,10 @@ public static void main(String[] args) throws Exception { ActorRuntime.getInstance().registerActor(MyActorStringImpl.class); ActorRuntime.getInstance().registerActor(MyActorBinaryImpl.class); ActorRuntime.getInstance().registerActor(MyActorObjectImpl.class); + ActorRuntime.getInstance().getConfig().addActorTypeConfig(new ActorTypeConfig() + .setActorTypeName(ReentrantActorImpl.TYPE) + .setReentrancy(new ActorReentrancyConfig().setEnabled(true).setMaxStackDepth(5))); + ActorRuntime.getInstance().registerActor(ReentrantActorImpl.class); TestApplication.start(port); } diff --git a/sdk-tests/src/test/java/io/dapr/it/actors/app/ReentrantActor.java b/sdk-tests/src/test/java/io/dapr/it/actors/app/ReentrantActor.java new file mode 100644 index 0000000000..2db56d9e6e --- /dev/null +++ b/sdk-tests/src/test/java/io/dapr/it/actors/app/ReentrantActor.java @@ -0,0 +1,26 @@ +/* + * Copyright 2026 The Dapr Authors + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * http://www.apache.org/licenses/LICENSE-2.0 + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. +*/ + +package io.dapr.it.actors.app; + +import reactor.core.publisher.Mono; + +/** Actor API used by the actor reentrancy integration test. */ +public interface ReentrantActor { + + Mono call(String targetActorId); + + Mono returnTo(String actorId); + + Mono finish(); +} diff --git a/sdk-tests/src/test/java/io/dapr/it/actors/app/ReentrantActorImpl.java b/sdk-tests/src/test/java/io/dapr/it/actors/app/ReentrantActorImpl.java new file mode 100644 index 0000000000..f5e2b006d8 --- /dev/null +++ b/sdk-tests/src/test/java/io/dapr/it/actors/app/ReentrantActorImpl.java @@ -0,0 +1,57 @@ +/* + * Copyright 2026 The Dapr Authors + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * http://www.apache.org/licenses/LICENSE-2.0 + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. +*/ + +package io.dapr.it.actors.app; + +import io.dapr.actors.ActorId; +import io.dapr.actors.ActorType; +import io.dapr.actors.client.ActorClient; +import io.dapr.actors.client.ActorProxyBuilder; +import io.dapr.actors.runtime.AbstractActor; +import io.dapr.actors.runtime.ActorRuntimeContext; +import reactor.core.publisher.Mono; + +/** Actor whose call method performs an A -> B -> A actor call chain. */ +@ActorType(name = ReentrantActorImpl.TYPE) +public class ReentrantActorImpl extends AbstractActor implements ReentrantActor { + + public static final String TYPE = "ReentrantActorTest"; + + private static final ActorClient ACTOR_CLIENT = new ActorClient(); + + private static final ActorProxyBuilder PROXY_BUILDER = + new ActorProxyBuilder<>(TYPE, ReentrantActor.class, ACTOR_CLIENT); + + public ReentrantActorImpl(ActorRuntimeContext runtimeContext, ActorId id) { + super(runtimeContext, id); + } + + @Override + public Mono call(String targetActorId) { + return proxy(targetActorId).returnTo(getId().toString()); + } + + @Override + public Mono returnTo(String actorId) { + return proxy(actorId).finish(); + } + + @Override + public Mono finish() { + return Mono.just("reentrancy-ok"); + } + + private static ReentrantActor proxy(String actorId) { + return PROXY_BUILDER.build(new ActorId(actorId)); + } +} From 4a143bdb91cdd3a3a59187a00e628a3e2af767cd Mon Sep 17 00:00:00 2001 From: Alpha <2748109647@qq.com> Date: Sun, 25 Oct 2026 20:21:07 +0800 Subject: [PATCH 2/3] fix: preserve actor reentrancy for blocking calls Synchronous actor methods can call another actor through a proxy that blocks on its own subscription. Reactor context is not propagated through that path, so the reentrancy id must be scoped around the actor method invocation and made available to the outgoing Dapr client call. Add a ThreadLocal fallback with nested-scope restoration and cover sync A -> B -> A calls for both HTTP and gRPC actor hosts. Signed-off-by: Alpha <2748109647@qq.com> --- .../internal/ActorReentrancyContext.java | 33 ++++++++++- .../io/dapr/actors/runtime/ActorManager.java | 53 +++++++++-------- .../internal/ActorReentrancyContextTest.java | 58 +++++++++++++++++++ .../io/dapr/it/actors/ActorReentrancyIT.java | 20 +++++++ .../io/dapr/it/actors/app/ReentrantActor.java | 6 ++ .../it/actors/app/ReentrantActorImpl.java | 15 +++++ 6 files changed, 160 insertions(+), 25 deletions(-) create mode 100644 sdk-actors/src/test/java/io/dapr/actors/internal/ActorReentrancyContextTest.java diff --git a/sdk-actors/src/main/java/io/dapr/actors/internal/ActorReentrancyContext.java b/sdk-actors/src/main/java/io/dapr/actors/internal/ActorReentrancyContext.java index dc9fee6688..f407e18b9c 100644 --- a/sdk-actors/src/main/java/io/dapr/actors/internal/ActorReentrancyContext.java +++ b/sdk-actors/src/main/java/io/dapr/actors/internal/ActorReentrancyContext.java @@ -17,6 +17,7 @@ import reactor.util.context.ContextView; import java.util.Map; +import java.util.function.Supplier; /** * Internal helpers for carrying an incoming actor reentrancy id through a Reactor invocation. @@ -28,6 +29,8 @@ public final class ActorReentrancyContext { private static final String CONTEXT_KEY = ActorReentrancyContext.class.getName() + ".id"; + private static final ThreadLocal THREAD_LOCAL_REENTRANCY_ID = new ThreadLocal<>(); + private ActorReentrancyContext() { } @@ -46,6 +49,33 @@ public static Mono withReentrancyId(Mono publisher, String reentrancyI return publisher.contextWrite(context -> context.put(CONTEXT_KEY, reentrancyId)); } + /** + * Sets the reentrancy id for the duration of a synchronous actor method invocation. + * + * @param reentrancyId Id received from Dapr, or {@code null} when this is not a reentrant call. + * @param action Synchronous actor method invocation. + * @param Invocation result type. + * @return The invocation result. + */ + public static T withReentrancyId(String reentrancyId, Supplier action) { + String previous = THREAD_LOCAL_REENTRANCY_ID.get(); + if (reentrancyId == null || reentrancyId.isEmpty()) { + THREAD_LOCAL_REENTRANCY_ID.remove(); + } else { + THREAD_LOCAL_REENTRANCY_ID.set(reentrancyId); + } + + try { + return action.get(); + } finally { + if (previous == null) { + THREAD_LOCAL_REENTRANCY_ID.remove(); + } else { + THREAD_LOCAL_REENTRANCY_ID.set(previous); + } + } + } + /** * Reads the reentrancy id from a Reactor subscriber context. * @@ -53,7 +83,7 @@ public static Mono withReentrancyId(Mono publisher, String reentrancyI * @return The id, or {@code null} when no actor invocation is active. */ public static String getReentrancyId(ContextView context) { - return context.getOrDefault(CONTEXT_KEY, null); + return context.hasKey(CONTEXT_KEY) ? context.get(CONTEXT_KEY) : THREAD_LOCAL_REENTRANCY_ID.get(); } /** @@ -72,4 +102,5 @@ public static String getReentrancyId(Map metadata) { .findFirst() .orElse(null); } + } diff --git a/sdk-actors/src/main/java/io/dapr/actors/runtime/ActorManager.java b/sdk-actors/src/main/java/io/dapr/actors/runtime/ActorManager.java index 303736cf57..71f9f06c22 100644 --- a/sdk-actors/src/main/java/io/dapr/actors/runtime/ActorManager.java +++ b/sdk-actors/src/main/java/io/dapr/actors/runtime/ActorManager.java @@ -14,6 +14,7 @@ package io.dapr.actors.runtime; import io.dapr.actors.ActorId; +import io.dapr.actors.internal.ActorReentrancyContext; import io.dapr.serializer.DaprObjectSerializer; import io.dapr.utils.TypeRef; import reactor.core.publisher.Mono; @@ -260,20 +261,21 @@ private Mono invokeMethod(ActorId actorId, ActorMethodContext context, S * @return Asynchronous object response. */ private Mono invokeMethod(AbstractActor actor, Method method, Object input) { - return Mono.fromSupplier(() -> { - try { - if (method.getParameterCount() == 0) { - return method.invoke(actor); - } else { - // Actor methods must have a one or no parameter, which is guaranteed at this point. - return method.invoke(actor, input); - } - } catch (RuntimeException e) { - throw e; - } catch (Exception e) { - throw new RuntimeException(e); - } - }); + return Mono.deferContextual(context -> Mono.fromSupplier(() -> + ActorReentrancyContext.withReentrancyId(ActorReentrancyContext.getReentrancyId(context), () -> { + try { + if (method.getParameterCount() == 0) { + return method.invoke(actor); + } else { + // Actor methods must have a one or no parameter, which is guaranteed at this point. + return method.invoke(actor, input); + } + } catch (RuntimeException e) { + throw e; + } catch (Exception e) { + throw new RuntimeException(e); + } + }))); } /** @@ -285,16 +287,19 @@ private Mono invokeMethod(AbstractActor actor, Method method, Object inp * @return Asynchronous object response. */ private Mono invokeMonoMethod(AbstractActor actor, Method method, Object input) { - try { - if (method.getParameterCount() == 0) { - return (Mono) method.invoke(actor); - } else { - // Actor methods must have a one or no parameter, which is guaranteed at this point. - return (Mono) method.invoke(actor, input); - } - } catch (Exception e) { - return Mono.error(e); - } + return Mono.deferContextual(context -> ActorReentrancyContext.withReentrancyId( + ActorReentrancyContext.getReentrancyId(context), () -> { + try { + if (method.getParameterCount() == 0) { + return (Mono) method.invoke(actor); + } else { + // Actor methods must have a one or no parameter, which is guaranteed at this point. + return (Mono) method.invoke(actor, input); + } + } catch (Exception e) { + return Mono.error(e); + } + })); } /** diff --git a/sdk-actors/src/test/java/io/dapr/actors/internal/ActorReentrancyContextTest.java b/sdk-actors/src/test/java/io/dapr/actors/internal/ActorReentrancyContextTest.java new file mode 100644 index 0000000000..e3dd3c09dd --- /dev/null +++ b/sdk-actors/src/test/java/io/dapr/actors/internal/ActorReentrancyContextTest.java @@ -0,0 +1,58 @@ +/* + * Copyright 2026 The Dapr Authors + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * http://www.apache.org/licenses/LICENSE-2.0 + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package io.dapr.actors.internal; + +import org.junit.jupiter.api.Test; +import reactor.core.publisher.Mono; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertNull; + +class ActorReentrancyContextTest { + + @Test + void synchronousScopesAreVisibleToBlockingCallsAndRestored() { + String result = ActorReentrancyContext.withReentrancyId("outer", () -> { + assertEquals("outer", Mono.deferContextual(context -> + Mono.justOrEmpty(ActorReentrancyContext.getReentrancyId(context))).block()); + + String nestedResult = ActorReentrancyContext.withReentrancyId("inner", () -> + Mono.deferContextual(context -> Mono.justOrEmpty( + ActorReentrancyContext.getReentrancyId(context))).block()); + assertEquals("inner", nestedResult); + + return Mono.deferContextual(context -> Mono.justOrEmpty( + ActorReentrancyContext.getReentrancyId(context))).block(); + }); + + assertEquals("outer", result); + assertNull(Mono.deferContextual(context -> Mono.justOrEmpty( + ActorReentrancyContext.getReentrancyId(context))).block()); + } + + @Test + void nestedNonReentrantCallDoesNotInheritTheOuterId() { + String result = ActorReentrancyContext.withReentrancyId("outer", () -> { + String nestedResult = ActorReentrancyContext.withReentrancyId(null, () -> + Mono.deferContextual(context -> Mono.justOrEmpty( + ActorReentrancyContext.getReentrancyId(context))).block()); + assertNull(nestedResult); + return Mono.deferContextual(context -> Mono.justOrEmpty( + ActorReentrancyContext.getReentrancyId(context))).block(); + }); + + assertEquals("outer", result); + } + +} diff --git a/sdk-tests/src/test/java/io/dapr/it/actors/ActorReentrancyIT.java b/sdk-tests/src/test/java/io/dapr/it/actors/ActorReentrancyIT.java index f9ffd7e6d7..ba8ec3bfc1 100644 --- a/sdk-tests/src/test/java/io/dapr/it/actors/ActorReentrancyIT.java +++ b/sdk-tests/src/test/java/io/dapr/it/actors/ActorReentrancyIT.java @@ -72,11 +72,21 @@ public void reentersActorOverHttpHost() { assertReentrantCallCompletes(httpActorClient); } + @Test + public void reentersActorSynchronouslyOverHttpHost() { + assertSynchronousReentrantCallCompletes(httpActorClient); + } + @Test public void reentersActorOverGrpcHost() { assertReentrantCallCompletes(grpcActorClient); } + @Test + public void reentersActorSynchronouslyOverGrpcHost() { + assertSynchronousReentrantCallCompletes(grpcActorClient); + } + private static void assertReentrantCallCompletes(ActorClient actorClient) { ActorProxyBuilder builder = new ActorProxyBuilder<>(ReentrantActorImpl.TYPE, ReentrantActor.class, actorClient); @@ -86,4 +96,14 @@ private static void assertReentrantCallCompletes(ActorClient actorClient) { assertEquals("reentrancy-ok", actorA.call(actorBId).block()); } + + private static void assertSynchronousReentrantCallCompletes(ActorClient actorClient) { + ActorProxyBuilder builder = + new ActorProxyBuilder<>(ReentrantActorImpl.TYPE, ReentrantActor.class, actorClient); + String actorAId = UUID.randomUUID().toString(); + String actorBId = UUID.randomUUID().toString(); + ReentrantActor actorA = builder.build(new ActorId(actorAId)); + + assertEquals("sync-reentrancy-ok", actorA.callBlocking(actorBId)); + } } diff --git a/sdk-tests/src/test/java/io/dapr/it/actors/app/ReentrantActor.java b/sdk-tests/src/test/java/io/dapr/it/actors/app/ReentrantActor.java index 2db56d9e6e..ba389a9b76 100644 --- a/sdk-tests/src/test/java/io/dapr/it/actors/app/ReentrantActor.java +++ b/sdk-tests/src/test/java/io/dapr/it/actors/app/ReentrantActor.java @@ -23,4 +23,10 @@ public interface ReentrantActor { Mono returnTo(String actorId); Mono finish(); + + String callBlocking(String targetActorId); + + String returnToBlocking(String actorId); + + String finishBlocking(); } diff --git a/sdk-tests/src/test/java/io/dapr/it/actors/app/ReentrantActorImpl.java b/sdk-tests/src/test/java/io/dapr/it/actors/app/ReentrantActorImpl.java index f5e2b006d8..b451d9abbd 100644 --- a/sdk-tests/src/test/java/io/dapr/it/actors/app/ReentrantActorImpl.java +++ b/sdk-tests/src/test/java/io/dapr/it/actors/app/ReentrantActorImpl.java @@ -51,6 +51,21 @@ public Mono finish() { return Mono.just("reentrancy-ok"); } + @Override + public String callBlocking(String targetActorId) { + return proxy(targetActorId).returnToBlocking(getId().toString()); + } + + @Override + public String returnToBlocking(String actorId) { + return proxy(actorId).finishBlocking(); + } + + @Override + public String finishBlocking() { + return "sync-reentrancy-ok"; + } + private static ReentrantActor proxy(String actorId) { return PROXY_BUILDER.build(new ActorId(actorId)); } From 4bc1f03a1c2a7b72085cda3a5d73e5637f86d23e Mon Sep 17 00:00:00 2001 From: Alpha <2748109647@qq.com> Date: Mon, 26 Oct 2026 10:21:58 +0800 Subject: [PATCH 3/3] docs: clarify sync actor reentrancy Signed-off-by: Alpha <2748109647@qq.com> --- examples/src/main/java/io/dapr/examples/actors/README.md | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/examples/src/main/java/io/dapr/examples/actors/README.md b/examples/src/main/java/io/dapr/examples/actors/README.md index 5c5edf0775..f56258b108 100644 --- a/examples/src/main/java/io/dapr/examples/actors/README.md +++ b/examples/src/main/java/io/dapr/examples/actors/README.md @@ -79,7 +79,7 @@ runtime.getConfig().addActorTypeConfig(new ActorTypeConfig() .setReentrancy(new ActorReentrancyConfig().setEnabled(false))); ``` -Actor methods that make nested actor calls should compose the returned `Mono` so the invocation context can carry Dapr's reentrancy id across scheduler changes and concurrent calls. +For reactive actor methods, compose nested calls through the returned `Mono` so Reactor context can carry Dapr's reentrancy id across scheduler changes and concurrent calls. Synchronous actor methods can make nested calls through a blocking actor proxy; the SDK scopes the reentrancy id to the method invocation. `DaprApplication.start()` method will run the Spring Boot [DaprApplication](https://github.com/dapr/java-sdk/blob/master/sdk-tests/src/test/java/io/dapr/it/actors/services/springboot/DaprApplication.java), which registers the Dapr Spring Boot controller [DaprController](https://github.com/dapr/java-sdk/blob/master/sdk-springboot/src/main/java/io/dapr/springboot/DaprController.java). This controller contains all Actor methods implemented as endpoints. The Dapr's sidecar will call into the controller.