Skip to content

Add stateful-history delta work items to the workflow worker #1777

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
JoshVanL wants to merge 2 commits into
base: master
Choose a base branch
from
Open

Add stateful-history delta work items to the workflow worker #1777

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
2ac527f
Add stateful-history delta work items to the workflow worker
JoshVanL Jul 1, 2026
df737c9
Review comments
JoshVanL Aug 6, 2026
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
136 changes: 104 additions & 32 deletions durabletask-client/src/main/java/io/dapr/durabletask/DurableTaskGrpcWorker.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand Down Expand Up @@ -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;

Expand Down Expand Up @@ -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);
}

/**
Expand Down Expand Up @@ -137,6 +146,7 @@ public void close() {
this.workerThread.interrupt();
}
this.isNormalShutdown = true;
this.shutDownHistoryJanitor();
this.shutDownWorkerPool();
this.closeSideCarChannel();
}
Expand Down Expand Up @@ -172,41 +182,81 @@ public void startAndBlock() {
this.dataConverter,
logger);

this.startHistoryJanitor();

while (!this.isNormalShutdown && !Thread.currentThread().isInterrupted()) {
try {
OrchestratorService.GetWorkItemsRequest getWorkItemsRequest = OrchestratorService.GetWorkItemsRequest
.newBuilder().build();
Iterator<OrchestratorService.WorkItem> 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<OrchestratorService.WorkItem> 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, "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) {
Expand Down Expand Up @@ -264,6 +314,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) {
Expand Down
60 changes: 60 additions & 0 deletions durabletask-client/src/main/java/io/dapr/durabletask/DurableTaskGrpcWorkerBuilder.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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}.
Expand Down Expand Up @@ -145,6 +149,62 @@ public DurableTaskGrpcWorkerBuilder appId(String appId) {
return this;
}

/**
* Disables the stateful-history optimization.
*
* <p>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.</p>
*
* @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.
*
Expand Down
Loading
Loading