From a4e1158db2944d9a67ccc8cd319c1be07bd541c4 Mon Sep 17 00:00:00 2001 From: Javier Aliaga Date: Mon, 26 Oct 2026 17:58:23 +0100 Subject: [PATCH] fix: hold an actor's turn until a cancelled call finishes When a caller cancels an actor call while it runs, daprd releases the actor and sends the next call, but the SDK kept running the cancelled method, so two turns ran at once on the same actor. ActorManager now runs activation, method calls and deactivation through a per-actor turn lock. A turn keeps the actor until its code and the state save finish, even if its request is cancelled. Turns that are still waiting are dropped when cancelled. Fixes #1871 Signed-off-by: Javier Aliaga --- .../io/dapr/actors/runtime/ActorManager.java | 26 ++- .../io/dapr/actors/runtime/ActorTurnLock.java | 145 ++++++++++++ .../dapr/actors/runtime/ActorManagerTest.java | 122 ++++++++++ .../actors/runtime/ActorTurnLockTest.java | 217 ++++++++++++++++++ .../actors/AbstractActorCancelledCallIT.java | 69 ++++++ .../actors/ActorCancelledCallGrpcHostIT.java | 37 +++ .../dapr/it/actors/ActorCancelledCallIT.java | 39 ++++ .../java/io/dapr/it/actors/app/MyActor.java | 6 + .../io/dapr/it/actors/app/MyActorBase.java | 35 +++ .../dapr/it/containers/BaseContainerIT.java | 10 + 10 files changed, 703 insertions(+), 3 deletions(-) create mode 100644 sdk-actors/src/main/java/io/dapr/actors/runtime/ActorTurnLock.java create mode 100644 sdk-actors/src/test/java/io/dapr/actors/runtime/ActorTurnLockTest.java create mode 100644 sdk-tests/src/test/java/io/dapr/it/actors/AbstractActorCancelledCallIT.java create mode 100644 sdk-tests/src/test/java/io/dapr/it/actors/ActorCancelledCallGrpcHostIT.java create mode 100644 sdk-tests/src/test/java/io/dapr/it/actors/ActorCancelledCallIT.java diff --git a/sdk-actors/src/main/java/io/dapr/actors/runtime/ActorManager.java b/sdk-actors/src/main/java/io/dapr/actors/runtime/ActorManager.java index 303736cf57..53f6268046 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 @@ -50,6 +50,11 @@ class ActorManager { */ private final Map 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 { * @return Asynchronous void response. */ Mono 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 activateActor(ActorId actorId) { * @return Asynchronous void response. */ Mono 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 invokeMonoMethod(AbstractActor actor, Method method, Object * @return Asynchronous response for the returned object. */ private Mono invoke(ActorId actorId, ActorMethodContext context, Function> 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 Expected return type for the function call. + * @return Asynchronous response for the returned object. + */ + private Mono invokeInTurn(ActorId actorId, ActorMethodContext context, + Function> func) { try { AbstractActor actor = this.activeActors.getOrDefault(actorId, null); if (actor == null) { diff --git a/sdk-actors/src/main/java/io/dapr/actors/runtime/ActorTurnLock.java b/sdk-actors/src/main/java/io/dapr/actors/runtime/ActorTurnLock.java new file mode 100644 index 0000000000..8eb723faa3 --- /dev/null +++ b/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. + * + *

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. + * + *

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>> turns = new HashMap<>(); + + /** + * Runs a turn once the previous turns for the same actor have finished. + * + *

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. + * + *

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 Result type. + * @return Asynchronous result of the turn. + */ + Mono run(ActorId actorId, Supplier> body) { + return Mono.deferContextual(context -> { + Turn turn = new Turn<>(actorId, body, context); + boolean first; + synchronized (this) { + Deque> 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> 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> 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 { + + private final ActorId actorId; + + private final Supplier> body; + + private final ContextView context; + + private final CompletableFuture result = new CompletableFuture<>(); + + private T value; + + private Turn(ActorId actorId, Supplier> 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 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); + }); + } + } +} 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..b8be753ccf 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 @@ -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 receiveReminder(String reminderName, String state, Duration du } } + interface SlowActor { + Mono 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 release = Sinks.empty(); + + public SlowActorImpl(ActorRuntimeContext runtimeContext, ActorId id) { + super(runtimeContext, id); + CREATED.incrementAndGet(); + } + + @Override + public Mono hold() { + HOLDING.incrementAndGet(); + return release.asMono(); + } + + @Override + public int getPostCount() { + return POST_COUNT.get(); + } + + @Override + protected Mono 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 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 context = createContext(MyActorImpl.class); private ActorManager 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 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 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 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 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 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)); diff --git a/sdk-actors/src/test/java/io/dapr/actors/runtime/ActorTurnLockTest.java b/sdk-actors/src/test/java/io/dapr/actors/runtime/ActorTurnLockTest.java new file mode 100644 index 0000000000..a3ccfbf319 --- /dev/null +++ b/sdk-actors/src/test/java/io/dapr/actors/runtime/ActorTurnLockTest.java @@ -0,0 +1,217 @@ +/* + * 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 org.junit.jupiter.api.Test; +import reactor.core.Disposable; +import reactor.core.publisher.Mono; +import reactor.core.publisher.Sinks; +import reactor.util.context.Context; + +import java.time.Duration; +import java.util.List; +import java.util.concurrent.CompletableFuture; +import java.util.concurrent.CopyOnWriteArrayList; +import java.util.concurrent.CountDownLatch; +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.assertEquals; +import static org.junit.jupiter.api.Assertions.assertFalse; +import static org.junit.jupiter.api.Assertions.assertNull; +import static org.junit.jupiter.api.Assertions.assertThrows; +import static org.junit.jupiter.api.Assertions.assertTrue; + +public class ActorTurnLockTest { + + private static final Duration TIMEOUT = Duration.ofSeconds(5); + + private final ActorTurnLock lock = new ActorTurnLock(); + + @Test + public void runsTurnsForSameActorOneAtATime() { + ActorId actorId = new ActorId("1"); + AtomicInteger running = new AtomicInteger(); + AtomicInteger maxRunning = new AtomicInteger(); + + Mono turns = Mono.when( + lock.run(actorId, () -> track(running, maxRunning)), + lock.run(actorId, () -> track(running, maxRunning)), + lock.run(actorId, () -> track(running, maxRunning))); + + turns.block(TIMEOUT); + assertEquals(1, maxRunning.get()); + } + + @Test + public void runsTurnsForDifferentActorsAtTheSameTime() { + Sinks.Empty first = Sinks.empty(); + Mono firstTurn = lock.run(new ActorId("1"), first::asMono); + Disposable firstSubscription = firstTurn.subscribe(); + + assertEquals("done", lock.run(new ActorId("2"), () -> Mono.just("done")).block(TIMEOUT)); + + first.tryEmitEmpty(); + firstSubscription.dispose(); + } + + @Test + public void runsTurnsInArrivalOrder() { + ActorId actorId = new ActorId("1"); + Sinks.Empty first = Sinks.empty(); + List order = new CopyOnWriteArrayList<>(); + + Mono turns = Mono.when( + lock.run(actorId, () -> first.asMono().doOnTerminate(() -> order.add(1))), + lock.run(actorId, () -> Mono.fromRunnable(() -> order.add(2))), + lock.run(actorId, () -> Mono.fromRunnable(() -> order.add(3)))); + Disposable subscription = turns.subscribe(); + first.tryEmitEmpty(); + + waitFor(() -> order.size() == 3); + assertEquals(List.of(1, 2, 3), order); + subscription.dispose(); + } + + @Test + public void cancelledRunningTurnKeepsActorUntilItFinishes() throws Exception { + ActorId actorId = new ActorId("1"); + Sinks.Empty first = Sinks.empty(); + AtomicInteger firstFinished = new AtomicInteger(); + CountDownLatch firstStarted = new CountDownLatch(1); + + Disposable firstCaller = lock.run(actorId, () -> { + firstStarted.countDown(); + return first.asMono().doOnSuccess(v -> firstFinished.incrementAndGet()); + }).subscribe(); + assertTrue(firstStarted.await(5, TimeUnit.SECONDS)); + + // The caller of the running turn goes away, as when Dapr stops waiting for it. + firstCaller.dispose(); + + CompletableFuture second = lock.run(actorId, () -> Mono.fromSupplier(firstFinished::get)).toFuture(); + Thread.sleep(100); + assertFalse(second.isDone()); + + first.tryEmitEmpty(); + // The second turn saw the first one finished. + assertEquals(1, second.get(5, TimeUnit.SECONDS)); + } + + @Test + public void cancelledWaitingTurnNeverRuns() { + ActorId actorId = new ActorId("1"); + Sinks.Empty first = Sinks.empty(); + AtomicInteger secondRuns = new AtomicInteger(); + + Disposable firstCaller = lock.run(actorId, first::asMono).subscribe(); + Disposable secondCaller = lock.run(actorId, () -> Mono.fromRunnable(secondRuns::incrementAndGet)).subscribe(); + secondCaller.dispose(); + first.tryEmitEmpty(); + + assertEquals("third", lock.run(actorId, () -> Mono.just("third")).block(TIMEOUT)); + assertEquals(0, secondRuns.get()); + firstCaller.dispose(); + } + + @Test + public void failedTurnReleasesActor() { + ActorId actorId = new ActorId("1"); + + // A type of its own, so a timeout from block() cannot pass for the turn's error. + assertThrows(TurnFailure.class, + () -> lock.run(actorId, () -> Mono.error(new TurnFailure())).block(TIMEOUT)); + assertThrows(TurnFailure.class, () -> lock.run(actorId, () -> { + throw new TurnFailure(); + }).block(TIMEOUT)); + assertEquals("next", lock.run(actorId, () -> Mono.just("next")).block(TIMEOUT)); + } + + @Test + public void emptyTurnReleasesActor() { + ActorId actorId = new ActorId("1"); + + assertNull(lock.run(actorId, Mono::empty).block(TIMEOUT)); + assertEquals("next", lock.run(actorId, () -> Mono.just("next")).block(TIMEOUT)); + } + + @Test + public void turnDoesNotStartUntilSubscribed() { + AtomicInteger runs = new AtomicInteger(); + + lock.run(new ActorId("1"), () -> Mono.fromRunnable(runs::incrementAndGet)); + + assertEquals(0, runs.get()); + } + + @Test + public void runsLongQueueOfSynchronousTurns() { + ActorId actorId = new ActorId("1"); + Sinks.Empty first = Sinks.empty(); + AtomicInteger runs = new AtomicInteger(); + + Disposable firstCaller = lock.run(actorId, first::asMono).subscribe(); + Mono turns = Mono.when(IntStream.range(0, 10_000) + .mapToObj(i -> lock.run(actorId, () -> Mono.fromRunnable(runs::incrementAndGet))) + .toArray(Mono[]::new)); + Disposable subscription = turns.subscribe(); + first.tryEmitEmpty(); + + waitFor(() -> runs.get() == 10_000); + subscription.dispose(); + firstCaller.dispose(); + } + + @Test + public void turnSeesCallerContext() { + ActorId actorId = new ActorId("1"); + Sinks.Empty first = Sinks.empty(); + Mono readContext = Mono.deferContextual(context -> Mono.just(context.get("caller"))); + + CompletableFuture firstTurn = lock.run(actorId, () -> first.asMono().then(readContext)) + .contextWrite(Context.of("caller", "first")) + .toFuture(); + // Queued behind the first turn, so it is started by the lock rather than by its own subscriber. + CompletableFuture secondTurn = lock.run(actorId, () -> readContext) + .contextWrite(Context.of("caller", "second")) + .toFuture(); + first.tryEmitEmpty(); + + assertEquals("first", firstTurn.join()); + assertEquals("second", secondTurn.join()); + } + + private static class TurnFailure extends RuntimeException { + } + + private static Mono track(AtomicInteger running, AtomicInteger maxRunning) { + return Mono.fromRunnable(() -> maxRunning.accumulateAndGet(running.incrementAndGet(), Math::max)) + .then(Mono.delay(Duration.ofMillis(50))) + .doOnTerminate(running::decrementAndGet) + .then(); + } + + private static void waitFor(BooleanSupplier condition) { + long deadline = System.nanoTime() + TIMEOUT.toNanos(); + while (!condition.getAsBoolean()) { + if (System.nanoTime() > deadline) { + throw new AssertionError("timed out waiting for condition"); + } + Thread.onSpinWait(); + } + } +} diff --git a/sdk-tests/src/test/java/io/dapr/it/actors/AbstractActorCancelledCallIT.java b/sdk-tests/src/test/java/io/dapr/it/actors/AbstractActorCancelledCallIT.java new file mode 100644 index 0000000000..b8379f12f7 --- /dev/null +++ b/sdk-tests/src/test/java/io/dapr/it/actors/AbstractActorCancelledCallIT.java @@ -0,0 +1,69 @@ +/* + * Copyright 2026 The Dapr Authors + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * http://www.apache.org/licenses/LICENSE-2.0 + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and +limitations under the License. +*/ + +package io.dapr.it.actors; + +import io.dapr.actors.ActorId; +import io.dapr.actors.client.ActorClient; +import io.dapr.actors.client.ActorProxy; +import io.dapr.actors.client.ActorProxyBuilder; +import io.dapr.client.resiliency.ResiliencyOptions; +import io.dapr.exceptions.DaprException; +import io.dapr.it.containers.BaseContainerIT; +import io.dapr.testcontainers.DaprContainer; +import org.junit.jupiter.api.Test; + +import java.time.Duration; +import java.util.UUID; + +import static io.dapr.it.Retry.callWithRetry; +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertThrows; + +/** + * When a caller cancels an actor call while it runs, daprd releases the actor's turn and sends the next call, but + * the app keeps running the cancelled one. The SDK must hold the turn until the actor code finishes. + */ +abstract class AbstractActorCancelledCallIT extends BaseContainerIT { + + private static final int CALLS = 5; + + protected static DaprContainer dapr; + + @Test + public void cancelledCallsDoNotOverlap() throws Exception { + ActorId actorId = new ActorId(UUID.randomUUID().toString()); + // The timeout becomes a gRPC deadline, so daprd cancels each call 500ms in while holdTurn runs for 2s. + ActorClient cancellingClient = newActorClient(dapr, new ResiliencyOptions().setTimeout(Duration.ofMillis(500))); + ActorProxy cancelling = newProxy(cancellingClient, actorId); + ActorProxy proxy = newProxy(newActorClient(dapr), actorId); + + // Activate the actor first so every holdTurn call reaches the app quickly. + proxy.invokeMethod("getCompletedTurns", Integer.class).block(); + + // Each call starts after daprd released the previous one, so without a turn lock in the app they overlap. + for (int i = 0; i < CALLS; i++) { + // Every call must be cancelled, otherwise the calls run one after another and the test proves nothing. + DaprException e = assertThrows(DaprException.class, () -> cancelling.invokeMethod("holdTurn").block()); + assertEquals("DEADLINE_EXCEEDED", e.getErrorCode()); + Thread.sleep(100); + } + + callWithRetry(() -> assertEquals(CALLS, proxy.invokeMethod("getCompletedTurns", Integer.class).block()), 30000); + assertEquals(1, proxy.invokeMethod("getMaxConcurrentTurns", Integer.class).block()); + } + + private static ActorProxy newProxy(ActorClient client, ActorId actorId) { + return new ActorProxyBuilder<>("MyActorTest", ActorProxy.class, client).build(actorId); + } +} diff --git a/sdk-tests/src/test/java/io/dapr/it/actors/ActorCancelledCallGrpcHostIT.java b/sdk-tests/src/test/java/io/dapr/it/actors/ActorCancelledCallGrpcHostIT.java new file mode 100644 index 0000000000..23f2673387 --- /dev/null +++ b/sdk-tests/src/test/java/io/dapr/it/actors/ActorCancelledCallGrpcHostIT.java @@ -0,0 +1,37 @@ +/* + * Copyright 2026 The Dapr Authors + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * http://www.apache.org/licenses/LICENSE-2.0 + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and +limitations under the License. +*/ + +package io.dapr.it.actors; + +import io.dapr.it.actors.app.MyActorGrpcService; +import io.dapr.testcontainers.DaprProtocol; +import org.junit.jupiter.api.BeforeAll; + +/** + * Cancelled actor calls on an app that hosts actors over the Dapr actor event stream. + */ +public class ActorCancelledCallGrpcHostIT extends AbstractActorCancelledCallIT { + + private static final String APP_NAME = "actor-cancelled-call-grpc-host-it"; + + @BeforeAll + public static void start() throws Exception { + dapr = startActorHostAndAttach( + APP_NAME, + MyActorGrpcService.class, + () -> daprBuilder(APP_NAME) + .withAppProtocol(DaprProtocol.GRPC) + .withComponent(redisStateStore(STATE_STORE_NAME))).dapr(); + waitForActorsReady(dapr); + } +} diff --git a/sdk-tests/src/test/java/io/dapr/it/actors/ActorCancelledCallIT.java b/sdk-tests/src/test/java/io/dapr/it/actors/ActorCancelledCallIT.java new file mode 100644 index 0000000000..45adda8464 --- /dev/null +++ b/sdk-tests/src/test/java/io/dapr/it/actors/ActorCancelledCallIT.java @@ -0,0 +1,39 @@ +/* + * Copyright 2026 The Dapr Authors + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * http://www.apache.org/licenses/LICENSE-2.0 + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and +limitations under the License. +*/ + +package io.dapr.it.actors; + +import io.dapr.it.AppRun; +import io.dapr.it.actors.app.MyActorService; +import org.junit.jupiter.api.BeforeAll; + +/** + * Cancelled actor calls on an app that hosts actors over HTTP. + */ +public class ActorCancelledCallIT extends AbstractActorCancelledCallIT { + + private static final String APP_NAME = "actor-cancelled-call-it"; + + @BeforeAll + public static void start() throws Exception { + dapr = startAppAndAttach( + APP_NAME, + MyActorService.class, + AppRun.AppProtocol.HTTP, + appPort -> daprBuilder(APP_NAME) + .withAppPort(appPort) + .withAppChannelAddress("host.testcontainers.internal") + .withComponent(redisStateStore(STATE_STORE_NAME))).dapr(); + waitForActorsReady(dapr); + } +} diff --git a/sdk-tests/src/test/java/io/dapr/it/actors/app/MyActor.java b/sdk-tests/src/test/java/io/dapr/it/actors/app/MyActor.java index 4d1d1965de..155011f401 100644 --- a/sdk-tests/src/test/java/io/dapr/it/actors/app/MyActor.java +++ b/sdk-tests/src/test/java/io/dapr/it/actors/app/MyActor.java @@ -43,4 +43,10 @@ public interface MyActor { @ActorMethod(name = "DotNetMethodAsync") boolean dotNetMethod(); + + void holdTurn(); + + int getMaxConcurrentTurns(); + + int getCompletedTurns(); } \ No newline at end of file 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..fb971bd573 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 @@ -31,6 +31,8 @@ import java.util.List; import java.util.Objects; import java.util.TimeZone; +import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.atomic.AtomicInteger; import java.util.function.Function; public abstract class MyActorBase extends AbstractActor implements MyActor, Remindable { @@ -38,6 +40,9 @@ public abstract class MyActorBase extends AbstractActor implements MyActor, R public static final List ACTIVE_ACTOR = new ArrayList<>(); + // Per actor id: turns of holdTurn in progress, the most seen at once, and how many finished. + private static final ConcurrentHashMap TURNS = new ConcurrentHashMap<>(); + // this tracks method entries and is used to validate turn-based concurrency. public final ArrayList callLog = new ArrayList<>(); @@ -255,6 +260,36 @@ public boolean dotNetMethod() { return true; } + @Override + public void holdTurn() { + AtomicInteger[] turns = turns(); + int inProgress = turns[0].incrementAndGet(); + turns[1].accumulateAndGet(inProgress, Math::max); + try { + Thread.sleep(2000); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + } finally { + turns[0].decrementAndGet(); + turns[2].incrementAndGet(); + } + } + + @Override + public int getMaxConcurrentTurns() { + return turns()[1].get(); + } + + @Override + public int getCompletedTurns() { + return turns()[2].get(); + } + + private AtomicInteger[] turns() { + return TURNS.computeIfAbsent(super.getId().toString(), + id -> new AtomicInteger[] {new AtomicInteger(), new AtomicInteger(), new AtomicInteger()}); + } + private void formatAndLog(boolean isEnter, String methodName) { this.formatAndLog(isEnter, methodName, null); } diff --git a/sdk-tests/src/test/java/io/dapr/it/containers/BaseContainerIT.java b/sdk-tests/src/test/java/io/dapr/it/containers/BaseContainerIT.java index a6642012b9..3298a0c0c5 100644 --- a/sdk-tests/src/test/java/io/dapr/it/containers/BaseContainerIT.java +++ b/sdk-tests/src/test/java/io/dapr/it/containers/BaseContainerIT.java @@ -16,6 +16,7 @@ import io.dapr.actors.client.ActorClient; import io.dapr.client.DaprClient; import io.dapr.client.DaprClientBuilder; +import io.dapr.client.resiliency.ResiliencyOptions; import io.dapr.config.Properties; import io.dapr.config.Property; import io.dapr.it.AppRun; @@ -309,6 +310,15 @@ protected static ActorClient newActorClient(DaprContainer dapr) { return client; } + /** + * ActorClient overload with resiliency options, such as a timeout that cancels calls in daprd. + */ + protected static ActorClient newActorClient(DaprContainer dapr, ResiliencyOptions resiliencyOptions) { + ActorClient client = new ActorClient(new Properties(daprOverrides(dapr)), resiliencyOptions); + deferClose(client); + return client; + } + /** * ActorClient overload that injects HTTP headers (metadata) on actor calls. * Used by ITs that need to override request-level headers like Content-Length.