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..8cd30f4999 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 @@ -335,6 +335,13 @@ void resetState() { this.actorStateManager.clear(); } + /** + * Refuses any later state save from this instance once a newer activation replaced it. + */ + void discard() { + this.actorStateManager.discard(); + } + /** * Internal callback when an Actor is activated. * @@ -383,15 +390,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/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/ActorEventDispatcher.java b/sdk-actors/src/main/java/io/dapr/actors/runtime/ActorEventDispatcher.java index b0c6298c31..c10cca5f79 100644 --- a/sdk-actors/src/main/java/io/dapr/actors/runtime/ActorEventDispatcher.java +++ b/sdk-actors/src/main/java/io/dapr/actors/runtime/ActorEventDispatcher.java @@ -19,6 +19,7 @@ import com.google.protobuf.ByteString; import com.google.protobuf.BytesValue; import com.google.protobuf.InvalidProtocolBufferException; +import io.dapr.client.Headers; import io.dapr.v1.DaprActorsProtos.ActorEntityConfig; import io.dapr.v1.DaprActorsProtos.SubscribeActorEventsRequestAlpha1; import io.dapr.v1.DaprActorsProtos.SubscribeActorEventsRequestDeactivateResponseAlpha1; @@ -38,6 +39,7 @@ import java.io.IOException; import java.nio.charset.StandardCharsets; import java.time.Duration; +import java.util.Map; /** * Maps messages received on the actor event stream to {@link ActorRuntime} calls, and their results back to @@ -148,7 +150,8 @@ private static Mono invoke( byte[] payload = request.getData().isEmpty() ? null : request.getData().toByteArray(); String contentType = runtime.getObjectSerializer(request.getActorType()).getContentType(); - return runtime.invoke(request.getActorType(), request.getActorId(), request.getMethod(), payload) + return runtime.invoke(request.getActorType(), request.getActorId(), request.getMethod(), payload, + activationId(request.getMetadataMap())) .map(ByteString::copyFrom) .defaultIfEmpty(ByteString.EMPTY) .map(data -> SubscribeActorEventsRequestAlpha1.newBuilder() @@ -169,7 +172,8 @@ private static Mono reminder( return Mono.fromCallable(() -> buildCallbackBody( null, request.getDueTime(), request.getPeriod(), request.hasData() ? request.getData() : null)) - .flatMap(body -> runtime.invokeReminder(request.getActorType(), request.getActorId(), request.getName(), body)) + .flatMap(body -> runtime.invokeReminder(request.getActorType(), request.getActorId(), request.getName(), body, + activationId(request.getMetadataMap()))) .then(Mono.fromSupplier(() -> SubscribeActorEventsRequestAlpha1.newBuilder() .setReminderResponse(SubscribeActorEventsRequestReminderResponseAlpha1.newBuilder().setId(id)) .build())) @@ -186,7 +190,8 @@ private static Mono timer( return Mono.fromCallable(() -> buildCallbackBody( request.getCallback(), request.getDueTime(), request.getPeriod(), request.hasData() ? request.getData() : null)) - .flatMap(body -> runtime.invokeTimer(request.getActorType(), request.getActorId(), request.getName(), body)) + .flatMap(body -> runtime.invokeTimer(request.getActorType(), request.getActorId(), request.getName(), body, + activationId(request.getMetadataMap()))) .then(Mono.fromSupplier(() -> SubscribeActorEventsRequestAlpha1.newBuilder() .setTimerResponse(SubscribeActorEventsRequestReminderResponseAlpha1.newBuilder().setId(id)) .build())) @@ -249,6 +254,15 @@ private static boolean isRegistered(ActorRuntime runtime, String actorType) { return runtime.getConfig().getRegisteredActorTypes().contains(actorType); } + private static String activationId(Map metadata) { + for (Map.Entry entry : metadata.entrySet()) { + if (Headers.DAPR_ACTOR_ACTIVATION_ID.equalsIgnoreCase(entry.getKey()) && !entry.getValue().isEmpty()) { + return entry.getValue(); + } + } + return null; + } + private static SubscribeActorEventsRequestAlpha1 notRegistered(String id, String actorType) { return failed(id, Status.Code.NOT_FOUND, String.format("Actor type %s is not registered with Actor runtime.", actorType)); 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 53f6268046..8599e3ec0a 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 @@ -48,7 +48,7 @@ class ActorManager { /** * Active Actor instances. */ - private final Map activeActors; + private final Map> activeActors; /** * Runs one call at a time per Actor. @@ -73,13 +73,28 @@ class ActorManager { * @return Asynchronous void response. */ Mono activateActor(ActorId actorId) { - return this.turnLock.run(actorId, () -> 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) { + if (activationId == null || activationId.isEmpty() || activationId.equals(current.activationId)) { + return Mono.just(current); + } + + // Dropped before waiting on the turn, so neither a call still running on it nor its onDeactivate can save. + if (this.activeActors.remove(actorId, current)) { + current.actor.discard(); + return this.turnLock.run(actorId, () -> this.deactivateDiscarded(actorId, current.actor) + .then(this.createActor(actorId, activationId))); + } } - return this.runtimeContext.getActorFactory().createActor(runtimeContext, actorId); - }).flatMap(actor -> actor.onActivateInternal().then(this.onActivatedActor(actorId, actor)))); + return this.turnLock.run(actorId, () -> this.createActor(actorId, activationId)); + }); } /** @@ -89,8 +104,15 @@ Mono activateActor(ActorId actorId) { * @return Asynchronous void response. */ Mono deactivateActor(ActorId actorId) { - return this.turnLock.run(actorId, () -> Mono.fromSupplier(() -> this.activeActors.remove(actorId)) - .flatMap(actor -> actor.onDeactivateInternal())); + return Mono.defer(() -> { + ActiveActor active = this.activeActors.remove(actorId); + if (active == null) { + return Mono.empty(); + } + + // Removed before waiting on the turn, so a call still running on it skips its save. + return this.turnLock.run(actorId, active.actor::onDeactivateInternal); + }); } /** @@ -111,6 +133,14 @@ DaprObjectSerializer getObjectSerializer() { * @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; @@ -121,10 +151,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(); } /** @@ -136,29 +166,69 @@ 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.defer(() -> { + ActiveActor registered = this.activeActors.get(actorId); + if (registered != null) { + return this.joinRegistered(actorId, registered, activationId); + } + + T actor = this.runtimeContext.getActorFactory().createActor(runtimeContext, actorId); + ActiveActor active = new ActiveActor<>(actor, activationId); + return actor.onActivateInternal().then(Mono.fromSupplier(() -> { + this.activeActors.put(actorId, active); + return active; + })); + }); + } + + /** A queued activation with a different id fails, and Dapr retries it. */ + private Mono> joinRegistered(ActorId actorId, ActiveActor registered, String activationId) { + if (activationId == null || activationId.isEmpty() || activationId.equals(registered.activationId)) { + return Mono.just(registered); + } + + return 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(); + }); } /** @@ -210,25 +280,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); @@ -306,36 +381,29 @@ 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) { - return this.turnLock.run(actorId, () -> this.invokeInTurn(actorId, context, func)); + private Mono invoke(ActiveActor active, ActorMethodContext context, Function> func) { + return this.turnLock.run(active.actor.getId(), () -> this.invokeInTurn(active, context, func)); } /** * Invokes a method, timer or reminder for an Actor while holding the Actor's turn. * - * @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 invokeInTurn(ActorId actorId, ActorMethodContext context, - Function> func) { + private Mono invokeInTurn(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); @@ -343,12 +411,52 @@ private Mono invokeInTurn(ActorId actorId, ActorMethodContext context, }) .then(actor.onPreActorMethodInternal(context)) .then((Mono) func.apply(actor)) - .switchIfEmpty( - actor.onPostActorMethodInternal(context)) - .flatMap(r -> actor.onPostActorMethodInternal(context).thenReturn(r)) - .map(o -> (T) o); + .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); } } + + 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); + }); + } + + /** 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) { + 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 Mono.empty(); + }); + } + + 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 4ce332575e..ea6e6478e2 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 @@ -71,6 +71,11 @@ public class ActorRuntime implements Closeable { */ private final ActorRuntimeConfig config; + /** + * Decides whether to trust the activation id sent by Dapr. + */ + private final ActivationIdGate activationIdGate; + /** * A client used to communicate from the actor to the Dapr runtime. */ @@ -136,6 +141,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); } /** @@ -348,10 +354,26 @@ 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 -> this.activationIdGate.filter(activationId) + .flatMap(a -> m.activateActor(id, a.orElse(null))) + .flatMap(a -> m.invokeMethod((ActorManager.ActiveActor) a, actorMethodName, payload))); } /** @@ -364,10 +386,26 @@ 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 -> this.activationIdGate.filter(activationId) + .flatMap(a -> m.activateActor(id, a.orElse(null))) + .flatMap(a -> m.invokeReminder((ActorManager.ActiveActor) a, reminderName, params))); } /** @@ -380,10 +418,26 @@ 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 -> 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/ActorStateManager.java b/sdk-actors/src/main/java/io/dapr/actors/runtime/ActorStateManager.java index 1bf5430dd8..e530cde763 100644 --- a/sdk-actors/src/main/java/io/dapr/actors/runtime/ActorStateManager.java +++ b/sdk-actors/src/main/java/io/dapr/actors/runtime/ActorStateManager.java @@ -1,323 +1,341 @@ -/* - * Copyright 2021 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 io.dapr.utils.TypeRef; -import reactor.core.publisher.Mono; - -import java.util.ArrayList; -import java.util.List; -import java.util.Map; -import java.util.NoSuchElementException; -import java.util.concurrent.ConcurrentHashMap; - -/** - * Manages state changes of a given Actor instance. - * All changes are cached in-memory until save() is called. - */ -public class ActorStateManager { - - /** - * Provides states using a state store. - */ - private final DaprStateAsyncProvider stateProvider; - - /** - * Name of the Actor's type. - */ - private final String actorTypeName; - - /** - * Actor's identifier. - */ - private final ActorId actorId; - - /** - * Cache of state changes in this Actor's instance. - */ - private final Map stateChangeTracker; - - /** - * Instantiates a new state manager for the given Actor's instance. - * - * @param stateProvider State store provider. - * @param actorTypeName Name of Actor's type. - * @param actorId Actor's identifier. - */ - ActorStateManager(DaprStateAsyncProvider stateProvider, String actorTypeName, ActorId actorId) { - this.stateProvider = stateProvider; - this.actorTypeName = actorTypeName; - this.actorId = actorId; - this.stateChangeTracker = new ConcurrentHashMap<>(); - } - - /** - * Adds a given key/value to the Actor's state store's cache. - * - * @param stateName Name of the state being added. - * @param value Value to be added. - * @param Type of the object being added. - * @return Asynchronous void operation. - */ - public Mono add(String stateName, T value) { - return Mono.fromSupplier(() -> { - if (stateName == null) { - throw new IllegalArgumentException("State's name cannot be null."); - } - - return null; - }).then(this.stateProvider.contains(this.actorTypeName, this.actorId, stateName) - .map(exists -> { - if (this.stateChangeTracker.containsKey(stateName)) { - StateChangeMetadata metadata = this.stateChangeTracker.get(stateName); - - if (metadata.kind == ActorStateChangeKind.REMOVE) { - this.stateChangeTracker.put(stateName, new StateChangeMetadata(ActorStateChangeKind.UPDATE, value)); - return true; - } - - throw new IllegalStateException("Duplicate cached state: " + stateName); - } - - if (exists) { - throw new IllegalStateException("Duplicate state: " + stateName); - } - - this.stateChangeTracker.put(stateName, new StateChangeMetadata(ActorStateChangeKind.ADD, value)); - return true; - })) - .then(); - } - - /** - * Fetches the most recent value for the given state, including cached value. - * - * @param stateName Name of the state. - * @param clazz Class type for the value being fetched. - * @param Type being fetched. - * @return Asynchronous response with fetched object. - */ - public Mono get(String stateName, Class clazz) { - return this.get(stateName, TypeRef.get(clazz)); - } - - /** - * Fetches the most recent value for the given state, including cached value. - * - * @param stateName Name of the state. - * @param type Class type for the value being fetched. - * @param Type being fetched. - * @return Asynchronous response with fetched object. - */ - public Mono get(String stateName, TypeRef type) { - return Mono.fromSupplier(() -> { - if (stateName == null) { - throw new IllegalArgumentException("State's name cannot be null."); - } - - if (this.stateChangeTracker.containsKey(stateName)) { - StateChangeMetadata metadata = this.stateChangeTracker.get(stateName); - - if (metadata.kind == ActorStateChangeKind.REMOVE) { - throw new NoSuchElementException("State is marked for removal: " + stateName); - } - - return (T) metadata.value; - } - - return (T) null; - }).switchIfEmpty( - this.stateProvider.load(this.actorTypeName, this.actorId, stateName, type) - .switchIfEmpty(Mono.error(new NoSuchElementException("State not found: " + stateName))) - .map(v -> { - this.stateChangeTracker.put(stateName, new StateChangeMetadata(ActorStateChangeKind.NONE, v)); - return (T) v; - })); - } - - /** - * Updates a given key/value pair in the state store's cache. - * - * @param stateName Name of the state being updated. - * @param value Value to be set for given state. - * @param Type of the value being set. - * @return Asynchronous void result. - */ - public Mono set(String stateName, T value) { - return Mono.fromSupplier(() -> { - if (stateName == null) { - throw new IllegalArgumentException("State's name cannot be null."); - } - - if (this.stateChangeTracker.containsKey(stateName)) { - StateChangeMetadata metadata = this.stateChangeTracker.get(stateName); - - ActorStateChangeKind kind = metadata.kind; - if ((kind == ActorStateChangeKind.NONE) || (kind == ActorStateChangeKind.REMOVE)) { - kind = ActorStateChangeKind.UPDATE; - } - - this.stateChangeTracker.put(stateName, new StateChangeMetadata(kind, value)); - return true; - } - - return false; - }).filter(x -> x) - .switchIfEmpty(this.stateProvider.contains(this.actorTypeName, this.actorId, stateName) - .map(exists -> { - this.stateChangeTracker.put(stateName, - new StateChangeMetadata(exists ? ActorStateChangeKind.UPDATE : ActorStateChangeKind.ADD, value)); - return exists; - })) - .then(); - } - - /** - * Removes a given state from state store's cache. - * - * @param stateName State being stored. - * @return Asynchronous void result. - */ - public Mono remove(String stateName) { - return Mono.fromSupplier(() -> { - if (stateName == null) { - throw new IllegalArgumentException("State's name cannot be null."); - } - - if (this.stateChangeTracker.containsKey(stateName)) { - StateChangeMetadata metadata = this.stateChangeTracker.get(stateName); - - if (metadata.kind == ActorStateChangeKind.REMOVE) { - return true; - } - - if (metadata.kind == ActorStateChangeKind.ADD) { - this.stateChangeTracker.remove(stateName); - return true; - } - - this.stateChangeTracker.put(stateName, new StateChangeMetadata(ActorStateChangeKind.REMOVE, null)); - return true; - } - - return false; - }) - .filter(x -> x) - .switchIfEmpty(this.stateProvider.contains(this.actorTypeName, this.actorId, stateName)) - .filter(exists -> exists) - .map(exists -> { - this.stateChangeTracker.put(stateName, new StateChangeMetadata(ActorStateChangeKind.REMOVE, null)); - return exists; - }) - .then(); - } - - /** - * Checks if a given state exists in state store or cache. - * - * @param stateName State being checked. - * @return Asynchronous boolean result indicating whether state is present. - */ - public Mono contains(String stateName) { - return Mono.fromSupplier(() -> { - if (stateName == null) { - throw new IllegalArgumentException("State's name cannot be null."); - } - - return this.stateChangeTracker.get(stateName); - } - ).map(metadata -> { - if (metadata.kind == ActorStateChangeKind.REMOVE) { - return Boolean.FALSE; - } - - return Boolean.TRUE; - }).switchIfEmpty(this.stateProvider.contains(this.actorTypeName, this.actorId, stateName)); - } - - /** - * Saves all changes to state store. - * - * @return Asynchronous void result. - */ - public Mono save() { - return Mono.fromSupplier(() -> { - if (this.stateChangeTracker.isEmpty()) { - return null; - } - - List changes = new ArrayList<>(); - for (Map.Entry tuple : this.stateChangeTracker.entrySet()) { - if (tuple.getValue().kind == ActorStateChangeKind.NONE) { - continue; - } - - changes.add(new ActorStateChange(tuple.getKey(), tuple.getValue().value, tuple.getValue().kind)); - } - - return changes.toArray(new ActorStateChange[0]); - }).flatMap(changes -> this.stateProvider.apply(this.actorTypeName, this.actorId, changes)) - .then(Mono.fromRunnable(() -> this.flush())); - } - - /** - * Clears all changes not yet saved to state store. - */ - public void clear() { - this.stateChangeTracker.clear(); - } - - /** - * Commits the current cached values after successful save. - */ - private void flush() { - for (Map.Entry tuple : this.stateChangeTracker.entrySet()) { - String stateName = tuple.getKey(); - if (tuple.getValue().kind == ActorStateChangeKind.REMOVE) { - this.stateChangeTracker.remove(stateName); - } else { - StateChangeMetadata metadata = new StateChangeMetadata(ActorStateChangeKind.NONE, tuple.getValue().value); - this.stateChangeTracker.put(stateName, metadata); - } - } - } - - /** - * Internal class to represent value and change kind. - */ - private static final class StateChangeMetadata { - - /** - * Kind of change cached. - */ - private final ActorStateChangeKind kind; - - /** - * Value cached. - */ - private final Object value; - - /** - * Creates a new instance of the metadata on state change. - * - * @param kind Kind of change. - * @param value Value to be set. - */ - private StateChangeMetadata(ActorStateChangeKind kind, Object value) { - this.kind = kind; - this.value = value; - } - } -} +/* + * Copyright 2021 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 io.dapr.utils.TypeRef; +import reactor.core.publisher.Mono; + +import java.util.ArrayList; +import java.util.List; +import java.util.Map; +import java.util.NoSuchElementException; +import java.util.concurrent.ConcurrentHashMap; + +/** + * Manages state changes of a given Actor instance. + * All changes are cached in-memory until save() is called. + */ +public class ActorStateManager { + + /** + * Provides states using a state store. + */ + private final DaprStateAsyncProvider stateProvider; + + /** + * Name of the Actor's type. + */ + private final String actorTypeName; + + /** + * Actor's identifier. + */ + private final ActorId actorId; + + /** + * Cache of state changes in this Actor's instance. + */ + private final Map stateChangeTracker; + + /** + * Set once the Actor's instance is replaced, so it can't overwrite newer state. + */ + private volatile boolean discarded; + + /** + * Instantiates a new state manager for the given Actor's instance. + * + * @param stateProvider State store provider. + * @param actorTypeName Name of Actor's type. + * @param actorId Actor's identifier. + */ + ActorStateManager(DaprStateAsyncProvider stateProvider, String actorTypeName, ActorId actorId) { + this.stateProvider = stateProvider; + this.actorTypeName = actorTypeName; + this.actorId = actorId; + this.stateChangeTracker = new ConcurrentHashMap<>(); + } + + /** + * Adds a given key/value to the Actor's state store's cache. + * + * @param stateName Name of the state being added. + * @param value Value to be added. + * @param Type of the object being added. + * @return Asynchronous void operation. + */ + public Mono add(String stateName, T value) { + return Mono.fromSupplier(() -> { + if (stateName == null) { + throw new IllegalArgumentException("State's name cannot be null."); + } + + return null; + }).then(this.stateProvider.contains(this.actorTypeName, this.actorId, stateName) + .map(exists -> { + if (this.stateChangeTracker.containsKey(stateName)) { + StateChangeMetadata metadata = this.stateChangeTracker.get(stateName); + + if (metadata.kind == ActorStateChangeKind.REMOVE) { + this.stateChangeTracker.put(stateName, new StateChangeMetadata(ActorStateChangeKind.UPDATE, value)); + return true; + } + + throw new IllegalStateException("Duplicate cached state: " + stateName); + } + + if (exists) { + throw new IllegalStateException("Duplicate state: " + stateName); + } + + this.stateChangeTracker.put(stateName, new StateChangeMetadata(ActorStateChangeKind.ADD, value)); + return true; + })) + .then(); + } + + /** + * Fetches the most recent value for the given state, including cached value. + * + * @param stateName Name of the state. + * @param clazz Class type for the value being fetched. + * @param Type being fetched. + * @return Asynchronous response with fetched object. + */ + public Mono get(String stateName, Class clazz) { + return this.get(stateName, TypeRef.get(clazz)); + } + + /** + * Fetches the most recent value for the given state, including cached value. + * + * @param stateName Name of the state. + * @param type Class type for the value being fetched. + * @param Type being fetched. + * @return Asynchronous response with fetched object. + */ + public Mono get(String stateName, TypeRef type) { + return Mono.fromSupplier(() -> { + if (stateName == null) { + throw new IllegalArgumentException("State's name cannot be null."); + } + + if (this.stateChangeTracker.containsKey(stateName)) { + StateChangeMetadata metadata = this.stateChangeTracker.get(stateName); + + if (metadata.kind == ActorStateChangeKind.REMOVE) { + throw new NoSuchElementException("State is marked for removal: " + stateName); + } + + return (T) metadata.value; + } + + return (T) null; + }).switchIfEmpty( + this.stateProvider.load(this.actorTypeName, this.actorId, stateName, type) + .switchIfEmpty(Mono.error(new NoSuchElementException("State not found: " + stateName))) + .map(v -> { + this.stateChangeTracker.put(stateName, new StateChangeMetadata(ActorStateChangeKind.NONE, v)); + return (T) v; + })); + } + + /** + * Updates a given key/value pair in the state store's cache. + * + * @param stateName Name of the state being updated. + * @param value Value to be set for given state. + * @param Type of the value being set. + * @return Asynchronous void result. + */ + public Mono set(String stateName, T value) { + return Mono.fromSupplier(() -> { + if (stateName == null) { + throw new IllegalArgumentException("State's name cannot be null."); + } + + if (this.stateChangeTracker.containsKey(stateName)) { + StateChangeMetadata metadata = this.stateChangeTracker.get(stateName); + + ActorStateChangeKind kind = metadata.kind; + if ((kind == ActorStateChangeKind.NONE) || (kind == ActorStateChangeKind.REMOVE)) { + kind = ActorStateChangeKind.UPDATE; + } + + this.stateChangeTracker.put(stateName, new StateChangeMetadata(kind, value)); + return true; + } + + return false; + }).filter(x -> x) + .switchIfEmpty(this.stateProvider.contains(this.actorTypeName, this.actorId, stateName) + .map(exists -> { + this.stateChangeTracker.put(stateName, + new StateChangeMetadata(exists ? ActorStateChangeKind.UPDATE : ActorStateChangeKind.ADD, value)); + return exists; + })) + .then(); + } + + /** + * Removes a given state from state store's cache. + * + * @param stateName State being stored. + * @return Asynchronous void result. + */ + public Mono remove(String stateName) { + return Mono.fromSupplier(() -> { + if (stateName == null) { + throw new IllegalArgumentException("State's name cannot be null."); + } + + if (this.stateChangeTracker.containsKey(stateName)) { + StateChangeMetadata metadata = this.stateChangeTracker.get(stateName); + + if (metadata.kind == ActorStateChangeKind.REMOVE) { + return true; + } + + if (metadata.kind == ActorStateChangeKind.ADD) { + this.stateChangeTracker.remove(stateName); + return true; + } + + this.stateChangeTracker.put(stateName, new StateChangeMetadata(ActorStateChangeKind.REMOVE, null)); + return true; + } + + return false; + }) + .filter(x -> x) + .switchIfEmpty(this.stateProvider.contains(this.actorTypeName, this.actorId, stateName)) + .filter(exists -> exists) + .map(exists -> { + this.stateChangeTracker.put(stateName, new StateChangeMetadata(ActorStateChangeKind.REMOVE, null)); + return exists; + }) + .then(); + } + + /** + * Checks if a given state exists in state store or cache. + * + * @param stateName State being checked. + * @return Asynchronous boolean result indicating whether state is present. + */ + public Mono contains(String stateName) { + return Mono.fromSupplier(() -> { + if (stateName == null) { + throw new IllegalArgumentException("State's name cannot be null."); + } + + return this.stateChangeTracker.get(stateName); + } + ).map(metadata -> { + if (metadata.kind == ActorStateChangeKind.REMOVE) { + return Boolean.FALSE; + } + + return Boolean.TRUE; + }).switchIfEmpty(this.stateProvider.contains(this.actorTypeName, this.actorId, stateName)); + } + + /** + * Saves all changes to state store. + * + * @return Asynchronous void result. + */ + public Mono save() { + return Mono.fromSupplier(() -> { + if (this.discarded) { + throw new IllegalStateException( + String.format("Actor %s of type %s was replaced by a newer activation, state not saved.", + this.actorId, this.actorTypeName)); + } + + if (this.stateChangeTracker.isEmpty()) { + return null; + } + + List changes = new ArrayList<>(); + for (Map.Entry tuple : this.stateChangeTracker.entrySet()) { + if (tuple.getValue().kind == ActorStateChangeKind.NONE) { + continue; + } + + changes.add(new ActorStateChange(tuple.getKey(), tuple.getValue().value, tuple.getValue().kind)); + } + + return changes.toArray(new ActorStateChange[0]); + }).flatMap(changes -> this.stateProvider.apply(this.actorTypeName, this.actorId, changes)) + .then(Mono.fromRunnable(() -> this.flush())); + } + + /** + * Refuses every later save. + */ + void discard() { + this.discarded = true; + } + + /** + * Clears all changes not yet saved to state store. + */ + public void clear() { + this.stateChangeTracker.clear(); + } + + /** + * Commits the current cached values after successful save. + */ + private void flush() { + for (Map.Entry tuple : this.stateChangeTracker.entrySet()) { + String stateName = tuple.getKey(); + if (tuple.getValue().kind == ActorStateChangeKind.REMOVE) { + this.stateChangeTracker.remove(stateName); + } else { + StateChangeMetadata metadata = new StateChangeMetadata(ActorStateChangeKind.NONE, tuple.getValue().value); + this.stateChangeTracker.put(stateName, metadata); + } + } + } + + /** + * Internal class to represent value and change kind. + */ + private static final class StateChangeMetadata { + + /** + * Kind of change cached. + */ + private final ActorStateChangeKind kind; + + /** + * Value cached. + */ + private final Object value; + + /** + * Creates a new instance of the metadata on state change. + * + * @param kind Kind of change. + * @param value Value to be set. + */ + private StateChangeMetadata(ActorStateChangeKind kind, Object value) { + this.kind = kind; + this.value = value; + } + } +} 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/ActorEventDispatcherTest.java b/sdk-actors/src/test/java/io/dapr/actors/runtime/ActorEventDispatcherTest.java index fa59799994..1ddaaebe31 100644 --- a/sdk-actors/src/test/java/io/dapr/actors/runtime/ActorEventDispatcherTest.java +++ b/sdk-actors/src/test/java/io/dapr/actors/runtime/ActorEventDispatcherTest.java @@ -47,7 +47,12 @@ import static org.junit.jupiter.api.Assertions.assertFalse; import static org.junit.jupiter.api.Assertions.assertThrows; import static org.junit.jupiter.api.Assertions.assertTrue; +import static org.mockito.ArgumentMatchers.any; +import static org.mockito.ArgumentMatchers.eq; +import static org.mockito.ArgumentMatchers.isNull; import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.spy; +import static org.mockito.Mockito.verify; public class ActorEventDispatcherTest { @@ -190,6 +195,44 @@ public void invokePassesData() throws Exception { assertEquals("payload", LAST_TIMER_DATA.get()); } + @Test + public void callbacksPassTheActivationId() throws Exception { + ActorRuntime spied = spy(runtime); + String actorId = UUID.randomUUID().toString(); + + ActorEventDispatcher.dispatch(spied, SubscribeActorEventsResponseAlpha1.newBuilder() + .setInvokeRequest(SubscribeActorEventsResponseInvokeRequestAlpha1.newBuilder() + .setId("1") + .setActorType("MyRemindableActor") + .setActorId(actorId) + .setMethod("receiveTimer") + .setData(ByteString.copyFrom(SERIALIZER.serialize("payload"))) + .putMetadata("dapr-actor-activation-id", "a1")) + .build()).block(); + ActorEventDispatcher.dispatch(spied, SubscribeActorEventsResponseAlpha1.newBuilder() + .setReminderRequest(SubscribeActorEventsResponseReminderRequestAlpha1.newBuilder() + .setId("2") + .setActorType("MyRemindableActor") + .setActorId(actorId) + .setName("myreminder") + .putMetadata("Dapr-Actor-Activation-Id", "a2")) + .build()).block(); + ActorEventDispatcher.dispatch(spied, SubscribeActorEventsResponseAlpha1.newBuilder() + .setTimerRequest(SubscribeActorEventsResponseTimerRequestAlpha1.newBuilder() + .setId("3") + .setActorType("MyRemindableActor") + .setActorId(actorId) + .setName("mytimer") + .setCallback("receiveTimer") + .setData(daprData("tick")) + .putMetadata("Dapr-Actor-Activation-Id", "")) + .build()).block(); + + verify(spied).invoke(eq("MyRemindableActor"), eq(actorId), eq("receiveTimer"), any(), eq("a1")); + verify(spied).invokeReminder(eq("MyRemindableActor"), eq(actorId), eq("myreminder"), any(), eq("a2")); + verify(spied).invokeTimer(eq("MyRemindableActor"), eq(actorId), eq("mytimer"), any(), isNull()); + } + @Test public void callbacksForUnknownActorTypeAreNotFound() { String actorId = UUID.randomUUID().toString(); 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 b8be753ccf..d6017fa603 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 @@ -25,8 +25,14 @@ import reactor.core.scheduler.Schedulers; 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 java.util.function.BooleanSupplier; @@ -123,6 +129,216 @@ 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(); + } + + interface FailingActor { + Mono writeThenFail(String value); + } + + @ActorType(name = "GatedActor") + public static class GatedActorImpl extends AbstractActor implements GatedActor, FailingActor, Remindable { + + private final Mono gate; + + private int calls; + + private int deactivations; + + private int postHooks; + + public GatedActorImpl(ActorRuntimeContext runtimeContext, ActorId id, Mono gate) { + super(runtimeContext, id); + this.gate = gate; + } + + @Override + 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; + } + + @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; + } + } + + @ActorType(name = "SaveOnDeactivateActor") + public static class SaveOnDeactivateActorImpl extends AbstractActor implements GatedActor { + + private final Mono gate; + + private final String tag; + + public SaveOnDeactivateActorImpl(ActorRuntimeContext runtimeContext, ActorId id, Mono gate, String tag) { + super(runtimeContext, id); + this.gate = gate; + this.tag = tag; + } + + @Override + protected Mono onDeactivate() { + return this.getActorStateManager().set("value", this.tag + "-deactivated").then(this.saveState()); + } + + @Override + public Mono write(String value) { + return this.gate.then(Mono.defer(() -> this.getActorStateManager().set("value", value))) + .then(this.saveState()); + } + + @Override + public int increment() { + return 0; + } + } + + 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; + } + } + interface SlowActor { Mono hold(); @@ -337,6 +553,547 @@ 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(); + CompletableFuture reset = manager.activateActor(actorId, "a2").toFuture(); + gate.complete(null); + reset.get(5, TimeUnit.SECONDS); + write(manager, actorId, "fresh").block(); + + 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(); + CompletableFuture reset = manager.activateActor(actorId, "a2").toFuture(); + gate.complete(null); + reset.get(5, TimeUnit.SECONDS); + write(manager, actorId, "fresh").block(); + + 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(); + CompletableFuture reset = manager.activateActor(actorId, "a2").toFuture(); + gate.complete(null); + reset.get(5, TimeUnit.SECONDS); + write(manager, actorId, "fresh").block(); + + Assertions.assertEquals(List.of("value=fresh"), provider.saved); + 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(); + CompletableFuture reset = manager.activateActor(actorId, "a2").toFuture(); + gate.complete(null); + reset.get(5, TimeUnit.SECONDS); + + 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(); + CompletableFuture deactivation = manager.deactivateActor(actorId).toFuture(); + gate.complete(null); + deactivation.get(5, TimeUnit.SECONDS); + + 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(); + 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(); + CompletableFuture deactivation = manager.deactivateActor(actorId).toFuture(); + gate.complete(null); + deactivation.get(5, TimeUnit.SECONDS); + + 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 queuedActivationWithDifferentIdFailsAndKeepsFirst() throws Exception { + RecordingStateProvider provider = new RecordingStateProvider(); + List created = new ArrayList<>(); + CompletableFuture activateGate = new CompletableFuture<>(); + ActorManager manager = createLateActivationManager( + provider, created, activateGate, CompletableFuture.completedFuture(null), false); + ActorId actorId = newActorId(); + + CompletableFuture first = manager.activateActor(actorId, "a").toFuture(); + CompletableFuture queued = manager.activateActor(actorId, "b").toFuture(); + activateGate.complete(null); + + first.get(5, TimeUnit.SECONDS); + assertCause(IllegalStateException.class, queued); + Assertions.assertEquals(1, created.size()); + + // Dapr retries the failed call, which then replaces the first activation. + manager.activateActor(actorId, "b").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); + Assertions.assertEquals(List.of("value=late", "value=newer"), provider.saved); + } + + @Test + public void queuedActivationWithSameIdReusesFirst() throws Exception { + assertQueuedActivationReusesFirst("a"); + } + + @Test + public void queuedActivationWithoutIdReusesFirst() throws Exception { + assertQueuedActivationReusesFirst(null); + } + + private static void assertQueuedActivationReusesFirst(String queuedId) 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> first = + manager.activateActor(actorId, "a").toFuture(); + CompletableFuture> queued = + manager.activateActor(actorId, queuedId).toFuture(); + activateGate.complete(null); + + Assertions.assertSame(first.get(5, TimeUnit.SECONDS), queued.get(5, TimeUnit.SECONDS)); + Assertions.assertEquals(1, created.size()); + Assertions.assertEquals(List.of("value=late"), provider.saved); + } + + @Test + public void replacedActorCannotSaveFromOnDeactivate() throws Exception { + RecordingStateProvider provider = new RecordingStateProvider(); + List created = new ArrayList<>(); + ActorManager manager = + createSaveOnDeactivateManager(provider, created, CompletableFuture.completedFuture(null)); + ActorId actorId = newActorId(); + + manager.activateActor(actorId, "a1").block(); + manager.activateActor(actorId, "a2").block(); + Assertions.assertEquals(List.of(), provider.saved); + + manager.deactivateActor(actorId).block(); + Assertions.assertEquals(List.of("value=actor1-deactivated"), provider.saved); + } + + @Test + public void replacedActorCannotSaveFromRunningCall() throws Exception { + RecordingStateProvider provider = new RecordingStateProvider(); + List created = new ArrayList<>(); + CompletableFuture gate = new CompletableFuture<>(); + ActorManager manager = createSaveOnDeactivateManager(provider, created, gate); + ActorId actorId = newActorId(); + + manager.activateActor(actorId, "a1").block(); + CompletableFuture stale = manager.invokeMethod( + actorId, "write", new DefaultObjectSerializer().serialize("stale")).toFuture(); + CompletableFuture reset = manager.activateActor(actorId, "a2").toFuture(); + gate.complete(null); + + assertCause(IllegalStateException.class, stale); + reset.get(5, TimeUnit.SECONDS); + Assertions.assertEquals(List.of(), provider.saved); + Assertions.assertEquals(2, created.size()); + } + + @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(2, created.size()); + Assertions.assertEquals(1, created.get(0).deactivations); + Assertions.assertEquals(0, created.get(1).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 createSaveOnDeactivateManager( + 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(); + SaveOnDeactivateActorImpl actor = + new SaveOnDeactivateActorImpl(ctx, id, actorGate, "actor" + created.size()); + created.add(actor); + return actor; + }, + ActorTypeInformation.create(SaveOnDeactivateActorImpl.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); + } + @Test public void cancelledCallKeepsTurnThroughPostMethod() throws Exception { SlowActorImpl.reset(); 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..c049851464 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 put(K key, V value) { + V previous = super.put(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; @@ -158,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); } @@ -417,6 +516,230 @@ 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 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(); + 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); + pauseGate.complete(null); + live.get(0).get(5, TimeUnit.SECONDS); + + 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-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 fb971bd573..390ade1429 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 @@ -49,7 +49,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; @@ -106,7 +108,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)); 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"; }