From f96f7b252bf3abb84266ad2963470870d177024f Mon Sep 17 00:00:00 2001 From: Cassandra Coyle Date: Mon, 26 Oct 2026 14:44:44 -0500 Subject: [PATCH 1/4] fix(actors): Recreate actor object when Dapr reports a new activation Signed-off-by: Cassandra Coyle --- .../io/dapr/actors/runtime/AbstractActor.java | 6 +- .../io/dapr/actors/runtime/ActorManager.java | 180 +++-- .../io/dapr/actors/runtime/ActorRuntime.java | 57 +- .../dapr/actors/runtime/ActorManagerTest.java | 618 ++++++++++++++++++ .../dapr/actors/runtime/ActorRuntimeTest.java | 288 ++++++++ sdk-springboot/pom.xml | 10 + .../io/dapr/springboot/DaprController.java | 59 +- .../dapr/springboot/DaprControllerTest.java | 142 ++++ sdk/src/main/java/io/dapr/client/Headers.java | 3 + 9 files changed, 1309 insertions(+), 54 deletions(-) create mode 100644 sdk-springboot/src/test/java/io/dapr/springboot/DaprControllerTest.java diff --git a/sdk-actors/src/main/java/io/dapr/actors/runtime/AbstractActor.java b/sdk-actors/src/main/java/io/dapr/actors/runtime/AbstractActor.java index 11e1a941ec..11307356a3 100644 --- a/sdk-actors/src/main/java/io/dapr/actors/runtime/AbstractActor.java +++ b/sdk-actors/src/main/java/io/dapr/actors/runtime/AbstractActor.java @@ -21,6 +21,7 @@ import java.time.Duration; import java.util.UUID; import java.util.concurrent.atomic.AtomicBoolean; +import java.util.function.BooleanSupplier; /** * Represents the base class for actors. @@ -338,15 +339,16 @@ void resetState() { /** * Internal callback when an Actor is activated. * + * @param canSave Checked after onActivate, false skips the state save. * @return Asynchronous void response. */ - Mono onActivateInternal() { + Mono onActivateInternal(BooleanSupplier canSave) { return Mono.fromRunnable(() -> { this.actorTrace.writeInfo(TRACE_TYPE, this.id.toString(), "Activating ..."); this.resetState(); }).then(this.onActivate()) .then(this.doWriteInfo(TRACE_TYPE, this.id.toString(), "Activated")) - .then(this.saveState()); + .then(Mono.defer(() -> canSave.getAsBoolean() ? this.saveState() : Mono.empty())); } /** 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 c51e6e8bd4..3c072351d3 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 @@ -47,7 +47,7 @@ class ActorManager { /** * Active Actor instances. */ - private final Map activeActors; + private final Map> activeActors; /** * Instantiates a new manager for a given actor referenced in the runtimeContext. @@ -67,13 +67,26 @@ class ActorManager { * @return Asynchronous void response. */ Mono activateActor(ActorId actorId) { - return Mono.fromSupplier(() -> { - if (this.activeActors.containsKey(actorId)) { - return null; + return activateActor(actorId, null).then(); + } + + /** Replaces the live instance if Dapr reports a different activation. */ + Mono> activateActor(ActorId actorId, String activationId) { + return Mono.defer(() -> { + ActiveActor current = this.activeActors.get(actorId); + if (current == null) { + return this.createActor(actorId, activationId); } - return this.runtimeContext.getActorFactory().createActor(runtimeContext, actorId); - }).flatMap(actor -> actor.onActivateInternal().then(this.onActivatedActor(actorId, actor))); + if (activationId == null || activationId.isEmpty() || activationId.equals(current.activationId)) { + return Mono.just(current); + } + + Mono deactivate = this.activeActors.remove(actorId, current) + ? this.deactivateDiscarded(actorId, current.actor) + : Mono.empty(); + return deactivate.then(Mono.defer(() -> this.createActor(actorId, activationId))); + }); } /** @@ -83,7 +96,8 @@ Mono activateActor(ActorId actorId) { * @return Asynchronous void response. */ Mono deactivateActor(ActorId actorId) { - return Mono.fromSupplier(() -> this.activeActors.remove(actorId)).flatMap(actor -> actor.onDeactivateInternal()); + return Mono.fromSupplier(() -> this.activeActors.remove(actorId)) + .flatMap(active -> active.actor.onDeactivateInternal()); } /** @@ -95,6 +109,14 @@ Mono deactivateActor(ActorId actorId) { * @return Asynchronous void response. */ Mono invokeReminder(ActorId actorId, String reminderName, byte[] params) { + return this.invokeReminder(this.lookup(actorId), reminderName, params); + } + + Mono invokeReminder(ActiveActor active, String reminderName, byte[] params) { + return this.invokeReminder(Mono.just(active), reminderName, params); + } + + private Mono invokeReminder(Mono> active, String reminderName, byte[] params) { return Mono.fromSupplier(() -> { if (!this.runtimeContext.getActorTypeInformation().isRemindable()) { return null; @@ -105,10 +127,10 @@ Mono invokeReminder(ActorId actorId, String reminderName, byte[] params) { } catch (Exception e) { throw new RuntimeException(e); } - }).flatMap(p -> - invoke(actorId, + }).flatMap(p -> active.flatMap(a -> + invoke(a, ActorMethodContext.createForReminder(reminderName), - actor -> doReminderInvokation((Remindable) actor, reminderName, p))).then(); + actor -> doReminderInvokation((Remindable) actor, reminderName, p)))).then(); } /** @@ -120,29 +142,68 @@ Mono invokeReminder(ActorId actorId, String reminderName, byte[] params) { * @return Asynchronous void response. */ Mono invokeTimer(ActorId actorId, String timerName, byte[] params) { + return this.invokeTimer(this.lookup(actorId), timerName, params); + } + + Mono invokeTimer(ActiveActor active, String timerName, byte[] params) { + return this.invokeTimer(Mono.just(active), timerName, params); + } + + private Mono invokeTimer(Mono> active, String timerName, byte[] params) { return Mono.fromSupplier(() -> { try { return OBJECT_SERIALIZER.deserialize(params, ActorTimerParams.class); } catch (Exception e) { throw new RuntimeException(e); } - }).flatMap(p -> + }).flatMap(p -> active.flatMap(a -> invokeMethod( - actorId, + a, ActorMethodContext.createForTimer(timerName), p.getCallback(), - p.getData())).then(); + p.getData()))).then(); } - /** - * Internal callback for when Actor is activated. - * - * @param actorId Actor identifier. - * @param actor Actor's instance. - * @return Asynchronous void response. - */ - private Mono onActivatedActor(ActorId actorId, T actor) { - return Mono.fromRunnable(() -> this.activeActors.put(actorId, actor)); + private Mono> createActor(ActorId actorId, String activationId) { + return Mono.fromSupplier(() -> this.runtimeContext.getActorFactory().createActor(runtimeContext, actorId)) + .flatMap(actor -> actor.onActivateInternal(() -> !this.activeActors.containsKey(actorId)) + .then(this.onActivatedActor(actorId, actor, activationId))); + } + + /** Discards the new instance if a concurrent activation registered first. */ + private Mono> onActivatedActor(ActorId actorId, T actor, String activationId) { + return Mono.defer(() -> { + ActiveActor active = new ActiveActor<>(actor, activationId); + ActiveActor registered = this.activeActors.putIfAbsent(actorId, active); + if (registered == null) { + return Mono.just(active); + } + + Mono discard = this.deactivateDiscarded(actorId, actor); + if (activationId == null || activationId.isEmpty() || activationId.equals(registered.activationId)) { + return discard.thenReturn(registered); + } + + return discard.then(Mono.error(new IllegalStateException( + String.format("Actor %s of type %s was activated by activation %s during this activation.", + actorId.toString(), + this.runtimeContext.getActorTypeInformation().getName(), + registered.activationId)))); + }); + } + + /** Logs errors so a failed cleanup doesn't fail the call. */ + private Mono deactivateDiscarded(ActorId actorId, T actor) { + return Mono.defer(actor::onDeactivateInternal) + .onErrorResume(e -> { + this.runtimeContext.getActorTrace().writeError( + this.getClass().getSimpleName(), + actorId.toString(), + "Deactivating discarded actor of type %s failed: %s", + this.runtimeContext.getActorTypeInformation().getName(), + e.toString()); + return Mono.empty(); + }); } /** @@ -194,25 +255,30 @@ private Mono doReminderInvokation( * @return Asynchronous void response. */ Mono invokeMethod(ActorId actorId, String methodName, byte[] request) { - return invokeMethod(actorId, null, methodName, request); + return this.lookup(actorId).flatMap(active -> this.invokeMethod(active, methodName, request)); + } + + Mono invokeMethod(ActiveActor active, String methodName, byte[] request) { + return invokeMethod(active, null, methodName, request); } /** * Internal method to actually invoke Actor's method. * - * @param actorId Identifier for the Actor. + * @param active Actor instance to invoke. * @param context Method context to be invoked. * @param methodName Method name to be invoked. * @param request Input object to be passed in to the invoked method. * @return Asynchronous serialized response. */ - private Mono invokeMethod(ActorId actorId, ActorMethodContext context, String methodName, byte[] request) { + private Mono invokeMethod( + ActiveActor active, ActorMethodContext context, String methodName, byte[] request) { ActorMethodContext actorMethodContext = context; if (actorMethodContext == null) { actorMethodContext = ActorMethodContext.createForActor(methodName); } - return this.invoke(actorId, actorMethodContext, actor -> { + return this.invoke(active, actorMethodContext, actor -> { try { // Finds the actor method with the given name and 1 or no parameter. Method method = this.actorMethods.get(methodName); @@ -290,22 +356,15 @@ private Mono invokeMonoMethod(AbstractActor actor, Method method, Object /** * Internal call to invoke a method, timer or reminder for an Actor. * - * @param actorId Actor identifier. + * @param active Actor instance to invoke. * @param context Context for the method/timer/reminder call. * @param func Function to perform the method call. - * @param Expected return type for the function call. + * @param Expected return type for the function call. * @return Asynchronous response for the returned object. */ - private Mono invoke(ActorId actorId, ActorMethodContext context, Function> func) { + private Mono invoke(ActiveActor active, ActorMethodContext context, Function> func) { try { - AbstractActor actor = this.activeActors.getOrDefault(actorId, null); - if (actor == null) { - throw new IllegalArgumentException( - String.format("Could not find actor %s of type %s.", - actorId.toString(), - this.runtimeContext.getActorTypeInformation().getName())); - } - + AbstractActor actor = active.actor; return Mono.fromRunnable(() -> actor.rollback(true)) .onErrorMap(throwable -> { actor.rollback(false); @@ -314,11 +373,52 @@ private Mono invoke(ActorId actorId, ActorMethodContext context, Function .then(actor.onPreActorMethodInternal(context)) .then((Mono) func.apply(actor)) .switchIfEmpty( - actor.onPostActorMethodInternal(context)) - .flatMap(r -> actor.onPostActorMethodInternal(context).thenReturn(r)) - .map(o -> (T) o); + this.ifLive(active, actor.onPostActorMethodInternal(context))) + .flatMap(r -> this.ifLive(active, actor.onPostActorMethodInternal(context)).thenReturn(r)) + .map(o -> (R) o); } catch (Exception e) { return Mono.error(e); } } + + private Mono> lookup(ActorId actorId) { + return Mono.defer(() -> { + ActiveActor active = this.activeActors.get(actorId); + if (active == null) { + return Mono.error(new IllegalArgumentException( + String.format("Could not find actor %s of type %s.", + actorId.toString(), + this.runtimeContext.getActorTypeInformation().getName()))); + } + + return Mono.just(active); + }); + } + + /** Skips the post step state save if the instance was replaced during the call. */ + private Mono ifLive(ActiveActor active, Mono post) { + return Mono.defer(() -> { + ActorId actorId = active.actor.getId(); + if (this.activeActors.get(actorId) != active) { + return Mono.error(new IllegalStateException( + String.format("Actor %s of type %s was replaced during the call, state not saved.", + actorId.toString(), + this.runtimeContext.getActorTypeInformation().getName()))); + } + + return post; + }); + } + + static final class ActiveActor { + + private final A actor; + + private final String activationId; + + private ActiveActor(A actor, String activationId) { + this.actor = actor; + this.activationId = activationId; + } + } } diff --git a/sdk-actors/src/main/java/io/dapr/actors/runtime/ActorRuntime.java b/sdk-actors/src/main/java/io/dapr/actors/runtime/ActorRuntime.java index d329946e5a..da766cfcbf 100644 --- a/sdk-actors/src/main/java/io/dapr/actors/runtime/ActorRuntime.java +++ b/sdk-actors/src/main/java/io/dapr/actors/runtime/ActorRuntime.java @@ -274,10 +274,25 @@ public Mono deactivate(String actorTypeName, String actorId) { * @return Response for the actor method. */ public Mono invoke(String actorTypeName, String actorId, String actorMethodName, byte[] payload) { + return invoke(actorTypeName, actorId, actorMethodName, payload, null); + } + + /** + * Invokes the method on the activation Dapr sent. + * + * @param actorTypeName Actor type name. + * @param actorId Actor id. + * @param actorMethodName Method name. + * @param payload RAW payload. + * @param activationId Activation id sent by Dapr, or null if absent. + * @return Response for the actor method. + */ + public Mono invoke( + String actorTypeName, String actorId, String actorMethodName, byte[] payload, String activationId) { ActorId id = new ActorId(actorId); return Mono.fromSupplier(() -> this.getActorManager(actorTypeName)) - .flatMap(m -> m.activateActor(id).thenReturn(m)) - .flatMap(m -> ((ActorManager) m).invokeMethod(id, actorMethodName, payload)); + .flatMap(m -> m.activateActor(id, activationId) + .flatMap(a -> m.invokeMethod((ActorManager.ActiveActor) a, actorMethodName, payload))); } /** @@ -290,10 +305,25 @@ public Mono invoke(String actorTypeName, String actorId, String actorMet * @return Async void task. */ public Mono invokeReminder(String actorTypeName, String actorId, String reminderName, byte[] params) { + return invokeReminder(actorTypeName, actorId, reminderName, params, null); + } + + /** + * Fires a reminder on the activation Dapr sent. + * + * @param actorTypeName Actor type name. + * @param actorId Actor id. + * @param reminderName Reminder name. + * @param params Reminder params. + * @param activationId Activation id sent by Dapr, or null if absent. + * @return Async void task. + */ + public Mono invokeReminder( + String actorTypeName, String actorId, String reminderName, byte[] params, String activationId) { ActorId id = new ActorId(actorId); return Mono.fromSupplier(() -> this.getActorManager(actorTypeName)) - .flatMap(m -> m.activateActor(id).thenReturn(m)) - .flatMap(m -> ((ActorManager) m).invokeReminder(new ActorId(actorId), reminderName, params)); + .flatMap(m -> m.activateActor(id, activationId) + .flatMap(a -> m.invokeReminder((ActorManager.ActiveActor) a, reminderName, params))); } /** @@ -306,10 +336,25 @@ public Mono invokeReminder(String actorTypeName, String actorId, String re * @return Async void task. */ public Mono invokeTimer(String actorTypeName, String actorId, String timerName, byte[] params) { + return invokeTimer(actorTypeName, actorId, timerName, params, null); + } + + /** + * Fires a timer on the activation Dapr sent. + * + * @param actorTypeName Actor type name. + * @param actorId Actor id. + * @param timerName Timer name. + * @param params Timer params. + * @param activationId Activation id sent by Dapr, or null if absent. + * @return Async void task. + */ + public Mono invokeTimer( + String actorTypeName, String actorId, String timerName, byte[] params, String activationId) { ActorId id = new ActorId(actorId); return Mono.fromSupplier(() -> this.getActorManager(actorTypeName)) - .flatMap(m -> m.activateActor(id).thenReturn(m)) - .flatMap(m -> ((ActorManager) m).invokeTimer(new ActorId(actorId), timerName, params)); + .flatMap(m -> m.activateActor(id, activationId) + .flatMap(a -> m.invokeTimer((ActorManager.ActiveActor) a, timerName, params))); } /** diff --git a/sdk-actors/src/test/java/io/dapr/actors/runtime/ActorManagerTest.java b/sdk-actors/src/test/java/io/dapr/actors/runtime/ActorManagerTest.java index 6fd1069436..7175ef77e2 100644 --- a/sdk-actors/src/test/java/io/dapr/actors/runtime/ActorManagerTest.java +++ b/sdk-actors/src/test/java/io/dapr/actors/runtime/ActorManagerTest.java @@ -22,7 +22,15 @@ import reactor.core.publisher.Mono; import java.io.IOException; +import java.lang.reflect.Field; import java.time.Duration; +import java.util.ArrayList; +import java.util.Collections; +import java.util.HashMap; +import java.util.List; +import java.util.concurrent.CompletableFuture; +import java.util.concurrent.ExecutionException; +import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicInteger; import static org.junit.jupiter.api.Assertions.assertThrows; @@ -116,6 +124,170 @@ public Mono receiveReminder(String reminderName, String state, Duration du } } + interface ActivationActor { + int increment(); + } + + @ActorType(name = "ActivationActor") + public static class ActivationActorImpl extends AbstractActor implements ActivationActor { + + private int calls; + + private int activations; + + private int deactivations; + + public ActivationActorImpl(ActorRuntimeContext runtimeContext, ActorId id) { + super(runtimeContext, id); + } + + @Override + protected Mono onActivate() { + return Mono.fromRunnable(() -> this.activations++); + } + + @Override + protected Mono onDeactivate() { + return Mono.fromRunnable(() -> this.deactivations++); + } + + @Override + public int increment() { + return ++this.calls; + } + } + + interface GatedActor { + Mono write(String value); + + int increment(); + } + + @ActorType(name = "GatedActor") + public static class GatedActorImpl extends AbstractActor implements GatedActor, Remindable { + + private final Mono gate; + + private int calls; + + private int deactivations; + + public GatedActorImpl(ActorRuntimeContext runtimeContext, ActorId id, Mono gate) { + super(runtimeContext, id); + this.gate = gate; + } + + @Override + protected Mono onDeactivate() { + return Mono.fromRunnable(() -> this.deactivations++); + } + + @Override + public Mono write(String value) { + return this.gate.then(Mono.defer(() -> this.getActorStateManager().set("value", value))); + } + + @Override + public int increment() { + return ++this.calls; + } + + @Override + public TypeRef getStateType() { + return TypeRef.STRING; + } + + @Override + public Mono receiveReminder(String reminderName, String state, Duration dueTime, Duration period) { + return this.write(state); + } + } + + @ActorType(name = "ActivateWriteActor") + public static class ActivateWriteActorImpl extends AbstractActor implements GatedActor { + + private final Mono activateGate; + + private final Mono callGate; + + private final String tag; + + private final boolean failDeactivate; + + private int calls; + + private int deactivations; + + public ActivateWriteActorImpl(ActorRuntimeContext runtimeContext, ActorId id, Mono activateGate, + Mono callGate, String tag, boolean failDeactivate) { + super(runtimeContext, id); + this.activateGate = activateGate; + this.callGate = callGate; + this.tag = tag; + this.failDeactivate = failDeactivate; + } + + @Override + protected Mono onActivate() { + return this.activateGate.then(Mono.defer(() -> this.getActorStateManager().set("value", this.tag))); + } + + @Override + protected Mono onDeactivate() { + this.deactivations++; + return this.failDeactivate ? Mono.error(new IllegalArgumentException("deactivate failed")) : Mono.empty(); + } + + @Override + public Mono write(String value) { + return this.callGate.then(Mono.defer(() -> this.getActorStateManager().set("value", value))); + } + + @Override + public int increment() { + return ++this.calls; + } + } + + static class RecordingStateProvider extends DaprStateAsyncProvider { + + private final List saved = Collections.synchronizedList(new ArrayList<>()); + + RecordingStateProvider() { + super(mock(DaprClient.class), new DefaultObjectSerializer()); + } + + @Override + Mono contains(String actorType, ActorId actorId, String stateName) { + return Mono.just(false); + } + + @Override + Mono apply(String actorType, ActorId actorId, ActorStateChange... stateChanges) { + return Mono.fromRunnable(() -> { + for (ActorStateChange change : stateChanges) { + this.saved.add(change.getStateName() + "=" + change.getValue()); + } + }); + } + } + + static class GetHookMap extends HashMap { + + private Runnable onGet; + + @Override + public V get(Object key) { + V value = super.get(key); + Runnable hook = this.onGet; + this.onGet = null; + if (hook != null) { + hook.run(); + } + return value; + } + } + private ActorRuntimeContext context = createContext(MyActorImpl.class); private ActorManager manager = new ActorManager<>(context); @@ -279,6 +451,452 @@ public void activateInvokeTimerDeactivateThenInvokeTimer() throws IOException { assertThrows(IllegalArgumentException.class, () -> this.manager.invokeTimer(actorId, "count", createTimerParams("incrementCount", 2)).block()); } + @Test + public void sameActivationIdReusesActor() throws Exception { + List created = new ArrayList<>(); + ActorManager manager = createActivationManager(created); + ActorId actorId = newActorId(); + + manager.activateActor(actorId, "a1").block(); + Assertions.assertEquals(1, increment(manager, actorId)); + manager.activateActor(actorId, "a1").block(); + Assertions.assertEquals(2, increment(manager, actorId)); + + Assertions.assertEquals(1, created.size()); + Assertions.assertEquals(1, created.get(0).activations); + Assertions.assertEquals(0, created.get(0).deactivations); + } + + @Test + public void newActivationIdReplacesActor() throws Exception { + List created = new ArrayList<>(); + ActorManager manager = createActivationManager(created); + ActorId actorId = newActorId(); + + manager.activateActor(actorId, "a1").block(); + Assertions.assertEquals(1, increment(manager, actorId)); + Assertions.assertEquals(2, increment(manager, actorId)); + manager.activateActor(actorId, "a2").block(); + + Assertions.assertEquals(2, created.size()); + Assertions.assertEquals(1, created.get(0).deactivations); + Assertions.assertEquals(1, created.get(1).activations); + Assertions.assertEquals(0, created.get(1).deactivations); + Assertions.assertEquals(1, increment(manager, actorId)); + Assertions.assertEquals(2, created.get(0).calls); + } + + @Test + public void missingActivationIdNeverResets() throws Exception { + List created = new ArrayList<>(); + ActorManager manager = createActivationManager(created); + ActorId actorId = newActorId(); + + manager.activateActor(actorId, "a1").block(); + Assertions.assertEquals(1, increment(manager, actorId)); + manager.activateActor(actorId, "").block(); + Assertions.assertEquals(2, increment(manager, actorId)); + manager.activateActor(actorId, null).block(); + Assertions.assertEquals(3, increment(manager, actorId)); + manager.activateActor(actorId).block(); + Assertions.assertEquals(4, increment(manager, actorId)); + + Assertions.assertEquals(1, created.size()); + Assertions.assertEquals(0, created.get(0).deactivations); + } + + @Test + public void activationIdAfterMissingActivationIdResets() throws Exception { + List created = new ArrayList<>(); + ActorManager manager = createActivationManager(created); + ActorId actorId = newActorId(); + + manager.activateActor(actorId).block(); + Assertions.assertEquals(1, increment(manager, actorId)); + manager.activateActor(actorId, "a1").block(); + Assertions.assertEquals(2, created.size()); + Assertions.assertEquals(1, created.get(0).deactivations); + Assertions.assertEquals(1, increment(manager, actorId)); + + manager.activateActor(actorId, "a1").block(); + Assertions.assertEquals(2, increment(manager, actorId)); + Assertions.assertEquals(2, created.size()); + } + + @Test + public void activationIdIsKeptPerActor() throws Exception { + List created = new ArrayList<>(); + ActorManager manager = createActivationManager(created); + ActorId first = newActorId(); + ActorId second = newActorId(); + + manager.activateActor(first, "a1").block(); + manager.activateActor(second, "b1").block(); + manager.activateActor(first, "a2").block(); + manager.activateActor(second, "b1").block(); + + Assertions.assertEquals(3, created.size()); + Assertions.assertEquals(1, created.get(0).deactivations); + Assertions.assertEquals(0, created.get(1).deactivations); + } + + @Test + public void deactivateThenNewActivationIdCreatesFreshActor() throws Exception { + List created = new ArrayList<>(); + ActorManager manager = createActivationManager(created); + ActorId actorId = newActorId(); + + manager.activateActor(actorId, "a1").block(); + increment(manager, actorId); + manager.deactivateActor(actorId).block(); + manager.activateActor(actorId, "a2").block(); + + Assertions.assertEquals(2, created.size()); + Assertions.assertEquals(1, created.get(0).deactivations); + Assertions.assertEquals(1, increment(manager, actorId)); + } + + @Test + public void activateWithNewActivationIdIsLazy() throws Exception { + List created = new ArrayList<>(); + ActorManager manager = createActivationManager(created); + ActorId actorId = newActorId(); + + manager.activateActor(actorId, "a1").block(); + Mono reset = manager.activateActor(actorId, "a2"); + Assertions.assertEquals(1, created.size()); + Assertions.assertEquals(0, created.get(0).deactivations); + + reset.block(); + Assertions.assertEquals(2, created.size()); + Assertions.assertEquals(1, created.get(0).deactivations); + } + + @Test + public void replacedActivationSkipsStaleSave() throws Exception { + RecordingStateProvider provider = new RecordingStateProvider(); + List created = new ArrayList<>(); + CompletableFuture gate = new CompletableFuture<>(); + ActorManager manager = createGatedManager(provider, created, gate); + ActorId actorId = newActorId(); + + manager.activateActor(actorId, "a1").block(); + CompletableFuture stale = write(manager, actorId, "stale").toFuture(); + manager.activateActor(actorId, "a2").block(); + write(manager, actorId, "fresh").block(); + gate.complete(null); + + Assertions.assertEquals(List.of("value=fresh"), provider.saved); + assertCause(IllegalStateException.class, stale); + } + + @Test + public void replacedActivationSkipsStaleTimerSave() throws Exception { + RecordingStateProvider provider = new RecordingStateProvider(); + List created = new ArrayList<>(); + CompletableFuture gate = new CompletableFuture<>(); + ActorManager manager = createGatedManager(provider, created, gate); + ActorId actorId = newActorId(); + + manager.activateActor(actorId, "a1").block(); + CompletableFuture stale = manager.invokeTimer(actorId, "t", createTimerParams("write", "stale")).toFuture(); + manager.activateActor(actorId, "a2").block(); + write(manager, actorId, "fresh").block(); + gate.complete(null); + + Assertions.assertEquals(List.of("value=fresh"), provider.saved); + assertCause(IllegalStateException.class, stale); + } + + @Test + public void replacedActivationSkipsStaleReminderSave() throws Exception { + RecordingStateProvider provider = new RecordingStateProvider(); + List created = new ArrayList<>(); + CompletableFuture gate = new CompletableFuture<>(); + ActorManager manager = createGatedManager(provider, created, gate); + ActorId actorId = newActorId(); + + manager.activateActor(actorId, "a1").block(); + CompletableFuture stale = manager.invokeReminder(actorId, "r", createReminderParams("stale")).toFuture(); + manager.activateActor(actorId, "a2").block(); + write(manager, actorId, "fresh").block(); + gate.complete(null); + + Assertions.assertEquals(List.of("value=fresh"), provider.saved); + assertCause(IllegalStateException.class, stale); + } + + @Test + public void liveActivationSavesAfterSameOrMissingId() throws Exception { + RecordingStateProvider provider = new RecordingStateProvider(); + List created = new ArrayList<>(); + CompletableFuture gate = new CompletableFuture<>(); + ActorManager manager = createGatedManager(provider, created, gate); + ActorId actorId = newActorId(); + + manager.activateActor(actorId, "a1").block(); + CompletableFuture call = write(manager, actorId, "live").toFuture(); + manager.activateActor(actorId, "a1").block(); + manager.activateActor(actorId, null).block(); + manager.activateActor(actorId, "").block(); + gate.complete(null); + + call.get(5, TimeUnit.SECONDS); + Assertions.assertEquals(List.of("value=live"), provider.saved); + Assertions.assertEquals(1, created.size()); + } + + @Test + public void deactivatedActorSkipsSave() throws Exception { + RecordingStateProvider provider = new RecordingStateProvider(); + List created = new ArrayList<>(); + CompletableFuture gate = new CompletableFuture<>(); + ActorManager manager = createGatedManager(provider, created, gate); + ActorId actorId = newActorId(); + + manager.activateActor(actorId, "a1").block(); + CompletableFuture call = write(manager, actorId, "stale").toFuture(); + manager.deactivateActor(actorId).block(); + gate.complete(null); + + Assertions.assertEquals(List.of(), provider.saved); + assertCause(IllegalStateException.class, call); + } + + @Test + public void previousActivationIdCreatesFreshActor() throws Exception { + List created = new ArrayList<>(); + ActorManager manager = createActivationManager(created); + ActorId actorId = newActorId(); + + manager.activateActor(actorId, "a1").block(); + manager.activateActor(actorId, "a2").block(); + Assertions.assertEquals(1, increment(manager, actorId)); + manager.activateActor(actorId, "a1").block(); + + Assertions.assertEquals(3, created.size()); + Assertions.assertEquals(1, created.get(1).deactivations); + Assertions.assertEquals(1, created.get(2).activations); + Assertions.assertEquals(1, increment(manager, actorId)); + } + + @Test + public void missingActivationIdAfterReplaceKeepsActor() throws Exception { + List created = new ArrayList<>(); + ActorManager manager = createActivationManager(created); + ActorId actorId = newActorId(); + + manager.activateActor(actorId, "a1").block(); + manager.activateActor(actorId, "a2").block(); + Assertions.assertEquals(1, increment(manager, actorId)); + manager.activateActor(actorId, null).block(); + manager.activateActor(actorId, "").block(); + Assertions.assertEquals(2, increment(manager, actorId)); + Assertions.assertEquals(2, created.size()); + } + + @Test + public void lateActivationWithDifferentIdFailsAndKeepsNewer() throws Exception { + RecordingStateProvider provider = new RecordingStateProvider(); + List created = new ArrayList<>(); + CompletableFuture activateGate = new CompletableFuture<>(); + CompletableFuture callGate = new CompletableFuture<>(); + ActorManager manager = + createLateActivationManager(provider, created, activateGate, callGate, false); + ActorId actorId = newActorId(); + + CompletableFuture late = manager.activateActor(actorId, "a").toFuture(); + manager.activateActor(actorId, "b").block(); + CompletableFuture inFlight = manager.invokeMethod( + actorId, "write", new DefaultObjectSerializer().serialize("fromB")).toFuture(); + activateGate.complete(null); + + assertCause(IllegalStateException.class, late); + callGate.complete(null); + inFlight.get(5, TimeUnit.SECONDS); + + Assertions.assertEquals(List.of("value=newer", "value=fromB"), provider.saved); + Assertions.assertEquals(1, created.get(0).deactivations); + Assertions.assertEquals(0, created.get(1).deactivations); + manager.invokeMethod(actorId, "increment", null).block(); + Assertions.assertEquals(0, created.get(0).calls); + Assertions.assertEquals(1, created.get(1).calls); + } + + @Test + public void lateActivationWithSameIdReusesNewer() throws Exception { + RecordingStateProvider provider = new RecordingStateProvider(); + List created = new ArrayList<>(); + CompletableFuture activateGate = new CompletableFuture<>(); + ActorManager manager = + createLateActivationManager(provider, created, activateGate, new CompletableFuture<>(), false); + ActorId actorId = newActorId(); + + CompletableFuture> late = + manager.activateActor(actorId, "a").toFuture(); + manager.activateActor(actorId, "a").block(); + activateGate.complete(null); + + manager.invokeMethod(late.get(5, TimeUnit.SECONDS), "increment", null).block(); + manager.invokeMethod(actorId, "increment", null).block(); + + Assertions.assertEquals(List.of("value=newer"), provider.saved); + Assertions.assertEquals(1, created.get(0).deactivations); + Assertions.assertEquals(0, created.get(0).calls); + Assertions.assertEquals(0, created.get(1).deactivations); + Assertions.assertEquals(2, created.get(1).calls); + } + + @Test + public void lateActivationWithoutIdReusesNewer() throws Exception { + RecordingStateProvider provider = new RecordingStateProvider(); + List created = new ArrayList<>(); + CompletableFuture activateGate = new CompletableFuture<>(); + ActorManager manager = + createLateActivationManager(provider, created, activateGate, new CompletableFuture<>(), true); + ActorId actorId = newActorId(); + + CompletableFuture> late = + manager.activateActor(actorId, null).toFuture(); + manager.activateActor(actorId, "b").block(); + activateGate.complete(null); + + manager.invokeMethod(late.get(5, TimeUnit.SECONDS), "increment", null).block(); + + Assertions.assertEquals(List.of("value=newer"), provider.saved); + Assertions.assertEquals(1, created.get(0).deactivations); + Assertions.assertEquals(0, created.get(0).calls); + Assertions.assertEquals(1, created.get(1).calls); + } + + @Test + public void resetSurvivesFailingDeactivate() throws Exception { + List created = new ArrayList<>(); + ActorManager manager = createLateActivationManager(new RecordingStateProvider(), created, + CompletableFuture.completedFuture(null), new CompletableFuture<>(), true); + ActorId actorId = newActorId(); + + manager.activateActor(actorId, "a1").block(); + manager.activateActor(actorId, "a2").block(); + manager.invokeMethod(actorId, "increment", null).block(); + + Assertions.assertEquals(2, created.size()); + Assertions.assertEquals(1, created.get(0).deactivations); + Assertions.assertEquals(0, created.get(0).calls); + Assertions.assertEquals(1, created.get(1).calls); + } + + @Test + public void deactivateFailsWhenOnDeactivateThrows() throws Exception { + List created = new ArrayList<>(); + ActorManager manager = createLateActivationManager(new RecordingStateProvider(), created, + CompletableFuture.completedFuture(null), new CompletableFuture<>(), true); + ActorId actorId = newActorId(); + + manager.activateActor(actorId, "a1").block(); + + assertThrows(IllegalArgumentException.class, () -> manager.deactivateActor(actorId).block()); + Assertions.assertEquals(1, created.get(0).deactivations); + } + + @Test + public void concurrentResetKeepsTheWinner() throws Exception { + List created = new ArrayList<>(); + ActorManager manager = createActivationManager(created); + GetHookMap activeActors = hookActiveActors(manager); + ActorId actorId = newActorId(); + + manager.activateActor(actorId, "a1").block(); + List> winner = new ArrayList<>(); + activeActors.onGet = () -> winner.add(manager.activateActor(actorId, "a2").block()); + ActorManager.ActiveActor loser = manager.activateActor(actorId, "a2").block(); + + Assertions.assertSame(winner.get(0), loser); + Assertions.assertEquals(3, created.size()); + Assertions.assertEquals(1, created.get(0).deactivations); + Assertions.assertEquals(0, created.get(1).deactivations); + Assertions.assertEquals(1, created.get(2).deactivations); + Assertions.assertEquals(1, increment(manager, actorId)); + } + + private static GetHookMap hookActiveActors(ActorManager manager) throws Exception { + Field actors = ActorManager.class.getDeclaredField("activeActors"); + actors.setAccessible(true); + GetHookMap map = new GetHookMap<>(); + actors.set(manager, map); + return map; + } + + private static ActorManager createLateActivationManager( + RecordingStateProvider provider, List created, CompletableFuture activateGate, + CompletableFuture callGate, boolean failLateDeactivate) { + ActorRuntimeContext context = new ActorRuntimeContext( + mock(ActorRuntime.class), + new DefaultObjectSerializer(), + (ActorFactory) (ctx, id) -> { + ActivateWriteActorImpl actor = created.isEmpty() + ? new ActivateWriteActorImpl(ctx, id, Mono.fromFuture(activateGate), Mono.empty(), "late", + failLateDeactivate) + : new ActivateWriteActorImpl(ctx, id, Mono.empty(), Mono.fromFuture(callGate), "newer", false); + created.add(actor); + return actor; + }, + ActorTypeInformation.create(ActivateWriteActorImpl.class), + mock(DaprClient.class), + provider + ); + return new ActorManager<>(context); + } + + private static ActorManager createGatedManager( + RecordingStateProvider provider, List created, CompletableFuture gate) { + ActorRuntimeContext context = new ActorRuntimeContext( + mock(ActorRuntime.class), + new DefaultObjectSerializer(), + (ActorFactory) (ctx, id) -> { + Mono actorGate = created.isEmpty() ? Mono.fromFuture(gate) : Mono.empty(); + GatedActorImpl actor = new GatedActorImpl(ctx, id, actorGate); + created.add(actor); + return actor; + }, + ActorTypeInformation.create(GatedActorImpl.class), + mock(DaprClient.class), + provider + ); + return new ActorManager<>(context); + } + + private static Mono write(ActorManager manager, ActorId actorId, String value) + throws IOException { + return manager.invokeMethod(actorId, "write", new DefaultObjectSerializer().serialize(value)); + } + + private static void assertCause(Class expected, CompletableFuture future) { + ExecutionException e = assertThrows(ExecutionException.class, () -> future.get(5, TimeUnit.SECONDS)); + Assertions.assertTrue(expected.isInstance(e.getCause()), String.valueOf(e.getCause())); + } + + private static ActorManager createActivationManager(List created) { + ActorRuntimeContext context = new ActorRuntimeContext( + mock(ActorRuntime.class), + new DefaultObjectSerializer(), + (ActorFactory) (ctx, id) -> { + ActivationActorImpl actor = new ActivationActorImpl(ctx, id); + created.add(actor); + return actor; + }, + ActorTypeInformation.create(ActivationActorImpl.class), + mock(DaprClient.class), + mock(DaprStateAsyncProvider.class) + ); + return new ActorManager<>(context); + } + + private static int increment(ActorManager manager, ActorId actorId) throws IOException { + byte[] response = manager.invokeMethod(actorId, "increment", null).block(); + return new DefaultObjectSerializer().deserialize(response, Integer.class); + } + private byte[] createReminderParams(String data) throws IOException { byte[] serializedData = this.context.getObjectSerializer().serialize(data); ActorReminderParams p = new ActorReminderParams(serializedData, Duration.ofSeconds(1), Duration.ofSeconds(1)); diff --git a/sdk-actors/src/test/java/io/dapr/actors/runtime/ActorRuntimeTest.java b/sdk-actors/src/test/java/io/dapr/actors/runtime/ActorRuntimeTest.java index 5254a68410..c2e3dc8dc3 100644 --- a/sdk-actors/src/test/java/io/dapr/actors/runtime/ActorRuntimeTest.java +++ b/sdk-actors/src/test/java/io/dapr/actors/runtime/ActorRuntimeTest.java @@ -29,11 +29,19 @@ import java.lang.reflect.Field; import java.lang.reflect.InvocationTargetException; import java.time.Duration; +import java.util.ArrayList; import java.util.Arrays; +import java.util.HashMap; +import java.util.List; +import java.util.Map; import java.util.UUID; +import java.util.concurrent.CompletableFuture; +import java.util.concurrent.ExecutionException; +import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicReference; import static org.junit.jupiter.api.Assertions.assertThrows; +import static org.mockito.ArgumentMatchers.any; import static org.mockito.Mockito.mock; import static org.mockito.Mockito.never; import static org.mockito.Mockito.times; @@ -137,6 +145,96 @@ public void receiveTimer(String data) { } } + private static final String ACTIVATION_ACTOR_NAME = "MyActivationActor"; + + public interface MyActivationActor { + int increment(); + } + + @ActorType(name = ACTIVATION_ACTOR_NAME) + public static class MyActivationActorImpl extends AbstractActor + implements MyActivationActor, Remindable { + + private int calls; + + private int reminders; + + private int deactivations; + + public MyActivationActorImpl(ActorRuntimeContext runtimeContext, ActorId id) { + super(runtimeContext, id); + } + + @Override + protected Mono onDeactivate() { + return Mono.fromRunnable(() -> this.deactivations++); + } + + @Override + public int increment() { + return ++this.calls; + } + + @Override + public TypeRef getStateType() { + return TypeRef.STRING; + } + + @Override + public Mono receiveReminder(String reminderName, String state, Duration dueTime, Duration period) { + return Mono.fromRunnable(() -> this.reminders++); + } + } + + private static final String GATED_ACTOR_NAME = "MyGatedActor"; + + public interface MyGatedActor { + Mono hold(String value); + + Mono pause(); + } + + @ActorType(name = GATED_ACTOR_NAME) + public static class MyGatedActorImpl extends AbstractActor implements MyGatedActor { + + private final Mono holdGate; + + private final Mono pauseGate; + + public MyGatedActorImpl(ActorRuntimeContext runtimeContext, ActorId id, Mono holdGate, + Mono pauseGate) { + super(runtimeContext, id); + this.holdGate = holdGate; + this.pauseGate = pauseGate; + } + + @Override + public Mono hold(String value) { + return this.getActorStateManager().set("value", value).then(this.holdGate); + } + + @Override + public Mono pause() { + return this.pauseGate; + } + } + + static class RegisterHookMap extends HashMap { + + private Runnable onPut; + + @Override + public V putIfAbsent(K key, V value) { + V previous = super.putIfAbsent(key, value); + Runnable hook = this.onPut; + this.onPut = null; + if (hook != null) { + hook.run(); + } + return previous; + } + } + private static final ActorObjectSerializer ACTOR_STATE_SERIALIZER = new ActorObjectSerializer(); private static Constructor constructor; @@ -417,6 +515,196 @@ public void lazyInvoke() throws Exception { Assertions.assertEquals(1, count); } + @Test + public void invokeWithSameActivationIdReusesActor() throws Exception { + List created = registerActivationActor(); + String actorId = UUID.randomUUID().toString(); + + Assertions.assertEquals(1, invokeIncrement(actorId, "a1")); + Assertions.assertEquals(2, invokeIncrement(actorId, "a1")); + Assertions.assertEquals(1, created.size()); + } + + @Test + public void invokeWithNewActivationIdReplacesActor() throws Exception { + List created = registerActivationActor(); + String actorId = UUID.randomUUID().toString(); + + Assertions.assertEquals(1, invokeIncrement(actorId, "a1")); + Assertions.assertEquals(2, invokeIncrement(actorId, "a1")); + Assertions.assertEquals(1, invokeIncrement(actorId, "a2")); + + Assertions.assertEquals(2, created.size()); + Assertions.assertEquals(1, created.get(0).deactivations); + Assertions.assertEquals(0, created.get(1).deactivations); + } + + @Test + public void invokeWithoutActivationIdKeepsActor() throws Exception { + List created = registerActivationActor(); + String actorId = UUID.randomUUID().toString(); + + Assertions.assertEquals(1, invokeIncrement(actorId, "a1")); + Assertions.assertEquals(2, invokeIncrement(actorId, null)); + Assertions.assertEquals(3, invokeIncrement(actorId, "")); + byte[] response = this.runtime.invoke(ACTIVATION_ACTOR_NAME, actorId, "increment", null).block(); + Assertions.assertEquals(4, (int) ACTOR_STATE_SERIALIZER.deserialize(response, Integer.class)); + + Assertions.assertEquals(1, created.size()); + Assertions.assertEquals(0, created.get(0).deactivations); + } + + @Test + public void invokeWithActivationIdAfterLegacyActivationReplacesActor() throws Exception { + List created = registerActivationActor(); + String actorId = UUID.randomUUID().toString(); + + this.runtime.invoke(ACTIVATION_ACTOR_NAME, actorId, "increment", null).block(); + Assertions.assertEquals(1, invokeIncrement(actorId, "a1")); + + Assertions.assertEquals(2, created.size()); + Assertions.assertEquals(1, created.get(0).deactivations); + } + + @Test + public void invokeReminderWithNewActivationIdReplacesActor() throws Exception { + List created = registerActivationActor(); + String actorId = UUID.randomUUID().toString(); + + Assertions.assertEquals(1, invokeIncrement(actorId, "a1")); + this.runtime.invokeReminder( + ACTIVATION_ACTOR_NAME, actorId, "myReminder", createReminderParams("hello"), "a1").block(); + Assertions.assertEquals(1, created.size()); + Assertions.assertEquals(1, created.get(0).reminders); + + this.runtime.invokeReminder( + ACTIVATION_ACTOR_NAME, actorId, "myReminder", createReminderParams("hello"), "a2").block(); + Assertions.assertEquals(2, created.size()); + Assertions.assertEquals(1, created.get(0).deactivations); + Assertions.assertEquals(1, created.get(1).reminders); + Assertions.assertEquals(1, invokeIncrement(actorId, "a2")); + } + + @Test + public void invokeReminderWithoutActivationIdKeepsActor() throws Exception { + List created = registerActivationActor(); + String actorId = UUID.randomUUID().toString(); + + invokeIncrement(actorId, "a1"); + this.runtime.invokeReminder( + ACTIVATION_ACTOR_NAME, actorId, "myReminder", createReminderParams("hello"), null).block(); + this.runtime.invokeReminder( + ACTIVATION_ACTOR_NAME, actorId, "myReminder", createReminderParams("hello"), "").block(); + + Assertions.assertEquals(1, created.size()); + Assertions.assertEquals(2, created.get(0).reminders); + } + + @Test + public void invokeTimerWithNewActivationIdReplacesActor() throws Exception { + List created = registerActivationActor(); + String actorId = UUID.randomUUID().toString(); + + Assertions.assertEquals(1, invokeIncrement(actorId, "a1")); + this.runtime.invokeTimer( + ACTIVATION_ACTOR_NAME, actorId, "myTimer", createTimerParams("increment", "x"), "a1").block(); + Assertions.assertEquals(1, created.size()); + Assertions.assertEquals(2, created.get(0).calls); + + this.runtime.invokeTimer( + ACTIVATION_ACTOR_NAME, actorId, "myTimer", createTimerParams("increment", "x"), "a2").block(); + Assertions.assertEquals(2, created.size()); + Assertions.assertEquals(1, created.get(0).deactivations); + Assertions.assertEquals(1, created.get(1).calls); + } + + @Test + public void invokeTimerWithoutActivationIdKeepsActor() throws Exception { + List created = registerActivationActor(); + String actorId = UUID.randomUUID().toString(); + + invokeIncrement(actorId, "a1"); + this.runtime.invokeTimer( + ACTIVATION_ACTOR_NAME, actorId, "myTimer", createTimerParams("increment", "x"), null).block(); + this.runtime.invokeTimer( + ACTIVATION_ACTOR_NAME, actorId, "myTimer", createTimerParams("increment", "x"), "").block(); + + Assertions.assertEquals(1, created.size()); + Assertions.assertEquals(3, created.get(0).calls); + } + + @Test + public void deactivateThenInvokeWithSameActivationIdCreatesActor() throws Exception { + List created = registerActivationActor(); + String actorId = UUID.randomUUID().toString(); + + invokeIncrement(actorId, "a1"); + this.runtime.deactivate(ACTIVATION_ACTOR_NAME, actorId).block(); + Assertions.assertEquals(1, invokeIncrement(actorId, "a1")); + + Assertions.assertEquals(2, created.size()); + Assertions.assertEquals(1, created.get(0).deactivations); + } + + @Test + public void callRunsOnTheActorItsActivationSettledOn() throws Exception { + List saved = new ArrayList<>(); + when(this.mockDaprClient.getState(any(), any(), any())).thenReturn(Mono.empty()); + when(this.mockDaprClient.saveStateTransactionally(any(), any(), any())).thenAnswer(invocation -> { + saved.addAll(invocation.getArgument(2)); + return Mono.empty(); + }); + + CompletableFuture holdGate = new CompletableFuture<>(); + CompletableFuture pauseGate = new CompletableFuture<>(); + this.runtime.registerActor(MyGatedActorImpl.class, + (context, id) -> new MyGatedActorImpl(context, id, Mono.fromFuture(holdGate), Mono.fromFuture(pauseGate))); + + RegisterHookMap activeActors = hookActiveActors(GATED_ACTOR_NAME); + String actorId = UUID.randomUUID().toString(); + byte[] value = new DefaultObjectSerializer().serialize("fresh"); + List> live = new ArrayList<>(); + activeActors.onPut = () -> live.add( + this.runtime.invoke(GATED_ACTOR_NAME, actorId, "hold", value, "a2").toFuture()); + + CompletableFuture late = this.runtime.invoke(GATED_ACTOR_NAME, actorId, "pause", null, "a1").toFuture(); + holdGate.complete(null); + live.get(0).get(5, TimeUnit.SECONDS); + pauseGate.complete(null); + + Assertions.assertEquals(1, saved.size()); + Assertions.assertEquals("value", saved.get(0).getKey()); + Assertions.assertEquals("\"fresh\"", saved.get(0).getValue()); + ExecutionException e = assertThrows(ExecutionException.class, () -> late.get(5, TimeUnit.SECONDS)); + Assertions.assertTrue(e.getCause() instanceof IllegalStateException, String.valueOf(e.getCause())); + } + + private RegisterHookMap hookActiveActors(String actorType) throws Exception { + Field managers = ActorRuntime.class.getDeclaredField("actorManagers"); + managers.setAccessible(true); + Object manager = ((Map) managers.get(this.runtime)).get(actorType); + Field actors = ActorManager.class.getDeclaredField("activeActors"); + actors.setAccessible(true); + RegisterHookMap map = new RegisterHookMap<>(); + actors.set(manager, map); + return map; + } + + private List registerActivationActor() { + List created = new ArrayList<>(); + this.runtime.registerActor(MyActivationActorImpl.class, (context, id) -> { + MyActivationActorImpl actor = new MyActivationActorImpl(context, id); + created.add(actor); + return actor; + }); + return created; + } + + private int invokeIncrement(String actorId, String activationId) throws IOException { + byte[] response = this.runtime.invoke(ACTIVATION_ACTOR_NAME, actorId, "increment", null, activationId).block(); + return ACTOR_STATE_SERIALIZER.deserialize(response, Integer.class); + } + private static byte[] createReminderParams(String data) throws IOException { byte[] serializedData = new DefaultObjectSerializer().serialize(data); ActorReminderParams params = diff --git a/sdk-springboot/pom.xml b/sdk-springboot/pom.xml index fbe963c74d..223a6c7a9e 100644 --- a/sdk-springboot/pom.xml +++ b/sdk-springboot/pom.xml @@ -59,6 +59,16 @@ spring-boot-starter-test test + + org.springframework + spring-webmvc + test + + + jakarta.servlet + jakarta.servlet-api + test + 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..4f003baf45 100644 --- a/sdk-springboot/src/main/java/io/dapr/springboot/DaprController.java +++ b/sdk-springboot/src/main/java/io/dapr/springboot/DaprController.java @@ -14,6 +14,7 @@ package io.dapr.springboot; import io.dapr.actors.runtime.ActorRuntime; +import io.dapr.client.Headers; import io.dapr.serializer.DefaultObjectSerializer; import org.springframework.http.MediaType; import org.springframework.web.bind.annotation.DeleteMapping; @@ -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,29 @@ public Mono deactivateActor(@PathVariable("type") String type, * @param id Actor Id. * @param method Actor method. * @param body Raw request body. + * @param activationId Actor activation id sent by Dapr, or null if absent. * @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 = Headers.DAPR_ACTOR_ACTIVATION_ID, required = false) + String activationId) { + return ActorRuntime.getInstance().invoke(type, id, method, body, activationId); + } + + /** + * Kept for source compatibility, not mapped to a route. + * @param type Actor type. + * @param id Actor Id. + * @param method Actor method. + * @param body Raw request body. + * @return Raw response body. + */ + public Mono invokeActorMethod(String type, String id, String method, byte[] body) { + return invokeActorMethod(type, id, method, body, null); } /** @@ -98,14 +115,29 @@ public Mono invokeActorMethod(@PathVariable("type") String type, * @param id Actor Id. * @param timer Actor timer's name. * @param body Raw request's body. + * @param activationId Actor activation id sent by Dapr, or null if absent. * @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 = Headers.DAPR_ACTOR_ACTIVATION_ID, required = false) + String activationId) { + return ActorRuntime.getInstance().invokeTimer(type, id, timer, body, activationId); + } + + /** + * Kept for source compatibility, not mapped to a route. + * @param type Actor type. + * @param id Actor Id. + * @param timer Actor timer's name. + * @param body Raw request's body. + * @return Void. + */ + public Mono invokeActorTimer(String type, String id, String timer, byte[] body) { + return invokeActorTimer(type, id, timer, body, null); } /** @@ -114,13 +146,28 @@ public Mono invokeActorTimer(@PathVariable("type") String type, * @param id Actor Id. * @param reminder Actor reminder's name. * @param body Raw request's body. + * @param activationId Actor activation id sent by Dapr, or null if absent. * @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 = Headers.DAPR_ACTOR_ACTIVATION_ID, required = false) + String activationId) { + return ActorRuntime.getInstance().invokeReminder(type, id, reminder, body, activationId); + } + + /** + * Kept for source compatibility, not mapped to a route. + * @param type Actor type. + * @param id Actor Id. + * @param reminder Actor reminder's name. + * @param body Raw request's body. + * @return Void. + */ + public Mono invokeActorReminder(String type, String id, String reminder, byte[] body) { + return invokeActorReminder(type, id, reminder, body, null); } } diff --git a/sdk-springboot/src/test/java/io/dapr/springboot/DaprControllerTest.java b/sdk-springboot/src/test/java/io/dapr/springboot/DaprControllerTest.java new file mode 100644 index 0000000000..9701f24663 --- /dev/null +++ b/sdk-springboot/src/test/java/io/dapr/springboot/DaprControllerTest.java @@ -0,0 +1,142 @@ +/* + * 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.springboot; + +import io.dapr.actors.runtime.ActorRuntime; +import org.junit.jupiter.api.AfterEach; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.params.ParameterizedTest; +import org.junit.jupiter.params.provider.ValueSource; +import org.springframework.test.web.servlet.MockMvc; +import org.springframework.test.web.servlet.MvcResult; +import org.springframework.test.web.servlet.RequestBuilder; +import org.springframework.test.web.servlet.setup.MockMvcBuilders; +import reactor.core.publisher.Mono; + +import java.lang.reflect.Field; + +import static org.mockito.ArgumentMatchers.any; +import static org.mockito.ArgumentMatchers.isNull; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.verify; +import static org.mockito.Mockito.when; +import static org.springframework.test.web.servlet.request.MockMvcRequestBuilders.asyncDispatch; +import static org.springframework.test.web.servlet.request.MockMvcRequestBuilders.delete; +import static org.springframework.test.web.servlet.request.MockMvcRequestBuilders.put; +import static org.springframework.test.web.servlet.result.MockMvcResultMatchers.request; +import static org.springframework.test.web.servlet.result.MockMvcResultMatchers.status; + +public class DaprControllerTest { + + private static final byte[] BODY = "{}".getBytes(); + + private Field instanceField; + + private Object previousInstance; + + private ActorRuntime actorRuntime; + + private MockMvc mockMvc; + + @BeforeEach + public void setup() throws Exception { + instanceField = ActorRuntime.class.getDeclaredField("instance"); + instanceField.setAccessible(true); + previousInstance = instanceField.get(null); + actorRuntime = mock(ActorRuntime.class); + instanceField.set(null, actorRuntime); + mockMvc = MockMvcBuilders.standaloneSetup(new DaprController()).build(); + + when(actorRuntime.invoke(any(), any(), any(), any(), any())).thenReturn(Mono.just(new byte[0])); + when(actorRuntime.invokeTimer(any(), any(), any(), any(), any())).thenReturn(Mono.empty()); + when(actorRuntime.invokeReminder(any(), any(), any(), any(), any())).thenReturn(Mono.empty()); + when(actorRuntime.deactivate(any(), any())).thenReturn(Mono.empty()); + } + + @AfterEach + public void tearDown() throws Exception { + instanceField.set(null, previousInstance); + } + + @ParameterizedTest + @ValueSource(strings = {"Dapr-Actor-Activation-Id", "dapr-actor-activation-id", "DAPR-ACTOR-ACTIVATION-ID"}) + public void invokeActorMethodPassesActivationId(String header) throws Exception { + perform(put("/actors/MyActor/1/method/say").header(header, "act-1").content(BODY)); + + verify(actorRuntime).invoke("MyActor", "1", "say", BODY, "act-1"); + } + + @Test + public void invokeActorMethodWithoutActivationIdPassesNull() throws Exception { + perform(put("/actors/MyActor/1/method/say").content(BODY)); + + verify(actorRuntime).invoke(any(), any(), any(), any(), isNull()); + } + + @ParameterizedTest + @ValueSource(strings = {"Dapr-Actor-Activation-Id", "dapr-actor-activation-id"}) + public void invokeActorTimerPassesActivationId(String header) throws Exception { + perform(put("/actors/MyActor/1/method/timer/t1").header(header, "act-2").content(BODY)); + + verify(actorRuntime).invokeTimer("MyActor", "1", "t1", BODY, "act-2"); + } + + @Test + public void invokeActorTimerWithoutActivationIdPassesNull() throws Exception { + perform(put("/actors/MyActor/1/method/timer/t1").content(BODY)); + + verify(actorRuntime).invokeTimer(any(), any(), any(), any(), isNull()); + } + + @ParameterizedTest + @ValueSource(strings = {"Dapr-Actor-Activation-Id", "dapr-actor-activation-id"}) + public void invokeActorReminderPassesActivationId(String header) throws Exception { + perform(put("/actors/MyActor/1/method/remind/r1").header(header, "act-3").content(BODY)); + + verify(actorRuntime).invokeReminder("MyActor", "1", "r1", BODY, "act-3"); + } + + @Test + public void invokeActorReminderWithoutActivationIdPassesNull() throws Exception { + perform(put("/actors/MyActor/1/method/remind/r1").content(BODY)); + + verify(actorRuntime).invokeReminder(any(), any(), any(), any(), isNull()); + } + + @Test + public void deactivateActorIsUnchanged() throws Exception { + perform(delete("/actors/MyActor/1").header("Dapr-Actor-Activation-Id", "act-4")); + + verify(actorRuntime).deactivate("MyActor", "1"); + } + + @Test + public void legacyOverloadsPassNullActivationId() { + DaprController controller = new DaprController(); + + controller.invokeActorMethod("MyActor", "1", "say", BODY).block(); + controller.invokeActorTimer("MyActor", "1", "t1", BODY).block(); + controller.invokeActorReminder("MyActor", "1", "r1", BODY).block(); + + verify(actorRuntime).invoke("MyActor", "1", "say", BODY, null); + verify(actorRuntime).invokeTimer("MyActor", "1", "t1", BODY, null); + verify(actorRuntime).invokeReminder("MyActor", "1", "r1", BODY, null); + } + + private void perform(RequestBuilder requestBuilder) throws Exception { + MvcResult result = mockMvc.perform(requestBuilder).andExpect(request().asyncStarted()).andReturn(); + mockMvc.perform(asyncDispatch(result)).andExpect(status().isOk()); + } +} diff --git a/sdk/src/main/java/io/dapr/client/Headers.java b/sdk/src/main/java/io/dapr/client/Headers.java index 5b76510e55..9bd5616d60 100644 --- a/sdk/src/main/java/io/dapr/client/Headers.java +++ b/sdk/src/main/java/io/dapr/client/Headers.java @@ -37,4 +37,7 @@ public final class Headers { * W3C Baggage header for context propagation. */ public static final String BAGGAGE = "baggage"; + + /** Sent by Dapr on actor callbacks to identify one activation. */ + public static final String DAPR_ACTOR_ACTIVATION_ID = "Dapr-Actor-Activation-Id"; } From 9f3def03f6e20c4ea5bb3c399aad1673f990e19a Mon Sep 17 00:00:00 2001 From: Cassandra Coyle Date: Mon, 26 Oct 2026 15:14:18 -0500 Subject: [PATCH 2/4] skip save tweaks Signed-off-by: Cassandra Coyle --- .../io/dapr/actors/runtime/AbstractActor.java | 4 +- .../io/dapr/actors/runtime/ActorManager.java | 11 ++- .../dapr/actors/runtime/ActorManagerTest.java | 71 ++++++++++++++++++- 3 files changed, 78 insertions(+), 8 deletions(-) diff --git a/sdk-actors/src/main/java/io/dapr/actors/runtime/AbstractActor.java b/sdk-actors/src/main/java/io/dapr/actors/runtime/AbstractActor.java index 11307356a3..6b44cafad3 100644 --- a/sdk-actors/src/main/java/io/dapr/actors/runtime/AbstractActor.java +++ b/sdk-actors/src/main/java/io/dapr/actors/runtime/AbstractActor.java @@ -385,15 +385,17 @@ Mono onPreActorMethodInternal(ActorMethodContext actorMethodContext) { * Internal callback after method is invoked. * * @param actorMethodContext Method context. + * @param beforeSave Runs after onPostActorMethod, an error skips the state save. * @return Asynchronous void response. */ - Mono onPostActorMethodInternal(ActorMethodContext actorMethodContext) { + Mono onPostActorMethodInternal(ActorMethodContext actorMethodContext, Mono beforeSave) { return Mono.fromRunnable(() -> { if (!this.started.get()) { throw new IllegalStateException("Cannot complete a method before starting a call."); } }) .then(this.onPostActorMethod(actorMethodContext)) + .then(beforeSave) .then(this.saveState()) .then(Mono.fromRunnable(() -> this.started.set(false))); } 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 3c072351d3..66e718d763 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 @@ -372,9 +372,8 @@ private Mono invoke(ActiveActor active, ActorMethodContext context, Fu }) .then(actor.onPreActorMethodInternal(context)) .then((Mono) func.apply(actor)) - .switchIfEmpty( - this.ifLive(active, actor.onPostActorMethodInternal(context))) - .flatMap(r -> this.ifLive(active, actor.onPostActorMethodInternal(context)).thenReturn(r)) + .switchIfEmpty(actor.onPostActorMethodInternal(context, this.checkLive(active))) + .flatMap(r -> actor.onPostActorMethodInternal(context, this.checkLive(active)).thenReturn(r)) .map(o -> (R) o); } catch (Exception e) { return Mono.error(e); @@ -395,8 +394,8 @@ private Mono> lookup(ActorId actorId) { }); } - /** Skips the post step state save if the instance was replaced during the call. */ - private Mono ifLive(ActiveActor active, Mono post) { + /** Fails the call before its state save if the instance was replaced during the call. */ + private Mono checkLive(ActiveActor active) { return Mono.defer(() -> { ActorId actorId = active.actor.getId(); if (this.activeActors.get(actorId) != active) { @@ -406,7 +405,7 @@ private Mono ifLive(ActiveActor active, Mono post) { this.runtimeContext.getActorTypeInformation().getName()))); } - return post; + return Mono.empty(); }); } diff --git a/sdk-actors/src/test/java/io/dapr/actors/runtime/ActorManagerTest.java b/sdk-actors/src/test/java/io/dapr/actors/runtime/ActorManagerTest.java index 7175ef77e2..6cd2a604c0 100644 --- a/sdk-actors/src/test/java/io/dapr/actors/runtime/ActorManagerTest.java +++ b/sdk-actors/src/test/java/io/dapr/actors/runtime/ActorManagerTest.java @@ -163,8 +163,12 @@ interface GatedActor { int increment(); } + interface FailingActor { + Mono writeThenFail(String value); + } + @ActorType(name = "GatedActor") - public static class GatedActorImpl extends AbstractActor implements GatedActor, Remindable { + public static class GatedActorImpl extends AbstractActor implements GatedActor, FailingActor, Remindable { private final Mono gate; @@ -172,6 +176,8 @@ public static class GatedActorImpl extends AbstractActor implements GatedActor, private int deactivations; + private int postHooks; + public GatedActorImpl(ActorRuntimeContext runtimeContext, ActorId id, Mono gate) { super(runtimeContext, id); this.gate = gate; @@ -182,11 +188,21 @@ protected Mono onDeactivate() { return Mono.fromRunnable(() -> this.deactivations++); } + @Override + protected Mono onPostActorMethod(ActorMethodContext actorMethodContext) { + return Mono.fromRunnable(() -> this.postHooks++); + } + @Override public Mono write(String value) { return this.gate.then(Mono.defer(() -> this.getActorStateManager().set("value", value))); } + @Override + public Mono writeThenFail(String value) { + return this.write(value).then(Mono.error(new IllegalArgumentException("callback failed"))); + } + @Override public int increment() { return ++this.calls; @@ -626,6 +642,59 @@ public void replacedActivationSkipsStaleReminderSave() throws Exception { assertCause(IllegalStateException.class, stale); } + @Test + public void replacedActivationRunsPostHookButSkipsSave() throws Exception { + RecordingStateProvider provider = new RecordingStateProvider(); + List created = new ArrayList<>(); + CompletableFuture gate = new CompletableFuture<>(); + ActorManager manager = createGatedManager(provider, created, gate); + ActorId actorId = newActorId(); + + manager.activateActor(actorId, "a1").block(); + CompletableFuture stale = write(manager, actorId, "stale").toFuture(); + manager.activateActor(actorId, "a2").block(); + gate.complete(null); + + assertCause(IllegalStateException.class, stale); + Assertions.assertEquals(1, created.get(0).postHooks); + Assertions.assertEquals(List.of(), provider.saved); + } + + @Test + public void deactivatedActorRunsPostHookButSkipsSave() throws Exception { + RecordingStateProvider provider = new RecordingStateProvider(); + List created = new ArrayList<>(); + CompletableFuture gate = new CompletableFuture<>(); + ActorManager manager = createGatedManager(provider, created, gate); + ActorId actorId = newActorId(); + + manager.activateActor(actorId, "a1").block(); + CompletableFuture stale = manager.invokeTimer(actorId, "t", createTimerParams("write", "stale")).toFuture(); + manager.deactivateActor(actorId).block(); + gate.complete(null); + + assertCause(IllegalStateException.class, stale); + Assertions.assertEquals(1, created.get(0).postHooks); + Assertions.assertEquals(List.of(), provider.saved); + } + + @Test + public void failingTimerCallbackSkipsSave() throws Exception { + RecordingStateProvider provider = new RecordingStateProvider(); + List created = new ArrayList<>(); + ActorManager manager = + createGatedManager(provider, created, CompletableFuture.completedFuture(null)); + ActorId actorId = newActorId(); + + manager.activateActor(actorId, "a1").block(); + CompletableFuture call = + manager.invokeTimer(actorId, "t", createTimerParams("writeThenFail", "lost")).toFuture(); + + assertCause(IllegalArgumentException.class, call); + Assertions.assertEquals(0, created.get(0).postHooks); + Assertions.assertEquals(List.of(), provider.saved); + } + @Test public void liveActivationSavesAfterSameOrMissingId() throws Exception { RecordingStateProvider provider = new RecordingStateProvider(); From c5ee38af2feb83db32105163f2f8287c8418e0bf Mon Sep 17 00:00:00 2001 From: Cassandra Coyle Date: Mon, 26 Oct 2026 16:04:20 -0500 Subject: [PATCH 3/4] test(actors): don't keep reminder formatter in actor state across reactivation Signed-off-by: Cassandra Coyle --- .../src/test/java/io/dapr/it/actors/app/MyActorBase.java | 5 +++-- 1 file changed, 3 insertions(+), 2 deletions(-) diff --git a/sdk-tests/src/test/java/io/dapr/it/actors/app/MyActorBase.java b/sdk-tests/src/test/java/io/dapr/it/actors/app/MyActorBase.java index b05dc3b2bf..6a90f21d33 100644 --- a/sdk-tests/src/test/java/io/dapr/it/actors/app/MyActorBase.java +++ b/sdk-tests/src/test/java/io/dapr/it/actors/app/MyActorBase.java @@ -44,7 +44,9 @@ public abstract class MyActorBase extends AbstractActor implements MyActor, R private ActorReminderDataParam actorReminderDataParam = new ActorReminderDataParam("36", String.class.getSimpleName()); - private Function toStringFunc = Objects::toString; + // the object is rebuilt on reactivation, so this can't be set by startReminder + private final Function toStringFunc = + t -> t instanceof byte[] b ? Base64.getEncoder().encodeToString(b) : Objects.toString(t); private final TypeRef reminderStateRef; @@ -101,7 +103,6 @@ public void startReminder(String name) throws Exception { break; case "Binary": data = actorReminderDataParam.getBinaryData(); - toStringFunc = t -> Base64.getEncoder().encodeToString((byte[])t); break; case "Object": data = actorReminderDataParam.asObject(TypeRef.get(MyObject.class)); From 33cff3ea40101a092a3ceb9c3168f9cd9325c4e0 Mon Sep 17 00:00:00 2001 From: Cassandra Coyle Date: Tue, 27 Oct 2026 10:52:09 -0500 Subject: [PATCH 4/4] check version to prevent dos Signed-off-by: Cassandra Coyle --- .../dapr/actors/runtime/ActivationIdGate.java | 99 ++++++++++++++++ .../io/dapr/actors/runtime/ActorRuntime.java | 15 ++- .../io/dapr/actors/runtime/DaprClient.java | 7 ++ .../dapr/actors/runtime/DaprClientImpl.java | 11 ++ .../actors/runtime/ActivationIdGateTest.java | 112 ++++++++++++++++++ .../dapr/actors/runtime/ActorRuntimeTest.java | 35 ++++++ 6 files changed, 276 insertions(+), 3 deletions(-) create mode 100644 sdk-actors/src/main/java/io/dapr/actors/runtime/ActivationIdGate.java create mode 100644 sdk-actors/src/test/java/io/dapr/actors/runtime/ActivationIdGateTest.java diff --git a/sdk-actors/src/main/java/io/dapr/actors/runtime/ActivationIdGate.java b/sdk-actors/src/main/java/io/dapr/actors/runtime/ActivationIdGate.java new file mode 100644 index 0000000000..e13d6bded8 --- /dev/null +++ b/sdk-actors/src/main/java/io/dapr/actors/runtime/ActivationIdGate.java @@ -0,0 +1,99 @@ +/* + * 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; + +import reactor.core.publisher.Mono; + +import java.time.Duration; +import java.util.Optional; +import java.util.concurrent.atomic.AtomicReference; +import java.util.function.Supplier; +import java.util.regex.Matcher; +import java.util.regex.Pattern; + +/** + * Trusts the activation id only from a Dapr runtime that strips it from caller headers. + */ +class ActivationIdGate { + + private static final Pattern VERSION = Pattern.compile("^(\\d{1,9})\\.(\\d{1,9})\\.(\\d{1,9})([-+].*)?$"); + + private static final int[] MIN_VERSION = {1, 19, 1}; + + private static final Duration TIMEOUT = Duration.ofSeconds(5); + + private final Supplier> runtimeVersion; + + private final AtomicReference> lookup = new AtomicReference<>(); + + ActivationIdGate(Supplier> runtimeVersion) { + this.runtimeVersion = runtimeVersion; + } + + /** + * Returns the activation id if the Dapr runtime is trusted to set it, or empty otherwise. + * + * @param activationId Activation id sent by Dapr, or null if absent. + * @return The trusted activation id, or empty. + */ + Mono> filter(String activationId) { + if (activationId == null || activationId.isEmpty()) { + return Mono.just(Optional.empty()); + } + return Mono.defer(this::trusted) + .map(t -> t ? Optional.of(activationId) : Optional.empty()) + .onErrorReturn(Optional.empty()); + } + + private Mono trusted() { + Mono cached = this.lookup.get(); + if (cached != null) { + return cached; + } + AtomicReference> self = new AtomicReference<>(); + Mono next = Mono.defer(this.runtimeVersion) + .timeout(TIMEOUT) + .defaultIfEmpty("") + .map(ActivationIdGate::supported) + .doOnError(e -> this.lookup.compareAndSet(self.get(), null)) + .cache(); + self.set(next); + while (true) { + Mono current = this.lookup.get(); + if (current != null) { + return current; + } + if (this.lookup.compareAndSet(null, next)) { + return next; + } + } + } + + static boolean supported(String version) { + if (version == null) { + return false; + } + Matcher m = VERSION.matcher(version); + if (!m.matches()) { + return false; + } + for (int i = 0; i < MIN_VERSION.length; i++) { + int part = Integer.parseInt(m.group(i + 1)); + if (part != MIN_VERSION[i]) { + return part > MIN_VERSION[i]; + } + } + return true; + } +} diff --git a/sdk-actors/src/main/java/io/dapr/actors/runtime/ActorRuntime.java b/sdk-actors/src/main/java/io/dapr/actors/runtime/ActorRuntime.java index da766cfcbf..ccb73d9382 100644 --- a/sdk-actors/src/main/java/io/dapr/actors/runtime/ActorRuntime.java +++ b/sdk-actors/src/main/java/io/dapr/actors/runtime/ActorRuntime.java @@ -73,6 +73,11 @@ public class ActorRuntime implements Closeable { */ private final ConcurrentMap actorManagers; + /** + * Decides whether to trust the activation id sent by Dapr. + */ + private final ActivationIdGate activationIdGate; + /** * The default constructor. This should not be called directly. * @@ -117,6 +122,7 @@ private ActorRuntime(ManagedChannel channel, DaprClient daprClient) throws Illeg this.actorManagers = new ConcurrentHashMap<>(); this.daprClient = daprClient; this.channel = channel; + this.activationIdGate = new ActivationIdGate(daprClient::getRuntimeVersion); } /** @@ -291,7 +297,8 @@ public Mono invoke( String actorTypeName, String actorId, String actorMethodName, byte[] payload, String activationId) { ActorId id = new ActorId(actorId); return Mono.fromSupplier(() -> this.getActorManager(actorTypeName)) - .flatMap(m -> m.activateActor(id, activationId) + .flatMap(m -> this.activationIdGate.filter(activationId) + .flatMap(a -> m.activateActor(id, a.orElse(null))) .flatMap(a -> m.invokeMethod((ActorManager.ActiveActor) a, actorMethodName, payload))); } @@ -322,7 +329,8 @@ public Mono invokeReminder( String actorTypeName, String actorId, String reminderName, byte[] params, String activationId) { ActorId id = new ActorId(actorId); return Mono.fromSupplier(() -> this.getActorManager(actorTypeName)) - .flatMap(m -> m.activateActor(id, activationId) + .flatMap(m -> this.activationIdGate.filter(activationId) + .flatMap(a -> m.activateActor(id, a.orElse(null))) .flatMap(a -> m.invokeReminder((ActorManager.ActiveActor) a, reminderName, params))); } @@ -353,7 +361,8 @@ public Mono invokeTimer( String actorTypeName, String actorId, String timerName, byte[] params, String activationId) { ActorId id = new ActorId(actorId); return Mono.fromSupplier(() -> this.getActorManager(actorTypeName)) - .flatMap(m -> m.activateActor(id, activationId) + .flatMap(m -> this.activationIdGate.filter(activationId) + .flatMap(a -> m.activateActor(id, a.orElse(null))) .flatMap(a -> m.invokeTimer((ActorManager.ActiveActor) a, timerName, params))); } diff --git a/sdk-actors/src/main/java/io/dapr/actors/runtime/DaprClient.java b/sdk-actors/src/main/java/io/dapr/actors/runtime/DaprClient.java index 6d53361c43..f800d9acb3 100644 --- a/sdk-actors/src/main/java/io/dapr/actors/runtime/DaprClient.java +++ b/sdk-actors/src/main/java/io/dapr/actors/runtime/DaprClient.java @@ -87,4 +87,11 @@ Mono registerReminder( * @return Asynchronous void result. */ Mono unregisterTimer(String actorType, String actorId, String timerName); + + /** + * Gets the Dapr runtime version. + * + * @return Asynchronous result with the runtime version. + */ + Mono getRuntimeVersion(); } diff --git a/sdk-actors/src/main/java/io/dapr/actors/runtime/DaprClientImpl.java b/sdk-actors/src/main/java/io/dapr/actors/runtime/DaprClientImpl.java index bbe6a0e80a..7010048f89 100644 --- a/sdk-actors/src/main/java/io/dapr/actors/runtime/DaprClientImpl.java +++ b/sdk-actors/src/main/java/io/dapr/actors/runtime/DaprClientImpl.java @@ -21,6 +21,7 @@ import io.dapr.exceptions.DaprException; import io.dapr.v1.DaprActorsProtos; import io.dapr.v1.DaprGrpc; +import io.dapr.v1.DaprMetadataProtos; import io.grpc.ManagedChannel; import io.grpc.stub.StreamObserver; import reactor.core.publisher.Mono; @@ -220,6 +221,16 @@ public Mono unregisterTimer(String actorType, String actorId, String timer return Mono.create(it -> client.unregisterActorTimer(req, createStreamObserver(it))).then().then(); } + /** + * {@inheritDoc} + */ + @Override + public Mono getRuntimeVersion() { + DaprMetadataProtos.GetMetadataRequest req = DaprMetadataProtos.GetMetadataRequest.newBuilder().build(); + return Mono.create(it -> client.getMetadata(req, createStreamObserver(it))) + .map(DaprMetadataProtos.GetMetadataResponse::getRuntimeVersion); + } + private StreamObserver createStreamObserver(MonoSink sink) { return new StreamObserver() { @Override diff --git a/sdk-actors/src/test/java/io/dapr/actors/runtime/ActivationIdGateTest.java b/sdk-actors/src/test/java/io/dapr/actors/runtime/ActivationIdGateTest.java new file mode 100644 index 0000000000..d6147aa138 --- /dev/null +++ b/sdk-actors/src/test/java/io/dapr/actors/runtime/ActivationIdGateTest.java @@ -0,0 +1,112 @@ +/* + * 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; + +import org.junit.jupiter.api.Test; +import org.junit.jupiter.params.ParameterizedTest; +import org.junit.jupiter.params.provider.ValueSource; +import reactor.core.publisher.Mono; +import reactor.core.publisher.Sinks; + +import java.util.ArrayList; +import java.util.List; +import java.util.Optional; +import java.util.concurrent.atomic.AtomicInteger; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertFalse; +import static org.junit.jupiter.api.Assertions.assertTrue; + +public class ActivationIdGateTest { + + @ParameterizedTest + @ValueSource(strings = {"1.19.1", "1.19.1-rc.1", "1.19.2", "1.20.0", "2.0.0", "1.19.1+build"}) + public void supportedVersions(String version) { + assertTrue(ActivationIdGate.supported(version)); + } + + @ParameterizedTest + @ValueSource(strings = {"1.19.0", "1.19.0-rc.2", "1.18.9", "0.99.99", "edge", "dev", "", "1.19", "v1.19.1", + "1.19.1.1", "99999999999.0.0"}) + public void unsupportedVersions(String version) { + assertFalse(ActivationIdGate.supported(version)); + } + + @Test + public void nullVersionUnsupported() { + assertFalse(ActivationIdGate.supported(null)); + } + + @Test + public void emptyIdSkipsLookup() { + AtomicInteger lookups = new AtomicInteger(); + ActivationIdGate gate = new ActivationIdGate(() -> { + lookups.incrementAndGet(); + return Mono.just("1.19.1"); + }); + + assertEquals(Optional.empty(), gate.filter(null).block()); + assertEquals(Optional.empty(), gate.filter("").block()); + assertEquals(0, lookups.get()); + } + + @Test + public void oldRuntimeIgnoresIdAndCachesResult() { + AtomicInteger lookups = new AtomicInteger(); + ActivationIdGate gate = new ActivationIdGate(() -> { + lookups.incrementAndGet(); + return Mono.just("1.19.0"); + }); + + assertEquals(Optional.empty(), gate.filter("a1").block()); + assertEquals(Optional.empty(), gate.filter("a2").block()); + assertEquals(1, lookups.get()); + } + + @Test + public void failedLookupIgnoresIdThenRetries() { + AtomicInteger lookups = new AtomicInteger(); + ActivationIdGate gate = new ActivationIdGate(() -> lookups.incrementAndGet() == 1 + ? Mono.error(new RuntimeException("unavailable")) + : Mono.just("1.19.1")); + + assertEquals(Optional.empty(), gate.filter("a1").block()); + assertEquals(Optional.of("a2"), gate.filter("a2").block()); + assertEquals(Optional.of("a3"), gate.filter("a3").block()); + assertEquals(2, lookups.get()); + } + + @Test + public void concurrentCallersShareOneLookup() { + AtomicInteger lookups = new AtomicInteger(); + Sinks.One version = Sinks.one(); + ActivationIdGate gate = new ActivationIdGate(() -> { + lookups.incrementAndGet(); + return version.asMono(); + }); + + List> results = new ArrayList<>(); + for (int i = 0; i < 10; i++) { + gate.filter("a" + i).subscribe(results::add); + } + assertEquals(0, results.size()); + version.tryEmitValue("1.19.1"); + + assertEquals(10, results.size()); + for (int i = 0; i < 10; i++) { + assertEquals(Optional.of("a" + i), results.get(i)); + } + assertEquals(1, lookups.get()); + } +} diff --git a/sdk-actors/src/test/java/io/dapr/actors/runtime/ActorRuntimeTest.java b/sdk-actors/src/test/java/io/dapr/actors/runtime/ActorRuntimeTest.java index c2e3dc8dc3..471ae70e71 100644 --- a/sdk-actors/src/test/java/io/dapr/actors/runtime/ActorRuntimeTest.java +++ b/sdk-actors/src/test/java/io/dapr/actors/runtime/ActorRuntimeTest.java @@ -256,6 +256,7 @@ public static void beforeAll() throws Exception { @BeforeEach public void setup() throws Exception { this.mockDaprClient = mock(DaprClient.class); + when(this.mockDaprClient.getRuntimeVersion()).thenReturn(Mono.just("1.19.1")); this.runtime = constructor.newInstance(null, this.mockDaprClient); } @@ -539,6 +540,40 @@ public void invokeWithNewActivationIdReplacesActor() throws Exception { Assertions.assertEquals(0, created.get(1).deactivations); } + @Test + public void activationIdIgnoredOnOlderRuntime() throws Exception { + when(this.mockDaprClient.getRuntimeVersion()).thenReturn(Mono.just("1.19.0")); + List created = registerActivationActor(); + String actorId = UUID.randomUUID().toString(); + + Assertions.assertEquals(1, invokeIncrement(actorId, "a1")); + Assertions.assertEquals(2, invokeIncrement(actorId, "a2")); + this.runtime.invokeReminder( + ACTIVATION_ACTOR_NAME, actorId, "myReminder", createReminderParams("hello"), "a3").block(); + this.runtime.invokeTimer( + ACTIVATION_ACTOR_NAME, actorId, "myTimer", createTimerParams("increment", "x"), "a4").block(); + + Assertions.assertEquals(1, created.size()); + Assertions.assertEquals(0, created.get(0).deactivations); + verify(this.mockDaprClient, times(1)).getRuntimeVersion(); + } + + @Test + public void activationIdIgnoredWhenVersionLookupFailsThenRetried() throws Exception { + when(this.mockDaprClient.getRuntimeVersion()) + .thenReturn(Mono.error(new RuntimeException("unavailable")), Mono.just("1.19.1")); + List created = registerActivationActor(); + String actorId = UUID.randomUUID().toString(); + + Assertions.assertEquals(1, invokeIncrement(actorId, "a1")); + Assertions.assertEquals(1, created.size()); + Assertions.assertEquals(1, invokeIncrement(actorId, "a2")); + Assertions.assertEquals(2, invokeIncrement(actorId, "a2")); + + Assertions.assertEquals(2, created.size()); + verify(this.mockDaprClient, times(2)).getRuntimeVersion(); + } + @Test public void invokeWithoutActivationIdKeepsActor() throws Exception { List created = registerActivationActor();