Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension


Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
Original file line number Diff line number Diff line change
Expand Up @@ -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.
*
Expand Down Expand Up @@ -383,15 +390,17 @@ Mono<Void> 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<Void> onPostActorMethodInternal(ActorMethodContext actorMethodContext) {
Mono<Void> onPostActorMethodInternal(ActorMethodContext actorMethodContext, Mono<Void> 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)));
}
Expand Down
Original file line number Diff line number Diff line change
@@ -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<Mono<String>> runtimeVersion;

private final AtomicReference<Mono<Boolean>> lookup = new AtomicReference<>();

ActivationIdGate(Supplier<Mono<String>> 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<Optional<String>> 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.<String>empty())
.onErrorReturn(Optional.empty());
}

private Mono<Boolean> trusted() {
Mono<Boolean> cached = this.lookup.get();
if (cached != null) {
return cached;
}
AtomicReference<Mono<Boolean>> self = new AtomicReference<>();
Mono<Boolean> 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<Boolean> 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;
}
}
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand All @@ -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
Expand Down Expand Up @@ -148,7 +150,8 @@ private static Mono<SubscribeActorEventsRequestAlpha1> 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()
Expand All @@ -169,7 +172,8 @@ private static Mono<SubscribeActorEventsRequestAlpha1> 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()))
Expand All @@ -186,7 +190,8 @@ private static Mono<SubscribeActorEventsRequestAlpha1> 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()))
Expand Down Expand Up @@ -249,6 +254,15 @@ private static boolean isRegistered(ActorRuntime runtime, String actorType) {
return runtime.getConfig().getRegisteredActorTypes().contains(actorType);
}

private static String activationId(Map<String, String> metadata) {
for (Map.Entry<String, String> 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));
Expand Down
Loading
Loading