fix: hold an actor's turn until a cancelled call finishes

sdk-actors/src/main/java/io/dapr/actors/runtime/ActorManager.java

@@ -50,6 +50,11 @@ class ActorManager<T extends AbstractActor> {
    */
   private final Map<ActorId, T> activeActors;
 
+  /**
+   * Runs one call at a time per Actor.
+   */
+  private final ActorTurnLock turnLock = new ActorTurnLock();
+
   /**
    * Instantiates a new manager for a given actor referenced in the runtimeContext.
    *
@@ -68,13 +73,13 @@ class ActorManager<T extends AbstractActor> {
    * @return Asynchronous void response.
    */
   Mono<Void> activateActor(ActorId actorId) {
-    return Mono.fromSupplier(() -> {
+    return this.turnLock.run(actorId, () -> Mono.fromSupplier(() -> {
       if (this.activeActors.containsKey(actorId)) {
         return null;
       }
 
       return this.runtimeContext.getActorFactory().createActor(runtimeContext, actorId);
-    }).flatMap(actor -> actor.onActivateInternal().then(this.onActivatedActor(actorId, actor)));
+    }).flatMap(actor -> actor.onActivateInternal().then(this.onActivatedActor(actorId, actor))));
   }
 
   /**
@@ -84,7 +89,8 @@ Mono<Void> activateActor(ActorId actorId) {
    * @return Asynchronous void response.
    */
   Mono<Void> deactivateActor(ActorId actorId) {
-    return Mono.fromSupplier(() -> this.activeActors.remove(actorId)).flatMap(actor -> actor.onDeactivateInternal());
+    return this.turnLock.run(actorId, () -> Mono.fromSupplier(() -> this.activeActors.remove(actorId))
+        .flatMap(actor -> actor.onDeactivateInternal()));
   }
 
   /**
@@ -307,6 +313,20 @@ private Mono<Object> invokeMonoMethod(AbstractActor actor, Method method, Object
    * @return Asynchronous response for the returned object.
    */
   private <T> Mono<T> invoke(ActorId actorId, ActorMethodContext context, Function<AbstractActor, Mono<T>> func) {
+    return this.turnLock.run(actorId, () -> this.invokeInTurn(actorId, context, func));
+  }
+
+  /**
+   * Invokes a method, timer or reminder for an Actor while holding the Actor's turn.
+   *
+   * @param actorId Actor identifier.
+   * @param context Context for the method/timer/reminder call.
+   * @param func    Function to perform the method call.
+   * @param <T>     Expected return type for the function call.
+   * @return Asynchronous response for the returned object.
+   */
+  private <T> Mono<T> invokeInTurn(ActorId actorId, ActorMethodContext context,
+      Function<AbstractActor, Mono<T>> func) {
     try {
       AbstractActor actor = this.activeActors.getOrDefault(actorId, null);
       if (actor == null) {

sdk-actors/src/main/java/io/dapr/actors/runtime/ActorTurnLock.java

@@ -0,0 +1,145 @@
+/*
+ * 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 io.dapr.actors.ActorId;
+import reactor.core.publisher.Mono;
+import reactor.core.scheduler.Schedulers;
+import reactor.util.context.ContextView;
+
+import java.util.ArrayDeque;
+import java.util.Deque;
+import java.util.HashMap;
+import java.util.Map;
+import java.util.concurrent.CompletableFuture;
+import java.util.function.Supplier;
+
+/**
+ * Runs at most one turn at a time per actor, in arrival order.
+ *
+ * <p>Dapr can stop waiting before the actor code finishes, for example when the caller cancels, and then sends the
+ * next call for the same actor. A turn therefore keeps the actor until its code has finished, not until its request
+ * is cancelled.
+ *
+ * <p>There is no timeout: a turn that never finishes keeps its actor, and every later call for that actor waits.
+ * Reentrancy is not supported, so a call that needs the actor's own turn to finish first waits forever.
+ */
+final class ActorTurnLock {
+
+  /**
+   * Turns per actor, in arrival order. The first one is running.
+   */
+  private final Map<ActorId, Deque<Turn<?>>> turns = new HashMap<>();
+
+  /**
+   * Runs a turn once the previous turns for the same actor have finished.
+   *
+   * <p>Cancelling the returned Mono while the turn waits removes it. Once the turn is next in line or has started,
+   * cancelling only stops waiting for its result: the turn still runs to the end and keeps the actor until then.
+   *
+   * <p>The turn's work sees the Reactor context of the subscriber that queued it.
+   *
+   * @param actorId Actor identifier.
+   * @param body    Supplies the turn's work. Called when the turn starts.
+   * @param <T>     Result type.
+   * @return Asynchronous result of the turn.
+   */
+  <T> Mono<T> run(ActorId actorId, Supplier<Mono<T>> body) {
+    return Mono.deferContextual(context -> {
+      Turn<T> turn = new Turn<>(actorId, body, context);
+      boolean first;
+      synchronized (this) {
+        Deque<Turn<?>> queue = this.turns.computeIfAbsent(actorId, id -> new ArrayDeque<>());
+        queue.addLast(turn);
+        first = queue.size() == 1;
+      }
+
+      if (first) {
+        turn.start(false);
+      }
+
+      return Mono.fromFuture(turn.result, true).doOnCancel(() -> this.withdraw(turn));
+    });
+  }
+
+  /**
+   * Removes a turn that has not started. A started turn is left to finish.
+   */
+  private synchronized void withdraw(Turn<?> turn) {
+    Deque<Turn<?>> queue = this.turns.get(turn.actorId);
+    if (queue != null && queue.peekFirst() != turn) {
+      queue.remove(turn);
+    }
+  }
+
+  /**
+   * Ends the running turn and starts the next one, if any.
+   */
+  private void finish(Turn<?> turn) {
+    Turn<?> next;
+    synchronized (this) {
+      Deque<Turn<?>> queue = this.turns.get(turn.actorId);
+      queue.removeFirst();
+      next = queue.peekFirst();
+      if (next == null) {
+        this.turns.remove(turn.actorId);
+      }
+    }
+
+    if (next != null) {
+      next.start(true);
+    }
+  }
+
+  private final class Turn<T> {
+
+    private final ActorId actorId;
+
+    private final Supplier<Mono<T>> body;
+
+    private final ContextView context;
+
+    private final CompletableFuture<T> result = new CompletableFuture<>();
+
+    private T value;
+
+    private Turn(ActorId actorId, Supplier<Mono<T>> body, ContextView context) {
+      this.actorId = actorId;
+      this.body = body;
+      this.context = context;
+    }
+
+    /**
+     * Starts the turn. A turn handed over by the previous one runs on another thread, so a long queue of turns
+     * that finish synchronously does not grow the stack.
+     */
+    private void start(boolean handedOver) {
+      Mono<T> work = Mono.defer(this.body);
+      if (handedOver) {
+        work = work.subscribeOn(Schedulers.boundedElastic());
+      }
+
+      work.contextWrite(this.context).subscribe(
+          v -> this.value = v,
+          e -> {
+            finish(this);
+            this.result.completeExceptionally(e);
+          },
+          () -> {
+            finish(this);
+            this.result.complete(this.value);
+          });
+    }
+  }
+}

sdk-actors/src/test/java/io/dapr/actors/runtime/ActorManagerTest.java

@@ -19,11 +19,18 @@
 import io.dapr.utils.TypeRef;
 import org.junit.jupiter.api.Assertions;
 import org.junit.jupiter.api.Test;
+import reactor.core.Disposable;
 import reactor.core.publisher.Mono;
+import reactor.core.publisher.Sinks;
+import reactor.core.scheduler.Schedulers;
 
 import java.io.IOException;
 import java.time.Duration;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicInteger;
+import java.util.function.BooleanSupplier;
+import java.util.stream.IntStream;
 
 import static org.junit.jupiter.api.Assertions.assertThrows;
 import static org.mockito.ArgumentMatchers.any;
@@ -116,6 +123,57 @@ public Mono<Void> receiveReminder(String reminderName, String state, Duration du
     }
   }
 
+  interface SlowActor {
+    Mono<Void> hold();
+
+    int getPostCount();
+  }
+
+  public static class SlowActorImpl extends AbstractActor implements SlowActor {
+
+    static final AtomicInteger CREATED = new AtomicInteger();
+
+    static final AtomicInteger POST_COUNT = new AtomicInteger();
+
+    static final AtomicInteger HOLDING = new AtomicInteger();
+
+    static volatile Sinks.Empty<Void> release = Sinks.empty();
+
+    public SlowActorImpl(ActorRuntimeContext runtimeContext, ActorId id) {
+      super(runtimeContext, id);
+      CREATED.incrementAndGet();
+    }
+
+    @Override
+    public Mono<Void> hold() {
+      HOLDING.incrementAndGet();
+      return release.asMono();
+    }
+
+    @Override
+    public int getPostCount() {
+      return POST_COUNT.get();
+    }
+
+    @Override
+    protected Mono<Void> onActivate() {
+      // Widens the window in which two activations of the same actor could both create it.
+      return Mono.delay(Duration.ofMillis(50)).then();
+    }
+
+    @Override
+    protected Mono<Void> onPostActorMethod(ActorMethodContext actorMethodContext) {
+      return Mono.fromRunnable(POST_COUNT::incrementAndGet);
+    }
+
+    static void reset() {
+      CREATED.set(0);
+      POST_COUNT.set(0);
+      HOLDING.set(0);
+      release = Sinks.empty();
+    }
+  }
+
   private ActorRuntimeContext<MyActorImpl> context = createContext(MyActorImpl.class);
 
   private ActorManager<MyActorImpl> manager = new ActorManager<>(context);
@@ -279,6 +337,70 @@ public void activateInvokeTimerDeactivateThenInvokeTimer() throws IOException {
     assertThrows(IllegalArgumentException.class, () -> this.manager.invokeTimer(actorId, "count", createTimerParams("incrementCount", 2)).block());
   }
 
+  @Test
+  public void cancelledCallKeepsTurnThroughPostMethod() throws Exception {
+    SlowActorImpl.reset();
+    ActorManager<SlowActorImpl> manager = new ActorManager<>(createContext(SlowActorImpl.class));
+    ActorId actorId = newActorId();
+    manager.activateActor(actorId).block();
+
+    Disposable first = manager.invokeMethod(actorId, "hold", null).subscribe();
+    waitFor(() -> SlowActorImpl.HOLDING.get() == 1);
+    // Dapr stops waiting for the call, as when its caller cancels.
+    first.dispose();
+
+    CompletableFuture<byte[]> second = manager.invokeMethod(actorId, "getPostCount", null).toFuture();
+    Thread.sleep(100);
+    Assertions.assertFalse(second.isDone());
+
+    SlowActorImpl.release.tryEmitEmpty();
+    // The second call ran after the first call's post-method step.
+    Assertions.assertEquals("1", new String(second.get(5, TimeUnit.SECONDS)));
+  }
+
+  @Test
+  public void concurrentActivationsCreateOneActor() {
+    SlowActorImpl.reset();
+    ActorManager<SlowActorImpl> manager = new ActorManager<>(createContext(SlowActorImpl.class));
+    ActorId actorId = newActorId();
+
+    Mono.when(IntStream.range(0, 10)
+        .mapToObj(i -> manager.activateActor(actorId).subscribeOn(Schedulers.boundedElastic()))
+        .toArray(Mono[]::new))
+        .block(Duration.ofSeconds(5));
+
+    Assertions.assertEquals(1, SlowActorImpl.CREATED.get());
+  }
+
+  @Test
+  public void deactivationWaitsForRunningCall() throws Exception {
+    SlowActorImpl.reset();
+    ActorManager<SlowActorImpl> manager = new ActorManager<>(createContext(SlowActorImpl.class));
+    ActorId actorId = newActorId();
+    manager.activateActor(actorId).block();
+
+    Disposable call = manager.invokeMethod(actorId, "hold", null).subscribe();
+    waitFor(() -> SlowActorImpl.HOLDING.get() == 1);
+    CompletableFuture<Void> deactivation = manager.deactivateActor(actorId).toFuture();
+    Thread.sleep(100);
+    Assertions.assertFalse(deactivation.isDone());
+
+    SlowActorImpl.release.tryEmitEmpty();
+    deactivation.get(5, TimeUnit.SECONDS);
+    Assertions.assertEquals(1, SlowActorImpl.POST_COUNT.get());
+    call.dispose();
+  }
+
+  private static void waitFor(BooleanSupplier condition) throws InterruptedException {
+    long deadline = System.nanoTime() + TimeUnit.SECONDS.toNanos(5);
+    while (!condition.getAsBoolean()) {
+      if (System.nanoTime() > deadline) {
+        throw new AssertionError("timed out waiting for condition");
+      }
+      Thread.sleep(10);
+    }
+  }
+
   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));