From 2ac527f3458b6a0155d41684a5ffff1e52036b46 Mon Sep 17 00:00:00 2001 From: joshvanl Date: Wed, 1 Jul 2026 14:41:59 +0100 Subject: [PATCH 1/2] Add stateful-history delta work items to the workflow worker The sidecar re-sends a workflow instance's entire committed history to the worker on every turn. This adds the worker half of the "stateful history" optimization so that, once a worker is warm for an instance on a work-item stream, the sidecar sends only the new committed events (the delta) and the worker reconstructs the full history from its own cache. It mirrors the Go (durabletask-go), Python, and .NET SDK implementations and is on by default. Worker (durabletask-client): - WorkflowHistoryCache: a per-stream cache of each instance's committed history, bounded by a sliding TTL, an instance-count cap, and a byte budget with LRU eviction. Injectable clock for deterministic tests. - DurableTaskGrpcWorker: advertise WORKER_CAPABILITY_STATEFUL_HISTORY in GetWorkItemsRequest, reset the cache on every reconnect (the sidecar drops the old stream's warm set), and reclaim idle entries with a daemon janitor stopped on close. - OrchestratorRunner: before replay, resolve the full committed history (cached prefix + delta on a hit, or a GetInstanceHistory fetch on a miss) instead of using the request's pastEvents directly; after replay, cache the committed history, or drop it once the instance ends (a CompleteWorkflow action, covering completed/failed/terminated/continued-as-new). A TerminateWorkflow action targets a different instance and is deliberately not treated as a reset. Correctness never depends on the cache: any miss (cold stream, eviction, desync) self-heals via the GetInstanceHistory fallback, so this only changes per-turn bandwidth, not results. A fallback fetch that fails abandons the work item for backend redelivery rather than completing with a partial history. Configuration (DurableTaskGrpcWorkerBuilder): - disableStatefulHistory to opt out, plus historyCacheTtl, historyCacheMaxInstances, and historyCacheMaxBytes to tune the bounds. Signed-off-by: joshvanl --- .../durabletask/DurableTaskGrpcWorker.java | 137 ++++++++--- .../DurableTaskGrpcWorkerBuilder.java | 60 +++++ .../durabletask/WorkflowHistoryCache.java | 220 ++++++++++++++++++ .../runner/OrchestratorRunner.java | 113 ++++++++- ...ableTaskGrpcWorkerStatefulHistoryTest.java | 200 ++++++++++++++++ .../dapr/durabletask/IntegrationTestBase.java | 15 ++ .../dapr/durabletask/StatefulHistoryIT.java | 161 +++++++++++++ .../io/dapr/durabletask/WorkItemObserver.java | 132 +++++++++++ .../durabletask/WorkflowHistoryCacheTest.java | 180 ++++++++++++++ .../runner/OrchestratorRunnerHistoryTest.java | 128 ++++++++++ .../runtime/WorkflowRuntimeBuilder.java | 57 +++++ .../runtime/WorkflowRuntimeBuilderTest.java | 43 ++++ 12 files changed, 1412 insertions(+), 34 deletions(-) create mode 100644 durabletask-client/src/main/java/io/dapr/durabletask/WorkflowHistoryCache.java create mode 100644 durabletask-client/src/test/java/io/dapr/durabletask/DurableTaskGrpcWorkerStatefulHistoryTest.java create mode 100644 durabletask-client/src/test/java/io/dapr/durabletask/StatefulHistoryIT.java create mode 100644 durabletask-client/src/test/java/io/dapr/durabletask/WorkItemObserver.java create mode 100644 durabletask-client/src/test/java/io/dapr/durabletask/WorkflowHistoryCacheTest.java create mode 100644 durabletask-client/src/test/java/io/dapr/durabletask/runner/OrchestratorRunnerHistoryTest.java diff --git a/durabletask-client/src/main/java/io/dapr/durabletask/DurableTaskGrpcWorker.java b/durabletask-client/src/main/java/io/dapr/durabletask/DurableTaskGrpcWorker.java index 8fae93cf93..bd1dbe4a3b 100644 --- a/durabletask-client/src/main/java/io/dapr/durabletask/DurableTaskGrpcWorker.java +++ b/durabletask-client/src/main/java/io/dapr/durabletask/DurableTaskGrpcWorker.java @@ -36,6 +36,7 @@ import java.util.Map; import java.util.concurrent.ExecutorService; import java.util.concurrent.Executors; +import java.util.concurrent.ScheduledExecutorService; import java.util.concurrent.TimeUnit; import java.util.logging.Level; import java.util.logging.Logger; @@ -63,6 +64,13 @@ public final class DurableTaskGrpcWorker implements AutoCloseable { private final TaskHubSidecarServiceGrpc.TaskHubSidecarServiceBlockingStub sidecarClient; private final boolean isExecutorServiceManaged; + + // Per-stream cache of each instance's committed history for the stateful-history optimization, + // or null when disabled. Reset on every reconnect and swept for idle entries by the janitor. + private final WorkflowHistoryCache historyCache; + private static final Duration HISTORY_SWEEP_INTERVAL = Duration.ofMinutes(1); + private volatile ScheduledExecutorService historyJanitor; + private volatile boolean isNormalShutdown = false; private volatile Thread workerThread; @@ -104,7 +112,8 @@ public final class DurableTaskGrpcWorker implements AutoCloseable { this.isExecutorServiceManaged = builder.executorService == null; - + this.historyCache = builder.disableStatefulHistory ? null : new WorkflowHistoryCache( + builder.historyCacheTtl, builder.historyCacheMaxInstances, builder.historyCacheMaxBytes); } /** @@ -137,6 +146,7 @@ public void close() { this.workerThread.interrupt(); } this.isNormalShutdown = true; + this.shutDownHistoryJanitor(); this.shutDownWorkerPool(); this.closeSideCarChannel(); } @@ -172,41 +182,82 @@ public void startAndBlock() { this.dataConverter, logger); + this.startHistoryJanitor(); + while (!this.isNormalShutdown && !Thread.currentThread().isInterrupted()) { try { - OrchestratorService.GetWorkItemsRequest getWorkItemsRequest = OrchestratorService.GetWorkItemsRequest - .newBuilder().build(); - Iterator workItemStream = this.sidecarClient.getWorkItems(getWorkItemsRequest); - while (workItemStream.hasNext()) { - if (this.isNormalShutdown || Thread.currentThread().isInterrupted()) { - break; + // Each iteration establishes a fresh work-item stream. Start it cold: the sidecar drops + // the previous stream's warm set, so any histories cached from it are no longer in sync. + if (this.historyCache != null) { + this.historyCache.reset(); + } + + // Advertise the stateful-history capability so the sidecar can send deltas instead of the + // full history on each turn. Absent it, the sidecar always sends the full history. + OrchestratorService.GetWorkItemsRequest.Builder requestBuilder = OrchestratorService.GetWorkItemsRequest + .newBuilder(); + if (this.historyCache != null) { + requestBuilder.addCapabilities(OrchestratorService.WorkerCapability.WORKER_CAPABILITY_STATEFUL_HISTORY); + } + OrchestratorService.GetWorkItemsRequest getWorkItemsRequest = requestBuilder.build(); + + // Scope this stream to a cancellable gRPC context so a work item whose committed history + // cannot be resolved can drop the stream it arrived on. The sidecar then cancels and + // redelivers that stream's pending items straight away rather than waiting out the lease. + io.grpc.Context.CancellableContext streamContext = io.grpc.Context.current().withCancellation(); + try { + Iterator workItemStream; + io.grpc.Context previousContext = streamContext.attach(); + try { + // Started while streamContext is current, so the RPC inherits its cancellation. + workItemStream = this.sidecarClient.getWorkItems(getWorkItemsRequest); + } finally { + streamContext.detach(previousContext); } - OrchestratorService.WorkItem workItem = workItemStream.next(); - OrchestratorService.WorkItem.RequestCase requestType = workItem.getRequestCase(); - - if (requestType == OrchestratorService.WorkItem.RequestCase.WORKFLOWREQUEST) { - OrchestratorService.WorkflowRequest orchestratorRequest = workItem.getWorkflowRequest(); - logger.log(Level.FINEST, - String.format("Processing orchestrator request for instance: {0}", - orchestratorRequest.getInstanceId())); - - this.workerPool.submit(new OrchestratorRunner(workItem, taskOrchestrationExecutor, sidecarClient, tracer)); - } else if (requestType == OrchestratorService.WorkItem.RequestCase.ACTIVITYREQUEST) { - OrchestratorService.ActivityRequest activityRequest = workItem.getActivityRequest(); - - logger.log(Level.INFO, - String.format("Processing activity request: %s for instance: %s, gRPC thread context: %s", - activityRequest.getName(), - activityRequest.getWorkflowInstance().getInstanceId(), - Context.current())); - - this.workerPool.submit(new ActivityRunner(workItem, taskActivityExecutor, sidecarClient, tracer)); - - } else { - logger.log(Level.WARNING, - "Received and dropped an unknown '{0}' work-item from the sidecar.", - requestType); + + // Bound to this stream, so a late failure from a superseded connection is a no-op + // instead of killing whichever stream happens to be current. + Runnable teardownStream = () -> streamContext.cancel(Status.CANCELLED + .withDescription("worker could not resolve a workflow history") + .asRuntimeException()); + + while (workItemStream.hasNext()) { + if (this.isNormalShutdown || Thread.currentThread().isInterrupted()) { + break; + } + OrchestratorService.WorkItem workItem = workItemStream.next(); + OrchestratorService.WorkItem.RequestCase requestType = workItem.getRequestCase(); + + if (requestType == OrchestratorService.WorkItem.RequestCase.WORKFLOWREQUEST) { + OrchestratorService.WorkflowRequest orchestratorRequest = workItem.getWorkflowRequest(); + logger.log(Level.FINEST, + String.format("Processing orchestrator request for instance: {0}", + orchestratorRequest.getInstanceId())); + + this.workerPool.submit( + new OrchestratorRunner(workItem, taskOrchestrationExecutor, sidecarClient, tracer, + historyCache, teardownStream)); + } else if (requestType == OrchestratorService.WorkItem.RequestCase.ACTIVITYREQUEST) { + OrchestratorService.ActivityRequest activityRequest = workItem.getActivityRequest(); + + logger.log(Level.INFO, + String.format("Processing activity request: %s for instance: %s, gRPC thread context: %s", + activityRequest.getName(), + activityRequest.getWorkflowInstance().getInstanceId(), + Context.current())); + + this.workerPool.submit(new ActivityRunner(workItem, taskActivityExecutor, sidecarClient, tracer)); + + } else { + logger.log(Level.WARNING, + "Received and dropped an unknown '{0}' work-item from the sidecar.", + requestType); + } } + } finally { + // Releases this context's cancellation listener on its parent; without it the reconnect + // loop would accumulate one per iteration. + streamContext.close(); } } catch (StatusRuntimeException e) { if (e.getStatus().getCode() == Status.Code.UNAVAILABLE) { @@ -264,6 +315,28 @@ private void closeSideCarChannel() { } } + private void startHistoryJanitor() { + if (this.historyCache == null || this.historyJanitor != null) { + return; + } + ScheduledExecutorService janitor = Executors.newSingleThreadScheduledExecutor(runnable -> { + Thread thread = new Thread(runnable, "dapr-workflow-history-janitor"); + thread.setDaemon(true); + return thread; + }); + long sweepSeconds = HISTORY_SWEEP_INTERVAL.getSeconds(); + janitor.scheduleWithFixedDelay(this.historyCache::sweepExpired, sweepSeconds, sweepSeconds, TimeUnit.SECONDS); + this.historyJanitor = janitor; + } + + private void shutDownHistoryJanitor() { + ScheduledExecutorService janitor = this.historyJanitor; + if (janitor != null) { + janitor.shutdownNow(); + this.historyJanitor = null; + } + } + private void shutDownWorkerPool() { if (this.isExecutorServiceManaged) { if (!this.isNormalShutdown) { diff --git a/durabletask-client/src/main/java/io/dapr/durabletask/DurableTaskGrpcWorkerBuilder.java b/durabletask-client/src/main/java/io/dapr/durabletask/DurableTaskGrpcWorkerBuilder.java index ad60577256..b7135643eb 100644 --- a/durabletask-client/src/main/java/io/dapr/durabletask/DurableTaskGrpcWorkerBuilder.java +++ b/durabletask-client/src/main/java/io/dapr/durabletask/DurableTaskGrpcWorkerBuilder.java @@ -34,6 +34,10 @@ public final class DurableTaskGrpcWorkerBuilder { Duration maximumTimerInterval; ExecutorService executorService; String appId; // App ID for cross-app routing + boolean disableStatefulHistory; + Duration historyCacheTtl; + int historyCacheMaxInstances; + long historyCacheMaxBytes; /** * Adds an orchestration factory to be used by the constructed {@link DurableTaskGrpcWorker}. @@ -145,6 +149,62 @@ public DurableTaskGrpcWorkerBuilder appId(String appId) { return this; } + /** + * Disables the stateful-history optimization. + * + *

By default the worker advertises {@code WORKER_CAPABILITY_STATEFUL_HISTORY} and caches each + * instance's committed history per work-item stream, so the sidecar can send only the new events + * (the delta) each turn instead of the full history. A cache miss is always recovered safely via + * the GetInstanceHistory RPC, so disabling this only affects per-turn bandwidth, never correctness. + * When disabled, the worker always receives the full history.

+ * + * @param disableStatefulHistory whether to disable the stateful-history optimization + * @return this builder object + */ + public DurableTaskGrpcWorkerBuilder disableStatefulHistory(boolean disableStatefulHistory) { + this.disableStatefulHistory = disableStatefulHistory; + return this; + } + + /** + * Sets the sliding time-to-live for cached instance histories. An instance's entry is reclaimed + * once it has gone idle (no turn) for longer than this. If not specified, a default of one hour is + * used. Ignored when the stateful-history optimization is disabled. + * + * @param historyCacheTtl the sliding time-to-live for a cached instance history + * @return this builder object + */ + public DurableTaskGrpcWorkerBuilder historyCacheTtl(Duration historyCacheTtl) { + this.historyCacheTtl = historyCacheTtl; + return this; + } + + /** + * Sets the maximum number of per-instance histories retained on a single work-item stream; + * least-recently-used entries are evicted beyond it. A non-positive value uses the built-in + * default. Ignored when the stateful-history optimization is disabled. + * + * @param historyCacheMaxInstances the instance-count cap for the history cache + * @return this builder object + */ + public DurableTaskGrpcWorkerBuilder historyCacheMaxInstances(int historyCacheMaxInstances) { + this.historyCacheMaxInstances = historyCacheMaxInstances; + return this; + } + + /** + * Sets the byte budget for cached histories on a single work-item stream; least-recently-used + * entries are evicted beyond it. A non-positive value means unlimited (bounded only by the + * instance-count cap and the TTL). Ignored when the stateful-history optimization is disabled. + * + * @param historyCacheMaxBytes the byte budget for the history cache + * @return this builder object + */ + public DurableTaskGrpcWorkerBuilder historyCacheMaxBytes(long historyCacheMaxBytes) { + this.historyCacheMaxBytes = historyCacheMaxBytes; + return this; + } + /** * Initializes a new {@link DurableTaskGrpcWorker} object with the settings specified in the current builder object. * diff --git a/durabletask-client/src/main/java/io/dapr/durabletask/WorkflowHistoryCache.java b/durabletask-client/src/main/java/io/dapr/durabletask/WorkflowHistoryCache.java new file mode 100644 index 0000000000..515d469566 --- /dev/null +++ b/durabletask-client/src/main/java/io/dapr/durabletask/WorkflowHistoryCache.java @@ -0,0 +1,220 @@ +/* + * 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.durabletask; + +import io.dapr.durabletask.implementation.protobuf.HistoryEvents; + +import java.time.Duration; +import java.util.ArrayList; +import java.util.Collections; +import java.util.HashMap; +import java.util.List; +import java.util.Map; +import java.util.function.LongSupplier; + +/** + * Per-stream cache of each workflow instance's committed history, enabling "stateful history" + * delta work items: a worker that advertises {@code WORKER_CAPABILITY_STATEFUL_HISTORY} retains + * the history it has already replayed so the sidecar can send only the new events (the delta). + * Entries are reclaimed by a sliding TTL, an instance-count cap, and an optional byte budget + * (LRU eviction). Eviction is always safe: a miss is recovered via the GetInstanceHistory RPC. + * + *

Thread-safe; work items are processed concurrently on the worker pool.

+ */ +public final class WorkflowHistoryCache { + + static final Duration DEFAULT_TTL = Duration.ofHours(1); + static final int DEFAULT_MAX_INSTANCES = 100_000; + + private static final class Entry { + final List events; + final long bytes; + long lastAccess; + + Entry(List events, long bytes, long lastAccess) { + this.events = events; + this.bytes = bytes; + this.lastAccess = lastAccess; + } + } + + private final Object lock = new Object(); + private final Map entries = new HashMap<>(); + private final long ttlNanos; + private final int maxInstances; + private final long maxBytes; + private final LongSupplier clockNanos; + private long totalBytes; + + /** + * Constructs a cache with the default monotonic clock. Non-positive ttl/maxInstances use the + * package defaults; a non-positive maxBytes means unlimited (bounded by ttl and maxInstances). + * + * @param ttl sliding time-to-live for an idle instance's entry + * @param maxInstances instance-count cap + * @param maxBytes byte budget, or {@code <= 0} for unlimited + */ + public WorkflowHistoryCache(Duration ttl, int maxInstances, long maxBytes) { + this(ttl, maxInstances, maxBytes, System::nanoTime); + } + + /** + * Constructs a cache with an injectable clock, for deterministic tests. + * + * @param clockNanos supplier of a monotonic nanosecond timestamp (e.g. {@code System::nanoTime}) + */ + WorkflowHistoryCache(Duration ttl, int maxInstances, long maxBytes, LongSupplier clockNanos) { + Duration effectiveTtl = ttl != null && !ttl.isZero() && !ttl.isNegative() ? ttl : DEFAULT_TTL; + this.ttlNanos = effectiveTtl.toNanos(); + this.maxInstances = maxInstances > 0 ? maxInstances : DEFAULT_MAX_INSTANCES; + this.maxBytes = maxBytes > 0 ? maxBytes : 0; + this.clockNanos = clockNanos; + } + + /** + * Returns the cached committed history for an instance, refreshing its TTL, or {@code null} on a + * miss. + * + *

The returned list is an unmodifiable view of the cached entry rather than a copy, so reading + * it stays allocation-free on the hot path while a caller cannot corrupt the cache by mutating + * what it was handed. + * + * @param instanceId the workflow instance ID + * @return the cached committed history, or {@code null} if the instance is not cached + */ + public List get(String instanceId) { + synchronized (this.lock) { + Entry entry = this.entries.get(instanceId); + if (entry == null) { + return null; + } + entry.lastAccess = this.clockNanos.getAsLong(); + return Collections.unmodifiableList(entry.events); + } + } + + /** + * Caches an instance's committed history, evicting least-recently-used entries to stay within + * the configured bounds. + * + * @param instanceId the workflow instance ID + * @param events the committed history to cache for the instance + */ + public void put(String instanceId, List events) { + List snapshot = new ArrayList<>(events); + long bytes = 0; + for (HistoryEvents.HistoryEvent event : snapshot) { + bytes += event.getSerializedSize(); + } + + synchronized (this.lock) { + Entry existing = this.entries.get(instanceId); + if (existing != null) { + this.totalBytes -= existing.bytes; + } + this.entries.put(instanceId, new Entry(snapshot, bytes, this.clockNanos.getAsLong())); + this.totalBytes += bytes; + this.evictToFit(instanceId); + } + } + + /** + * Drops an instance's cached history (e.g. once it completes). + * + * @param instanceId the workflow instance ID + */ + public void remove(String instanceId) { + synchronized (this.lock) { + this.removeLocked(instanceId); + } + } + + /** Clears the cache; used when the stream reconnects (and starts cold). */ + public void reset() { + synchronized (this.lock) { + this.entries.clear(); + this.totalBytes = 0; + } + } + + /** Evicts entries whose last turn was longer ago than the TTL. */ + public void sweepExpired() { + long now = this.clockNanos.getAsLong(); + synchronized (this.lock) { + List expired = new ArrayList<>(); + for (Map.Entry entry : this.entries.entrySet()) { + if (now - entry.getValue().lastAccess > this.ttlNanos) { + expired.add(entry.getKey()); + } + } + for (String instanceId : expired) { + this.removeLocked(instanceId); + } + } + } + + int size() { + synchronized (this.lock) { + return this.entries.size(); + } + } + + long totalBytes() { + synchronized (this.lock) { + return this.totalBytes; + } + } + + private void removeLocked(String instanceId) { + Entry removed = this.entries.remove(instanceId); + if (removed != null) { + this.totalBytes -= removed.bytes; + } + } + + /** + * Evicts least-recently-used entries until within the count and byte bounds, always keeping the + * just-touched entry so the active working set is never evicted. A lone entry over the byte + * budget is kept (a soft overage) rather than thrashing. + */ + private void evictToFit(String keep) { + while (this.entries.size() > 1) { + boolean overCount = this.entries.size() > this.maxInstances; + boolean overBytes = this.maxBytes > 0 && this.totalBytes > this.maxBytes; + if (!overCount && !overBytes) { + return; + } + String victim = this.leastRecentlyUsedExcept(keep); + if (victim == null) { + return; + } + this.removeLocked(victim); + } + } + + private String leastRecentlyUsedExcept(String keep) { + String oldest = null; + long oldestAccess = Long.MAX_VALUE; + for (Map.Entry entry : this.entries.entrySet()) { + if (entry.getKey().equals(keep)) { + continue; + } + if (oldest == null || entry.getValue().lastAccess < oldestAccess) { + oldest = entry.getKey(); + oldestAccess = entry.getValue().lastAccess; + } + } + return oldest; + } +} diff --git a/durabletask-client/src/main/java/io/dapr/durabletask/runner/OrchestratorRunner.java b/durabletask-client/src/main/java/io/dapr/durabletask/runner/OrchestratorRunner.java index e46cbe978b..2c9d7c17e4 100644 --- a/durabletask-client/src/main/java/io/dapr/durabletask/runner/OrchestratorRunner.java +++ b/durabletask-client/src/main/java/io/dapr/durabletask/runner/OrchestratorRunner.java @@ -16,7 +16,10 @@ import com.google.protobuf.StringValue; import io.dapr.durabletask.TaskOrchestrationExecutor; import io.dapr.durabletask.TaskOrchestratorResult; +import io.dapr.durabletask.WorkflowHistoryCache; +import io.dapr.durabletask.implementation.protobuf.HistoryEvents; import io.dapr.durabletask.implementation.protobuf.Orchestration; +import io.dapr.durabletask.implementation.protobuf.OrchestratorActions; import io.dapr.durabletask.implementation.protobuf.OrchestratorService; import io.dapr.durabletask.implementation.protobuf.TaskHubSidecarServiceGrpc; import io.grpc.StatusRuntimeException; @@ -25,6 +28,8 @@ import javax.annotation.Nullable; +import java.util.ArrayList; +import java.util.List; import java.util.logging.Level; import java.util.logging.Logger; @@ -33,6 +38,10 @@ public class OrchestratorRunner extends DurableRunner { private final OrchestratorService.WorkflowRequest orchestratorRequest; private final TaskOrchestrationExecutor taskOrchestrationExecutor; + @Nullable + private final WorkflowHistoryCache historyCache; + @Nullable + private final Runnable teardownStream; /** * Constructs a new instance of the OrchestratorRunner class. @@ -41,26 +50,75 @@ public class OrchestratorRunner extends DurableRunner { * @param taskOrchestrationExecutor The executor responsible for running task orchestration logic. * @param sidecarClient The gRPC stub for communication with the Task Hub sidecar service. * @param tracer An optional tracer used for distributed tracing, can be null. + * @param historyCache The per-stream committed-history cache for stateful-history + * delta work items, or null when the optimization is disabled. */ public OrchestratorRunner( OrchestratorService.WorkItem workItem, TaskOrchestrationExecutor taskOrchestrationExecutor, TaskHubSidecarServiceGrpc.TaskHubSidecarServiceBlockingStub sidecarClient, - @Nullable Tracer tracer) { + @Nullable Tracer tracer, + @Nullable WorkflowHistoryCache historyCache) { + + this(workItem, taskOrchestrationExecutor, sidecarClient, tracer, historyCache, null); + } + + /** + * Constructs a new instance of the OrchestratorRunner class. + * + * @param workItem The work item containing details about the orchestrator task to be executed. + * @param taskOrchestrationExecutor The executor responsible for running task orchestration logic. + * @param sidecarClient The gRPC stub for communication with the Task Hub sidecar service. + * @param tracer An optional tracer used for distributed tracing, can be null. + * @param historyCache The per-stream committed-history cache for stateful-history + * delta work items, or null when the optimization is disabled. + * @param teardownStream Drops the work-item stream this item arrived on, so the sidecar + * cancels and redelivers its pending items. Invoked when the + * history cannot be resolved. May be null. + */ + public OrchestratorRunner( + OrchestratorService.WorkItem workItem, + TaskOrchestrationExecutor taskOrchestrationExecutor, + TaskHubSidecarServiceGrpc.TaskHubSidecarServiceBlockingStub sidecarClient, + @Nullable Tracer tracer, + @Nullable WorkflowHistoryCache historyCache, + @Nullable Runnable teardownStream) { super(workItem, sidecarClient, tracer); this.orchestratorRequest = workItem.getWorkflowRequest(); this.taskOrchestrationExecutor = taskOrchestrationExecutor; + this.historyCache = historyCache; + this.teardownStream = teardownStream; } @Override public void run() { + String instanceId = orchestratorRequest.getInstanceId(); + + List pastEvents; + try { + pastEvents = resolvePastEvents(instanceId); + } catch (StatusRuntimeException e) { + // The cache-miss fallback fetch failed and there is no per-item NACK. Abandon this work item + // rather than completing the turn with an incomplete history, and drop the stream it arrived + // on: the sidecar cancels this stream's pending items and redelivers them promptly, as a + // full-history send on the next (cold) stream. Without the teardown, redelivery would have to + // wait for the work item's lease to lapse. + logException(e); + if (this.teardownStream != null) { + this.teardownStream.run(); + } + return; + } + TaskOrchestratorResult taskOrchestratorResult = taskOrchestrationExecutor.execute( - orchestratorRequest.getPastEventsList(), + pastEvents, orchestratorRequest.getNewEventsList(), orchestratorRequest.hasPropagatedHistory() ? orchestratorRequest.getPropagatedHistory() : null); + updateHistoryCache(instanceId, pastEvents, taskOrchestratorResult); + var versionBuilder = Orchestration.WorkflowVersion.newBuilder(); if (StringUtils.isNotEmpty(taskOrchestratorResult.getVersion())) { @@ -88,4 +146,55 @@ public void run() { this.logException(e); } } + + /** + * Reconstructs the full committed history to replay. For a full send it is simply the request's + * pastEvents; for a delta send (cachedHistory) it is the cached prefix plus the delta, falling + * back to a GetInstanceHistory fetch on any cache miss. + */ + List resolvePastEvents(String instanceId) { + if (this.historyCache == null || !orchestratorRequest.hasCachedHistory()) { + return orchestratorRequest.getPastEventsList(); + } + + List cached = this.historyCache.get(instanceId); + int expected = orchestratorRequest.getCachedHistory().getEventCount(); + if (cached != null && cached.size() == expected) { + List full = + new ArrayList<>(cached.size() + orchestratorRequest.getPastEventsCount()); + full.addAll(cached); + full.addAll(orchestratorRequest.getPastEventsList()); + return full; + } + + // Cache miss: recover the full committed history from the sidecar. NewEvents is applied on + // top of this by the executor, so only the committed past is needed here. + OrchestratorService.GetInstanceHistoryResponse historyResponse = this.sidecarClient.getInstanceHistory( + OrchestratorService.GetInstanceHistoryRequest.newBuilder().setInstanceId(instanceId).build()); + return historyResponse.getEventsList(); + } + + /** + * Refreshes the per-stream cache after a turn so the next turn can be served as a delta. Caches + * only the committed history just replayed (never the not-yet-committed NewEvents), and drops the + * entry once the instance ends. A CompleteWorkflow action covers completed/failed/terminated/ + * continued-as-new; a TerminateWorkflow action targets a different instance and is deliberately + * not treated as a reset. + */ + void updateHistoryCache( + String instanceId, + List pastEvents, + TaskOrchestratorResult result) { + if (this.historyCache == null) { + return; + } + + boolean ended = result.getActions().stream() + .anyMatch(OrchestratorActions.WorkflowAction::hasCompleteWorkflow); + if (ended) { + this.historyCache.remove(instanceId); + } else { + this.historyCache.put(instanceId, pastEvents); + } + } } diff --git a/durabletask-client/src/test/java/io/dapr/durabletask/DurableTaskGrpcWorkerStatefulHistoryTest.java b/durabletask-client/src/test/java/io/dapr/durabletask/DurableTaskGrpcWorkerStatefulHistoryTest.java new file mode 100644 index 0000000000..ca42d5916a --- /dev/null +++ b/durabletask-client/src/test/java/io/dapr/durabletask/DurableTaskGrpcWorkerStatefulHistoryTest.java @@ -0,0 +1,200 @@ +/* + * 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.durabletask; + +import io.dapr.durabletask.implementation.protobuf.OrchestratorService; +import io.dapr.durabletask.implementation.protobuf.TaskHubSidecarServiceGrpc; +import io.grpc.ManagedChannel; +import io.grpc.Server; +import io.grpc.Status; +import io.grpc.inprocess.InProcessChannelBuilder; +import io.grpc.inprocess.InProcessServerBuilder; +import io.grpc.stub.StreamObserver; +import org.junit.jupiter.api.AfterEach; +import org.junit.jupiter.api.Test; + +import java.util.List; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicInteger; +import java.util.concurrent.atomic.AtomicReference; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertFalse; +import static org.junit.jupiter.api.Assertions.assertNotNull; +import static org.junit.jupiter.api.Assertions.assertTrue; + +/** + * Worker-level tests for the stateful-history protocol driven through the real + * {@link DurableTaskGrpcWorker} against an in-process fake sidecar: the capability is advertised by + * default and suppressed when disabled, and a delta work item against a cold cache falls back to + * the GetInstanceHistory RPC. + */ +class DurableTaskGrpcWorkerStatefulHistoryTest { + + private DurableTaskGrpcWorker worker; + private Server server; + private ManagedChannel channel; + + @AfterEach + void tearDown() throws Exception { + if (worker != null) { + worker.close(); + } + if (channel != null) { + channel.shutdownNow().awaitTermination(5, TimeUnit.SECONDS); + } + if (server != null) { + server.shutdownNow().awaitTermination(5, TimeUnit.SECONDS); + } + } + + private void startWorker(TaskHubSidecarServiceGrpc.TaskHubSidecarServiceImplBase service, boolean disableStatefulHistory) + throws Exception { + String serverName = InProcessServerBuilder.generateName(); + server = InProcessServerBuilder.forName(serverName).directExecutor().addService(service).build().start(); + channel = InProcessChannelBuilder.forName(serverName).directExecutor().build(); + worker = new DurableTaskGrpcWorkerBuilder() + .grpcChannel(channel) + .disableStatefulHistory(disableStatefulHistory) + .build(); + worker.start(); + } + + @Test + void advertisesStatefulHistoryCapabilityByDefault() throws Exception { + CountDownLatch latch = new CountDownLatch(1); + AtomicReference> captured = new AtomicReference<>(); + + startWorker(new TaskHubSidecarServiceGrpc.TaskHubSidecarServiceImplBase() { + @Override + public void getWorkItems(OrchestratorService.GetWorkItemsRequest request, + StreamObserver responseObserver) { + captured.compareAndSet(null, request.getCapabilitiesList()); + latch.countDown(); + // Keep the stream open so the worker does not reconnect in a tight loop. + } + }, false); + + assertTrue(latch.await(10, TimeUnit.SECONDS), "worker should have called getWorkItems"); + assertNotNull(captured.get()); + assertTrue(captured.get().contains(OrchestratorService.WorkerCapability.WORKER_CAPABILITY_STATEFUL_HISTORY), + "the worker must advertise WORKER_CAPABILITY_STATEFUL_HISTORY by default"); + } + + @Test + void doesNotAdvertiseCapabilityWhenDisabled() throws Exception { + CountDownLatch latch = new CountDownLatch(1); + AtomicReference> captured = new AtomicReference<>(); + + startWorker(new TaskHubSidecarServiceGrpc.TaskHubSidecarServiceImplBase() { + @Override + public void getWorkItems(OrchestratorService.GetWorkItemsRequest request, + StreamObserver responseObserver) { + captured.compareAndSet(null, request.getCapabilitiesList()); + latch.countDown(); + } + }, true); + + assertTrue(latch.await(10, TimeUnit.SECONDS), "worker should have called getWorkItems"); + assertNotNull(captured.get()); + assertFalse(captured.get().contains(OrchestratorService.WorkerCapability.WORKER_CAPABILITY_STATEFUL_HISTORY), + "a disabled worker must not advertise the capability"); + assertTrue(captured.get().isEmpty()); + } + + @Test + void deltaWorkItemWithColdCacheFallsBackToGetInstanceHistory() throws Exception { + CountDownLatch fetchLatch = new CountDownLatch(1); + AtomicInteger getHistoryCalls = new AtomicInteger(0); + + startWorker(new TaskHubSidecarServiceGrpc.TaskHubSidecarServiceImplBase() { + @Override + public void getWorkItems(OrchestratorService.GetWorkItemsRequest request, + StreamObserver responseObserver) { + // A delta work item (cachedHistory set) for an instance the freshly connected worker holds + // nothing for: its cache is cold, so it must fetch the full history. + OrchestratorService.WorkflowRequest workflowRequest = OrchestratorService.WorkflowRequest.newBuilder() + .setInstanceId("inst-miss") + .setCachedHistory(OrchestratorService.CachedHistory.newBuilder().setEventCount(5)) + .build(); + responseObserver.onNext(OrchestratorService.WorkItem.newBuilder() + .setWorkflowRequest(workflowRequest) + .build()); + // Keep the stream open. + } + + @Override + public void getInstanceHistory(OrchestratorService.GetInstanceHistoryRequest request, + StreamObserver responseObserver) { + getHistoryCalls.incrementAndGet(); + fetchLatch.countDown(); + responseObserver.onNext(OrchestratorService.GetInstanceHistoryResponse.getDefaultInstance()); + responseObserver.onCompleted(); + } + }, false); + + assertTrue(fetchLatch.await(10, TimeUnit.SECONDS), + "a delta work item against a cold cache must trigger a GetInstanceHistory fetch"); + assertEquals(1, getHistoryCalls.get()); + } + + @Test + void failedHistoryFetchDropsTheStreamInsteadOfCompletingTheTurn() throws Exception { + // Two connections: the first serves the doomed delta, the second proves the worker reconnected. + CountDownLatch reconnectLatch = new CountDownLatch(2); + AtomicInteger streams = new AtomicInteger(0); + AtomicInteger completions = new AtomicInteger(0); + + startWorker(new TaskHubSidecarServiceGrpc.TaskHubSidecarServiceImplBase() { + @Override + public void getWorkItems(OrchestratorService.GetWorkItemsRequest request, + StreamObserver responseObserver) { + reconnectLatch.countDown(); + // Only the first stream hands out the work item, otherwise the worker would loop on it. + if (streams.incrementAndGet() != 1) { + return; + } + OrchestratorService.WorkflowRequest workflowRequest = OrchestratorService.WorkflowRequest.newBuilder() + .setInstanceId("inst-unrecoverable") + .setCachedHistory(OrchestratorService.CachedHistory.newBuilder().setEventCount(5)) + .build(); + responseObserver.onNext(OrchestratorService.WorkItem.newBuilder() + .setWorkflowRequest(workflowRequest) + .build()); + } + + @Override + public void getInstanceHistory(OrchestratorService.GetInstanceHistoryRequest request, + StreamObserver responseObserver) { + // The cache-miss recovery is unavailable, so the worker cannot know what to replay. + responseObserver.onError(Status.UNAVAILABLE.withDescription("history unavailable").asRuntimeException()); + } + + @Override + public void completeOrchestratorTask(OrchestratorService.WorkflowResponse request, + StreamObserver responseObserver) { + completions.incrementAndGet(); + responseObserver.onNext(OrchestratorService.CompleteTaskResponse.getDefaultInstance()); + responseObserver.onCompleted(); + } + }, false); + + // The worker sleeps ~5s between reconnects, so allow well past that. + assertTrue(reconnectLatch.await(30, TimeUnit.SECONDS), + "an unresolvable history must drop the stream so the worker reconnects"); + assertEquals(0, completions.get(), + "the turn must be abandoned for redelivery, never completed with an incomplete history"); + } +} diff --git a/durabletask-client/src/test/java/io/dapr/durabletask/IntegrationTestBase.java b/durabletask-client/src/test/java/io/dapr/durabletask/IntegrationTestBase.java index d0a8a8faa6..6877d1ea43 100644 --- a/durabletask-client/src/test/java/io/dapr/durabletask/IntegrationTestBase.java +++ b/durabletask-client/src/test/java/io/dapr/durabletask/IntegrationTestBase.java @@ -14,6 +14,7 @@ package io.dapr.durabletask; import io.dapr.durabletask.orchestration.TaskOrchestrationFactory; +import io.grpc.Channel; import org.junit.jupiter.api.AfterEach; import java.time.Duration; @@ -55,6 +56,20 @@ public TestDurableTaskWorkerBuilder setMaximumTimerInterval(Duration maximumTime return this; } + /** + * Routes the worker over a caller-supplied channel, so a test can attach a + * {@link io.grpc.ClientInterceptor} and observe what the sidecar actually puts on the wire. + */ + public TestDurableTaskWorkerBuilder grpcChannel(Channel channel) { + this.innerBuilder.grpcChannel(channel); + return this; + } + + public TestDurableTaskWorkerBuilder disableStatefulHistory(boolean disableStatefulHistory) { + this.innerBuilder.disableStatefulHistory(disableStatefulHistory); + return this; + } + public TestDurableTaskWorkerBuilder addOrchestrator( String name, TaskOrchestration implementation) { diff --git a/durabletask-client/src/test/java/io/dapr/durabletask/StatefulHistoryIT.java b/durabletask-client/src/test/java/io/dapr/durabletask/StatefulHistoryIT.java new file mode 100644 index 0000000000..f2099f85be --- /dev/null +++ b/durabletask-client/src/test/java/io/dapr/durabletask/StatefulHistoryIT.java @@ -0,0 +1,161 @@ +/* + * 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.durabletask; + +import io.grpc.ManagedChannel; +import io.grpc.ManagedChannelBuilder; +import org.junit.jupiter.api.Tag; +import org.junit.jupiter.api.Test; + +import java.time.Duration; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.TimeoutException; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertNotNull; +import static org.junit.jupiter.api.Assertions.assertTrue; + +/** + * Wire-level verification that the sidecar really delivers history deltas. + * + *

Requires a sidecar implementing the stateful-history protocol (dapr/durabletask-go#110). CI's + * {@code build-durabletask} job builds dapr/durabletask-go from its default branch, which contains + * it. Against an older sidecar the capability is ignored and every turn arrives as a full send, + * which is exactly what {@link #deltaDeliveryReducesFullSends()} is written to catch. + * + *

Asserting on workflow output alone would prove nothing here: a correct delta path and a + * sidecar that never sends deltas produce identical results. The counts come from a gRPC + * interceptor watching the real work-item stream. + */ +@Tag("integration") +class StatefulHistoryIT extends IntegrationTestBase { + + private static final int TURNS = 20; + private static final Duration COMPLETION_TIMEOUT = Duration.ofSeconds(60); + + /** + * The sidecar records how much history a stream holds only after rewriting a work item, + * so the first turn (empty past) leaves the watermark at zero and the second still fails the + * "worker holds something" check. Deltas therefore start at the third turn. dapr's largehistory + * integration test asserts the same bound. + */ + private static final int MAX_WARMUP_FULL_SENDS = 2; + + /** Counts of how one run's work items were delivered, plus the value the workflow returned. */ + private static final class RunResult { + final int deltas; + final int fullSends; + final int historyFetches; + final int output; + + RunResult(int deltas, int fullSends, int historyFetches, int output) { + this.deltas = deltas; + this.fullSends = fullSends; + this.historyFetches = historyFetches; + this.output = output; + } + + @Override + public String toString() { + return String.format("deltas=%d, fullSends=%d, historyFetches=%d, output=%d", + this.deltas, this.fullSends, this.historyFetches, this.output); + } + } + + /** + * Runs a long sequential activity chain, so each activity result is its own turn and the + * committed history grows every turn. That is what makes the omitted prefix, and therefore the + * delta, large enough to be worth measuring. + * + *

Each run gets a fresh worker and channel, hence a fresh work-item stream, so the sidecar's + * warm set starts empty and the counts describe this run alone. + */ + private RunResult runAccumulate(boolean disableStatefulHistory) throws TimeoutException { + final String orchestratorName = "StatefulHistoryAccumulate"; + final String activityName = "PlusOne"; + + WorkItemObserver observer = new WorkItemObserver(); + ManagedChannel channel = ManagedChannelBuilder + .forAddress("127.0.0.1", 4001) + .usePlaintext() + .intercept(observer) + .build(); + + try { + DurableTaskGrpcWorker worker = this.createWorkerBuilder() + .grpcChannel(channel) + .disableStatefulHistory(disableStatefulHistory) + .addOrchestrator(orchestratorName, ctx -> { + int current = ctx.getInput(Integer.class); + for (int i = 0; i < TURNS; i++) { + current = ctx.callActivity(activityName, current, Integer.class).await(); + } + ctx.complete(current); + }) + .addActivity(activityName, ctx -> ctx.getInput(Integer.class) + 1) + .buildAndStart(); + + DurableTaskClient client = new DurableTaskGrpcClientBuilder().build(); + try (worker; client) { + String instanceId = client.scheduleNewOrchestrationInstance(orchestratorName, 0); + OrchestrationMetadata instance = client.waitForInstanceCompletion(instanceId, COMPLETION_TIMEOUT, true); + + assertNotNull(instance); + assertEquals(OrchestrationRuntimeStatus.COMPLETED, instance.getRuntimeStatus()); + + return new RunResult( + observer.deltas(instanceId), + observer.fullSends(instanceId), + observer.historyFetches(instanceId), + instance.readOutputAs(Integer.class)); + } + } finally { + channel.shutdownNow(); + try { + channel.awaitTermination(5, TimeUnit.SECONDS); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + } + } + } + + @Test + void deltaDeliveryReducesFullSends() throws TimeoutException { + RunResult result = runAccumulate(false); + + assertEquals(TURNS, result.output, () -> "workflow produced the wrong result: " + result); + assertTrue(result.deltas > 0, () -> "sidecar never sent a delta: " + result); + assertTrue(result.fullSends <= MAX_WARMUP_FULL_SENDS, () -> "too many full sends: " + result); + assertTrue(result.deltas >= TURNS - MAX_WARMUP_FULL_SENDS, + () -> "expected a delta for nearly every turn: " + result); + } + + @Test + void warmStreamNeverMissesItsCache() throws TimeoutException { + RunResult result = runAccumulate(false); + + assertEquals(TURNS, result.output, () -> "workflow produced the wrong result: " + result); + assertEquals(0, result.historyFetches, () -> "unexpected GetInstanceHistory recovery: " + result); + } + + @Test + void disabledWorkerReceivesOnlyFullHistories() throws TimeoutException { + RunResult result = runAccumulate(true); + + assertEquals(TURNS, result.output, () -> "workflow produced the wrong result: " + result); + assertEquals(0, result.deltas, + () -> "delta sent to a worker that never advertised support: " + result); + assertTrue(result.fullSends >= TURNS, () -> "expected a full send per turn: " + result); + } +} diff --git a/durabletask-client/src/test/java/io/dapr/durabletask/WorkItemObserver.java b/durabletask-client/src/test/java/io/dapr/durabletask/WorkItemObserver.java new file mode 100644 index 0000000000..60eb1a5146 --- /dev/null +++ b/durabletask-client/src/test/java/io/dapr/durabletask/WorkItemObserver.java @@ -0,0 +1,132 @@ +/* + * 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.durabletask; + +import io.dapr.durabletask.implementation.protobuf.OrchestratorService; +import io.grpc.CallOptions; +import io.grpc.Channel; +import io.grpc.ClientCall; +import io.grpc.ClientInterceptor; +import io.grpc.ForwardingClientCall; +import io.grpc.ForwardingClientCallListener; +import io.grpc.Metadata; +import io.grpc.MethodDescriptor; + +import java.util.Map; +import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.atomic.AtomicInteger; + +/** + * Counts how the sidecar delivers workflow history over the wire. + * + *

Mirrors the Go equivalent in dapr's integration framework + * ({@code tests/integration/framework/process/workflow/worker.go}) and the Python SDK's observer: + * every {@code WorkflowRequest} arrives either as a delta ({@code cachedHistory} set, so + * {@code pastEvents} carries only the events since the worker was last brought up to date) or as a + * full send, and every {@code GetInstanceHistory} call is a cache miss the worker had to recover + * from. + * + *

Without this, an end-to-end test cannot tell a working delta path from a sidecar that ignored + * {@code WORKER_CAPABILITY_STATEFUL_HISTORY} altogether: both produce identical workflow output. + * + *

gRPC delivers messages on channel executor threads while the test asserts from the JUnit + * thread, hence the concurrent counters. + */ +final class WorkItemObserver implements ClientInterceptor { + + private static final String GET_WORK_ITEMS = "TaskHubSidecarService/GetWorkItems"; + private static final String GET_INSTANCE_HISTORY = "TaskHubSidecarService/GetInstanceHistory"; + + private final Map deltas = new ConcurrentHashMap<>(); + private final Map fullSends = new ConcurrentHashMap<>(); + private final Map historyFetches = new ConcurrentHashMap<>(); + + /** Work items for this instance whose committed-history prefix the sidecar omitted. */ + int deltas(String instanceId) { + return count(this.deltas, instanceId); + } + + /** Work items for this instance carrying the full committed history. */ + int fullSends(String instanceId) { + return count(this.fullSends, instanceId); + } + + /** GetInstanceHistory calls for this instance, i.e. cache misses the worker recovered from. */ + int historyFetches(String instanceId) { + return count(this.historyFetches, instanceId); + } + + private static int count(Map counters, String instanceId) { + AtomicInteger counter = counters.get(instanceId); + return counter == null ? 0 : counter.get(); + } + + private static void increment(Map counters, String instanceId) { + counters.computeIfAbsent(instanceId, key -> new AtomicInteger()).incrementAndGet(); + } + + private void recordWorkItem(OrchestratorService.WorkItem workItem) { + if (!workItem.hasWorkflowRequest()) { + return; + } + OrchestratorService.WorkflowRequest request = workItem.getWorkflowRequest(); + if (request.hasCachedHistory()) { + increment(this.deltas, request.getInstanceId()); + } else { + increment(this.fullSends, request.getInstanceId()); + } + } + + @Override + public ClientCall interceptCall( + MethodDescriptor method, CallOptions callOptions, Channel next) { + + String fullMethodName = method.getFullMethodName(); + + if (fullMethodName.endsWith(GET_INSTANCE_HISTORY)) { + return new ForwardingClientCall.SimpleForwardingClientCall( + next.newCall(method, callOptions)) { + @Override + public void sendMessage(ReqT message) { + if (message instanceof OrchestratorService.GetInstanceHistoryRequest) { + increment(historyFetches, ((OrchestratorService.GetInstanceHistoryRequest) message).getInstanceId()); + } + super.sendMessage(message); + } + }; + } + + if (!fullMethodName.endsWith(GET_WORK_ITEMS)) { + return next.newCall(method, callOptions); + } + + return new ForwardingClientCall.SimpleForwardingClientCall( + next.newCall(method, callOptions)) { + @Override + public void start(Listener responseListener, Metadata headers) { + super.start( + new ForwardingClientCallListener.SimpleForwardingClientCallListener(responseListener) { + @Override + public void onMessage(RespT message) { + if (message instanceof OrchestratorService.WorkItem) { + recordWorkItem((OrchestratorService.WorkItem) message); + } + super.onMessage(message); + } + }, + headers); + } + }; + } +} diff --git a/durabletask-client/src/test/java/io/dapr/durabletask/WorkflowHistoryCacheTest.java b/durabletask-client/src/test/java/io/dapr/durabletask/WorkflowHistoryCacheTest.java new file mode 100644 index 0000000000..b027d51c9a --- /dev/null +++ b/durabletask-client/src/test/java/io/dapr/durabletask/WorkflowHistoryCacheTest.java @@ -0,0 +1,180 @@ +/* + * 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.durabletask; + +import io.dapr.durabletask.implementation.protobuf.HistoryEvents; +import org.junit.jupiter.api.Test; + +import java.time.Duration; +import java.util.ArrayList; +import java.util.List; +import java.util.concurrent.atomic.AtomicLong; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertNotNull; +import static org.junit.jupiter.api.Assertions.assertNull; +import static org.junit.jupiter.api.Assertions.assertThrows; +import static org.junit.jupiter.api.Assertions.assertTrue; + +/** + * Unit tests for the worker's stateful-history cache bounds. These mirror the Go reference + * (durabletask-go client/worker_history_test.go) and the Python/.NET SDKs: a sliding TTL, an + * instance-count cap, and a byte budget, all with least-recently-used eviction. + */ +class WorkflowHistoryCacheTest { + + /** Events with non-zero serialized size (eventId 0 is the proto default, which is 0 bytes). */ + private static List events(int count) { + List list = new ArrayList<>(count); + for (int i = 0; i < count; i++) { + list.add(HistoryEvents.HistoryEvent.newBuilder().setEventId(i + 1).build()); + } + return list; + } + + private static long bytesOf(int count) { + long total = 0; + for (HistoryEvents.HistoryEvent event : events(count)) { + total += event.getSerializedSize(); + } + return total; + } + + @Test + void getPutRemoveReset() { + WorkflowHistoryCache cache = new WorkflowHistoryCache(null, 0, 0); + + assertNull(cache.get("a")); + + cache.put("a", events(3)); + assertNotNull(cache.get("a")); + assertEquals(3, cache.get("a").size()); + + cache.remove("a"); + assertNull(cache.get("a")); + + cache.put("b", events(1)); + cache.reset(); + assertNull(cache.get("b")); + } + + @Test + void countCapEvictsLeastRecentlyUsed() { + AtomicLong clock = new AtomicLong(0); + WorkflowHistoryCache cache = new WorkflowHistoryCache(null, 2, 0, clock::get); + + cache.put("a", events(1)); + clock.incrementAndGet(); + cache.put("b", events(1)); + clock.incrementAndGet(); + cache.put("c", events(1)); // over the cap, evicts the LRU entry ("a") + + assertNull(cache.get("a")); + assertNotNull(cache.get("b")); + assertNotNull(cache.get("c")); + } + + @Test + void byteCapEvictsLeastRecentlyUsed() { + long entryBytes = bytesOf(4); + assertTrue(entryBytes > 0); + AtomicLong clock = new AtomicLong(0); + WorkflowHistoryCache cache = new WorkflowHistoryCache(null, 0, entryBytes + 1, clock::get); + + cache.put("a", events(4)); + clock.incrementAndGet(); + cache.put("b", events(4)); // two entries exceed the byte budget, evicts the LRU entry ("a") + + assertNull(cache.get("a")); + assertNotNull(cache.get("b")); + assertTrue(cache.totalBytes() <= entryBytes + 1); + } + + @Test + void singleOversizedEntryIsKept() { + WorkflowHistoryCache cache = new WorkflowHistoryCache(null, 0, 1); + cache.put("big", events(5)); + assertNotNull(cache.get("big")); + } + + @Test + void byteAccountingTracksReplaceAndRemove() { + WorkflowHistoryCache cache = new WorkflowHistoryCache(null, 0, 0); + + cache.put("a", events(3)); + cache.put("b", events(2)); + assertEquals(bytesOf(3) + bytesOf(2), cache.totalBytes()); + + cache.put("a", events(6)); // replace adjusts the running total to the new size + assertEquals(bytesOf(6) + bytesOf(2), cache.totalBytes()); + + cache.remove("a"); + assertEquals(bytesOf(2), cache.totalBytes()); + + cache.reset(); + assertEquals(0, cache.totalBytes()); + } + + @Test + void ttlSweepIsSliding() { + AtomicLong clock = new AtomicLong(0); + WorkflowHistoryCache cache = new WorkflowHistoryCache(Duration.ofSeconds(60), 0, 0, clock::get); + + cache.put("idle", events(2)); + cache.put("active", events(2)); + + clock.set(Duration.ofSeconds(120).toNanos()); // past the TTL... + assertNotNull(cache.get("active")); // ...but a turn refreshes "active" + + cache.sweepExpired(); + assertNull(cache.get("idle")); + assertNotNull(cache.get("active")); + } + + @Test + void nonPositiveConfigUsesDefaults() { + // ttl/maxInstances fall back to their (large) defaults; maxBytes becomes unlimited. None of + // these should evict the three modest entries below. + WorkflowHistoryCache cache = new WorkflowHistoryCache(Duration.ZERO, -1, -5); + + cache.put("a", events(1)); + cache.put("b", events(1)); + cache.put("c", events(1)); + + assertEquals(3, cache.size()); + } + + @Test + void getReturnsUnmodifiableViewSoCallersCannotCorruptTheCache() { + WorkflowHistoryCache cache = new WorkflowHistoryCache(null, 0, 0); + cache.put("a", events(3)); + + List cached = cache.get("a"); + + assertThrows(UnsupportedOperationException.class, + () -> cached.add(HistoryEvents.HistoryEvent.newBuilder().setEventId(99).build())); + assertEquals(3, cache.get("a").size()); + } + + @Test + void putSnapshotsSoLaterCallerMutationIsNotObserved() { + WorkflowHistoryCache cache = new WorkflowHistoryCache(null, 0, 0); + List source = new ArrayList<>(events(3)); + + cache.put("a", source); + source.add(HistoryEvents.HistoryEvent.newBuilder().setEventId(99).build()); + + assertEquals(3, cache.get("a").size()); + } +} diff --git a/durabletask-client/src/test/java/io/dapr/durabletask/runner/OrchestratorRunnerHistoryTest.java b/durabletask-client/src/test/java/io/dapr/durabletask/runner/OrchestratorRunnerHistoryTest.java new file mode 100644 index 0000000000..6407ac71ee --- /dev/null +++ b/durabletask-client/src/test/java/io/dapr/durabletask/runner/OrchestratorRunnerHistoryTest.java @@ -0,0 +1,128 @@ +/* + * 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.durabletask.runner; + +import io.dapr.durabletask.TaskOrchestratorResult; +import io.dapr.durabletask.WorkflowHistoryCache; +import io.dapr.durabletask.implementation.protobuf.HistoryEvents; +import io.dapr.durabletask.implementation.protobuf.OrchestratorActions; +import io.dapr.durabletask.implementation.protobuf.OrchestratorService; +import org.junit.jupiter.api.Test; + +import java.util.ArrayList; +import java.util.List; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertNotNull; +import static org.junit.jupiter.api.Assertions.assertNull; + +/** + * Deterministic tests for the worker's history resolution and cache update, driving + * {@link OrchestratorRunner}'s package-visible helpers directly. The full-send and cache-hit paths + * make no RPC, so a null sidecar client suffices; the cache-miss fallback (which does call + * GetInstanceHistory) is covered by the worker-level in-process test. + */ +class OrchestratorRunnerHistoryTest { + + private static List events(int count) { + List list = new ArrayList<>(count); + for (int i = 0; i < count; i++) { + list.add(HistoryEvents.HistoryEvent.newBuilder().setEventId(i + 1).build()); + } + return list; + } + + private static OrchestratorRunner runner(OrchestratorService.WorkflowRequest request, WorkflowHistoryCache cache) { + OrchestratorService.WorkItem workItem = OrchestratorService.WorkItem.newBuilder() + .setWorkflowRequest(request) + .build(); + return new OrchestratorRunner(workItem, null, null, null, cache); + } + + private static TaskOrchestratorResult resultWith(OrchestratorActions.WorkflowAction action) { + return new TaskOrchestratorResult(List.of(action), "", null, null); + } + + @Test + void fullSendReturnsRequestPastEvents() { + OrchestratorService.WorkflowRequest request = OrchestratorService.WorkflowRequest.newBuilder() + .setInstanceId("a") + .addAllPastEvents(events(4)) + .build(); + OrchestratorRunner runner = runner(request, new WorkflowHistoryCache(null, 0, 0)); + + assertEquals(4, runner.resolvePastEvents("a").size()); + } + + @Test + void cacheHitReconstructsPrefixPlusDelta() { + WorkflowHistoryCache cache = new WorkflowHistoryCache(null, 0, 0); + cache.put("a", events(5)); + OrchestratorService.WorkflowRequest request = OrchestratorService.WorkflowRequest.newBuilder() + .setInstanceId("a") + .addAllPastEvents(events(3)) + .setCachedHistory(OrchestratorService.CachedHistory.newBuilder().setEventCount(5)) + .build(); + OrchestratorRunner runner = runner(request, cache); + + assertEquals(8, runner.resolvePastEvents("a").size()); // 5 cached prefix + 3 delta + } + + @Test + void disabledCacheReturnsRequestPastEvents() { + OrchestratorService.WorkflowRequest request = OrchestratorService.WorkflowRequest.newBuilder() + .setInstanceId("a") + .addAllPastEvents(events(4)) + .setCachedHistory(OrchestratorService.CachedHistory.newBuilder().setEventCount(2)) + .build(); + OrchestratorRunner runner = runner(request, null); // stateful history disabled + + assertEquals(4, runner.resolvePastEvents("a").size()); + } + + @Test + void updateCachePutsWhenRunningThenEvictsOnComplete() { + WorkflowHistoryCache cache = new WorkflowHistoryCache(null, 0, 0); + OrchestratorService.WorkflowRequest request = OrchestratorService.WorkflowRequest.newBuilder() + .setInstanceId("a") + .build(); + OrchestratorRunner runner = runner(request, cache); + + OrchestratorActions.WorkflowAction running = OrchestratorActions.WorkflowAction.newBuilder() + .setScheduleTask(OrchestratorActions.ScheduleTaskAction.newBuilder()) + .build(); + runner.updateHistoryCache("a", events(6), resultWith(running)); + assertNotNull(cache.get("a")); + + OrchestratorActions.WorkflowAction completed = OrchestratorActions.WorkflowAction.newBuilder() + .setCompleteWorkflow(OrchestratorActions.CompleteWorkflowAction.newBuilder()) + .build(); + runner.updateHistoryCache("a", events(6), resultWith(completed)); + assertNull(cache.get("a")); + } + + @Test + void updateCacheSkippedWhenDisabled() { + OrchestratorService.WorkflowRequest request = OrchestratorService.WorkflowRequest.newBuilder() + .setInstanceId("a") + .build(); + OrchestratorRunner runner = runner(request, null); // no cache + + OrchestratorActions.WorkflowAction running = OrchestratorActions.WorkflowAction.newBuilder() + .setScheduleTask(OrchestratorActions.ScheduleTaskAction.newBuilder()) + .build(); + // Must not throw when the cache is disabled (null). + runner.updateHistoryCache("a", events(6), resultWith(running)); + } +} diff --git a/sdk-workflows/src/main/java/io/dapr/workflows/runtime/WorkflowRuntimeBuilder.java b/sdk-workflows/src/main/java/io/dapr/workflows/runtime/WorkflowRuntimeBuilder.java index e812e55aad..dc298e9cf4 100644 --- a/sdk-workflows/src/main/java/io/dapr/workflows/runtime/WorkflowRuntimeBuilder.java +++ b/sdk-workflows/src/main/java/io/dapr/workflows/runtime/WorkflowRuntimeBuilder.java @@ -28,6 +28,7 @@ import org.slf4j.Logger; import org.slf4j.LoggerFactory; +import java.time.Duration; import java.util.Collections; import java.util.HashSet; import java.util.Set; @@ -116,6 +117,62 @@ public WorkflowRuntimeBuilder withExecutorService(ExecutorService executorServic return this; } + /** + * Disables the stateful-history optimization. + * + *

By default the worker advertises {@code WORKER_CAPABILITY_STATEFUL_HISTORY} and caches each + * instance's committed history per work-item stream, so the sidecar can send only the new events + * (the delta) each turn instead of the full history. A cache miss is always recovered safely via + * the GetInstanceHistory RPC, so disabling this only affects per-turn bandwidth, never + * correctness. When disabled, the worker always receives the full history.

+ * + * @param disableStatefulHistory whether to disable the stateful-history optimization + * @return {@link WorkflowRuntimeBuilder}. + */ + public WorkflowRuntimeBuilder withStatefulHistoryDisabled(boolean disableStatefulHistory) { + this.builder.disableStatefulHistory(disableStatefulHistory); + return this; + } + + /** + * Sets the sliding time-to-live for cached instance histories. An instance's entry is reclaimed + * once it has gone idle (no turn) for longer than this. If not specified, a default of one hour + * is used. Ignored when the stateful-history optimization is disabled. + * + * @param historyCacheTtl the sliding time-to-live for a cached instance history + * @return {@link WorkflowRuntimeBuilder}. + */ + public WorkflowRuntimeBuilder withHistoryCacheTtl(Duration historyCacheTtl) { + this.builder.historyCacheTtl(historyCacheTtl); + return this; + } + + /** + * Sets the maximum number of per-instance histories retained on a single work-item stream; + * least-recently-used entries are evicted beyond it. A non-positive value uses the built-in + * default. Ignored when the stateful-history optimization is disabled. + * + * @param historyCacheMaxInstances the instance-count cap for the history cache + * @return {@link WorkflowRuntimeBuilder}. + */ + public WorkflowRuntimeBuilder withHistoryCacheMaxInstances(int historyCacheMaxInstances) { + this.builder.historyCacheMaxInstances(historyCacheMaxInstances); + return this; + } + + /** + * Sets the byte budget for cached histories on a single work-item stream; least-recently-used + * entries are evicted beyond it. A non-positive value means unlimited (bounded only by the + * instance-count cap and the TTL). Ignored when the stateful-history optimization is disabled. + * + * @param historyCacheMaxBytes the byte budget for the history cache + * @return {@link WorkflowRuntimeBuilder}. + */ + public WorkflowRuntimeBuilder withHistoryCacheMaxBytes(long historyCacheMaxBytes) { + this.builder.historyCacheMaxBytes(historyCacheMaxBytes); + return this; + } + /** * Registers a Workflow object. * diff --git a/sdk-workflows/src/test/java/io/dapr/workflows/runtime/WorkflowRuntimeBuilderTest.java b/sdk-workflows/src/test/java/io/dapr/workflows/runtime/WorkflowRuntimeBuilderTest.java index fdb5be655e..ee50b447a9 100644 --- a/sdk-workflows/src/test/java/io/dapr/workflows/runtime/WorkflowRuntimeBuilderTest.java +++ b/sdk-workflows/src/test/java/io/dapr/workflows/runtime/WorkflowRuntimeBuilderTest.java @@ -12,6 +12,7 @@ */ package io.dapr.workflows.runtime; +import io.dapr.durabletask.DurableTaskGrpcWorkerBuilder; import io.dapr.durabletask.TaskActivity; import io.dapr.durabletask.TaskActivityFactory; import io.dapr.durabletask.TaskOrchestration; @@ -26,6 +27,8 @@ import java.io.ByteArrayOutputStream; import java.io.PrintStream; +import java.lang.reflect.Field; +import java.time.Duration; import static org.junit.jupiter.api.Assertions.assertDoesNotThrow; import static org.mockito.ArgumentMatchers.eq; @@ -195,4 +198,44 @@ public void buildTest() { } }); } + + /** + * Reads a field off the wrapped {@link DurableTaskGrpcWorkerBuilder}. The stateful-history + * options are package-visible on the durabletask builder, so a test in this package cannot + * observe them any other way. + */ + private static Object innerBuilderField(WorkflowRuntimeBuilder runtimeBuilder, String name) + throws Exception { + Field builderField = WorkflowRuntimeBuilder.class.getDeclaredField("builder"); + builderField.setAccessible(true); + Object innerBuilder = builderField.get(runtimeBuilder); + + Field target = innerBuilder.getClass().getDeclaredField(name); + target.setAccessible(true); + return target.get(innerBuilder); + } + + @Test + public void statefulHistoryIsEnabledByDefault() throws Exception { + var runtimeBuilder = new WorkflowRuntimeBuilder(); + + Assertions.assertEquals(false, innerBuilderField(runtimeBuilder, "disableStatefulHistory")); + } + + @Test + public void statefulHistoryOptionsAreForwardedToTheWorkerBuilder() throws Exception { + var runtimeBuilder = new WorkflowRuntimeBuilder(); + + var returned = runtimeBuilder + .withStatefulHistoryDisabled(true) + .withHistoryCacheTtl(Duration.ofMinutes(2)) + .withHistoryCacheMaxInstances(50) + .withHistoryCacheMaxBytes(4096L); + + Assertions.assertSame(runtimeBuilder, returned); + Assertions.assertEquals(true, innerBuilderField(runtimeBuilder, "disableStatefulHistory")); + Assertions.assertEquals(Duration.ofMinutes(2), innerBuilderField(runtimeBuilder, "historyCacheTtl")); + Assertions.assertEquals(50, innerBuilderField(runtimeBuilder, "historyCacheMaxInstances")); + Assertions.assertEquals(4096L, innerBuilderField(runtimeBuilder, "historyCacheMaxBytes")); + } } From df737c93f47ba4a0b8c14c043a55e6f90c952c01 Mon Sep 17 00:00:00 2001 From: joshvanl Date: Thu, 6 Aug 2026 13:03:30 -0300 Subject: [PATCH 2/2] Review comments Signed-off-by: joshvanl --- .../io/dapr/durabletask/DurableTaskGrpcWorker.java | 5 ++--- .../io/dapr/durabletask/WorkflowHistoryCache.java | 12 +++++++----- 2 files changed, 9 insertions(+), 8 deletions(-) diff --git a/durabletask-client/src/main/java/io/dapr/durabletask/DurableTaskGrpcWorker.java b/durabletask-client/src/main/java/io/dapr/durabletask/DurableTaskGrpcWorker.java index bd1dbe4a3b..88672d6811 100644 --- a/durabletask-client/src/main/java/io/dapr/durabletask/DurableTaskGrpcWorker.java +++ b/durabletask-client/src/main/java/io/dapr/durabletask/DurableTaskGrpcWorker.java @@ -230,9 +230,8 @@ public void startAndBlock() { if (requestType == OrchestratorService.WorkItem.RequestCase.WORKFLOWREQUEST) { OrchestratorService.WorkflowRequest orchestratorRequest = workItem.getWorkflowRequest(); - logger.log(Level.FINEST, - String.format("Processing orchestrator request for instance: {0}", - orchestratorRequest.getInstanceId())); + logger.log(Level.FINEST, "Processing orchestrator request for instance: {0}", + orchestratorRequest.getInstanceId()); this.workerPool.submit( new OrchestratorRunner(workItem, taskOrchestrationExecutor, sidecarClient, tracer, diff --git a/durabletask-client/src/main/java/io/dapr/durabletask/WorkflowHistoryCache.java b/durabletask-client/src/main/java/io/dapr/durabletask/WorkflowHistoryCache.java index 515d469566..99fe027c4b 100644 --- a/durabletask-client/src/main/java/io/dapr/durabletask/WorkflowHistoryCache.java +++ b/durabletask-client/src/main/java/io/dapr/durabletask/WorkflowHistoryCache.java @@ -38,6 +38,7 @@ public final class WorkflowHistoryCache { static final int DEFAULT_MAX_INSTANCES = 100_000; private static final class Entry { + /** Already wrapped unmodifiable, so {@link #get(String)} hands it out without allocating. */ final List events; final long bytes; long lastAccess; @@ -86,9 +87,8 @@ public WorkflowHistoryCache(Duration ttl, int maxInstances, long maxBytes) { * Returns the cached committed history for an instance, refreshing its TTL, or {@code null} on a * miss. * - *

The returned list is an unmodifiable view of the cached entry rather than a copy, so reading - * it stays allocation-free on the hot path while a caller cannot corrupt the cache by mutating - * what it was handed. + *

The returned list is the entry's unmodifiable view, wrapped once when it was cached, so a + * caller cannot corrupt the cache by mutating what it was handed and this path allocates nothing. * * @param instanceId the workflow instance ID * @return the cached committed history, or {@code null} if the instance is not cached @@ -100,7 +100,7 @@ public List get(String instanceId) { return null; } entry.lastAccess = this.clockNanos.getAsLong(); - return Collections.unmodifiableList(entry.events); + return entry.events; } } @@ -112,7 +112,9 @@ public List get(String instanceId) { * @param events the committed history to cache for the instance */ public void put(String instanceId, List events) { - List snapshot = new ArrayList<>(events); + // Snapshot so a later mutation by the caller is not observed, and wrap once here rather than on + // every get(), which is the hot path. + List snapshot = Collections.unmodifiableList(new ArrayList<>(events)); long bytes = 0; for (HistoryEvents.HistoryEvent event : snapshot) { bytes += event.getSerializedSize();