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.