From 024b00bf955290d121530ec55b1adc1f0a59b9e4 Mon Sep 17 00:00:00 2001 From: Spencer Judge Date: Fri, 10 Jul 2026 15:10:53 -0700 Subject: [PATCH 1/3] Add patch activation callback --- CHANGELOG.md | 4 + README.md | 4 +- src/Temporalio/Worker/PatchActivationInput.cs | 18 ++ src/Temporalio/Worker/TemporalWorker.cs | 1 + .../Worker/TemporalWorkerOptions.cs | 12 + src/Temporalio/Worker/WorkflowInstance.cs | 14 +- .../Worker/WorkflowInstanceDetails.cs | 2 + src/Temporalio/Worker/WorkflowReplayer.cs | 1 + src/Temporalio/Worker/WorkflowWorker.cs | 1 + .../Worker/WorkflowWorkerOptions.cs | 1 + .../Worker/WorkflowWorkerTests.cs | 281 ++++++++++++++++++ 11 files changed, 337 insertions(+), 2 deletions(-) create mode 100644 src/Temporalio/Worker/PatchActivationInput.cs diff --git a/CHANGELOG.md b/CHANGELOG.md index 6f7ac2b3..cce69425 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -32,6 +32,10 @@ to docs, or any other relevant information. connections now compress outbound requests and accept gzip-compressed responses by default. If the remote service does not support gzip compression, the connection is downgraded to uncompressed requests. Set it to `GrpcCompression.None` to opt out. +- Added AWS Lambda worker support packages, including OpenTelemetry helpers for Lambda workers. +- Added the experimental `TemporalWorkerOptions.PatchActivationCallback`, allowing workers to + decide whether a first non-replay `Workflow.Patched` call should activate a patch during rolling + deployments. ### Fixed diff --git a/README.md b/README.md index 9633e3cf..5402f65f 100644 --- a/README.md +++ b/README.md @@ -681,7 +681,9 @@ can be used from workflows including: workflow calls elsewhere. * `GetExternalWorkflowHandle` - Get a handle to an external workflow to issue cancellation requests and signals. * `NewGuid` - Create a deterministically random UUIDv4 GUID. - * `Patched` and `DeprecatePatch` - Support for patch-based versioning inside the workflow. + * `Patched` and `DeprecatePatch` - Support for patch-based versioning inside the workflow. Workers + can set `PatchActivationCallback` to decide whether a newly introduced patch should activate + during rolling deployments. * `UpsertMemo` - Update the memo values for the workflow. * `UpsertTypedSearchAttributes` - Update the search attributes for the workflow. diff --git a/src/Temporalio/Worker/PatchActivationInput.cs b/src/Temporalio/Worker/PatchActivationInput.cs new file mode 100644 index 00000000..d0487607 --- /dev/null +++ b/src/Temporalio/Worker/PatchActivationInput.cs @@ -0,0 +1,18 @@ +using Temporalio.Workflows; + +namespace Temporalio.Worker +{ + /// + /// Input for . + /// + /// Information about the workflow execution calling + /// . + /// Patch ID passed to . + /// + /// WARNING: This constructor may have required properties added. Do not rely on the exact + /// constructor, only use "with" clauses. + /// + public record PatchActivationInput( + WorkflowInfo WorkflowInfo, + string PatchId); +} diff --git a/src/Temporalio/Worker/TemporalWorker.cs b/src/Temporalio/Worker/TemporalWorker.cs index deea8248..4e909586 100644 --- a/src/Temporalio/Worker/TemporalWorker.cs +++ b/src/Temporalio/Worker/TemporalWorker.cs @@ -121,6 +121,7 @@ public TemporalWorker(IWorkerClient client, TemporalWorkerOptions options) RuntimeMetricMeter: MetricMeter, WorkerLevelFailureExceptionTypes: options.WorkflowFailureExceptionTypes, DisableEagerActivityExecution: options.DisableEagerActivityExecution, + PatchActivationCallback: options.PatchActivationCallback, AssertValidLocalActivity: (activityType) => (activityWorker ?? throw new InvalidOperationException( $"Activity {activityType} is not registered on this worker," + diff --git a/src/Temporalio/Worker/TemporalWorkerOptions.cs b/src/Temporalio/Worker/TemporalWorkerOptions.cs index e3953a91..ca990fc5 100644 --- a/src/Temporalio/Worker/TemporalWorkerOptions.cs +++ b/src/Temporalio/Worker/TemporalWorkerOptions.cs @@ -288,6 +288,18 @@ public TemporalWorkerOptions() /// public IReadOnlyCollection? WorkflowFailureExceptionTypes { get; set; } + /// + /// Gets or sets the callback that decides whether the first non-replay call to + /// for a patch ID should activate that patch. + /// + /// + /// The callback is only invoked for a newly encountered patch. Existing history markers, + /// replay, and bypass the callback. Returning false + /// leaves the patch inactive and does not record a patch marker. + /// + /// WARNING: This property is experimental and may change in the future. + public Func? PatchActivationCallback { get; set; } + /// /// Gets or sets a value indicating whether deadlock detection will be disabled for all /// workflows. If unset, this value defaults to true only if diff --git a/src/Temporalio/Worker/WorkflowInstance.cs b/src/Temporalio/Worker/WorkflowInstance.cs index c0c1072e..6ba916a9 100644 --- a/src/Temporalio/Worker/WorkflowInstance.cs +++ b/src/Temporalio/Worker/WorkflowInstance.cs @@ -76,6 +76,7 @@ internal class WorkflowInstance : TaskScheduler, IWorkflowInstance, IWorkflowCon private readonly Action onTaskCompleted; private readonly IReadOnlyCollection? workerLevelFailureExceptionTypes; private readonly bool disableEagerActivityExecution; + private readonly Func? patchActivationCallback; private readonly Handlers inProgressHandlers = new(); private readonly WorkflowDefinitionOptions definitionOptions; private WorkflowActivationCompletion? completion; @@ -221,6 +222,7 @@ public WorkflowInstance(WorkflowInstanceDetails details) TracingEventsEnabled = !details.DisableTracingEvents; workerLevelFailureExceptionTypes = details.WorkerLevelFailureExceptionTypes; disableEagerActivityExecution = details.DisableEagerActivityExecution; + patchActivationCallback = details.PatchActivationCallback; AssertValidLocalActivity = details.AssertValidLocalActivity; definitionOptions = new() { @@ -450,7 +452,17 @@ public bool Patch(string patchId, bool deprecated) { return patched; } - patched = !IsReplaying || patchesNotified.Contains(patchId); + // Replay and history markers already determine the branch, and deprecation must keep + // existing patch semantics, so only a genuinely new patch consults the callback. + if (!deprecated && !IsReplaying && !patchesNotified.Contains(patchId) && + patchActivationCallback != null) + { + patched = patchActivationCallback(new(Info, patchId)); + } + else + { + patched = !IsReplaying || patchesNotified.Contains(patchId); + } patchesMemoized[patchId] = patched; if (patched) { diff --git a/src/Temporalio/Worker/WorkflowInstanceDetails.cs b/src/Temporalio/Worker/WorkflowInstanceDetails.cs index a2c94522..59b5938e 100644 --- a/src/Temporalio/Worker/WorkflowInstanceDetails.cs +++ b/src/Temporalio/Worker/WorkflowInstanceDetails.cs @@ -29,6 +29,7 @@ namespace Temporalio.Worker /// Lazy runtime-level metric meter. /// Failure exception types at worker level. /// Whether to disable eager at the worker level. + /// Callback for deciding whether to activate a patch. /// Checks the validity of the local activity and throws if it is invalid. internal record WorkflowInstanceDetails( string Namespace, @@ -49,5 +50,6 @@ internal record WorkflowInstanceDetails( Lazy RuntimeMetricMeter, IReadOnlyCollection? WorkerLevelFailureExceptionTypes, bool DisableEagerActivityExecution, + Func? PatchActivationCallback, Action AssertValidLocalActivity); } diff --git a/src/Temporalio/Worker/WorkflowReplayer.cs b/src/Temporalio/Worker/WorkflowReplayer.cs index 723719a5..d3f27632 100644 --- a/src/Temporalio/Worker/WorkflowReplayer.cs +++ b/src/Temporalio/Worker/WorkflowReplayer.cs @@ -230,6 +230,7 @@ public WorkflowHistoryRunner(WorkflowReplayerOptions options, bool throwOnReplay RuntimeMetricMeter: new(() => runtime.MetricMeter), WorkerLevelFailureExceptionTypes: options.WorkflowFailureExceptionTypes, DisableEagerActivityExecution: false, + PatchActivationCallback: null, AssertValidLocalActivity: _ => { }, DefaultVersioningBehavior: null, DeploymentOptions: null), diff --git a/src/Temporalio/Worker/WorkflowWorker.cs b/src/Temporalio/Worker/WorkflowWorker.cs index 11d6a8e7..4eee2d4c 100644 --- a/src/Temporalio/Worker/WorkflowWorker.cs +++ b/src/Temporalio/Worker/WorkflowWorker.cs @@ -363,6 +363,7 @@ private IWorkflowInstance CreateInstance( RuntimeMetricMeter: options.RuntimeMetricMeter, WorkerLevelFailureExceptionTypes: options.WorkerLevelFailureExceptionTypes, DisableEagerActivityExecution: options.DisableEagerActivityExecution, + PatchActivationCallback: options.PatchActivationCallback, AssertValidLocalActivity: options.AssertValidLocalActivity)); } } diff --git a/src/Temporalio/Worker/WorkflowWorkerOptions.cs b/src/Temporalio/Worker/WorkflowWorkerOptions.cs index 71d6701c..fe78b52e 100644 --- a/src/Temporalio/Worker/WorkflowWorkerOptions.cs +++ b/src/Temporalio/Worker/WorkflowWorkerOptions.cs @@ -23,6 +23,7 @@ internal record WorkflowWorkerOptions( Lazy RuntimeMetricMeter, IReadOnlyCollection? WorkerLevelFailureExceptionTypes, bool DisableEagerActivityExecution, + Func? PatchActivationCallback, Action AssertValidLocalActivity, VersioningBehavior? DefaultVersioningBehavior, WorkerDeploymentOptions? DeploymentOptions); diff --git a/tests/Temporalio.Tests/Worker/WorkflowWorkerTests.cs b/tests/Temporalio.Tests/Worker/WorkflowWorkerTests.cs index 020490ee..d840a941 100644 --- a/tests/Temporalio.Tests/Worker/WorkflowWorkerTests.cs +++ b/tests/Temporalio.Tests/Worker/WorkflowWorkerTests.cs @@ -2949,6 +2949,287 @@ await ExecuteWorkerAsync( new TemporalWorkerOptions().AddActivity(PatchSearchAttributeWorkflow.SomeActivity)); } + [Workflow] + public class PatchActivationWorkflow + { + private bool released; + private bool firstPatchResult; + + [WorkflowRun] + public async Task> RunAsync(string patchId, bool waitForRelease) + { + firstPatchResult = Workflow.Patched(patchId); + if (waitForRelease) + { + await Workflow.WaitConditionAsync(() => released); + } + return new[] { firstPatchResult, Workflow.Patched(patchId) }; + } + + [WorkflowQuery] + public bool FirstPatchResult() => firstPatchResult; + + [WorkflowSignal] + public Task ReleaseAsync() + { + released = true; + return Task.CompletedTask; + } + } + + [Workflow] + public class PatchActivationDeprecateWorkflow + { + [WorkflowRun] + public Task RunAsync(string patchId) + { + Workflow.DeprecatePatch(patchId); + return Task.FromResult(Workflow.Patched(patchId)); + } + } + + [Workflow("PatchActivationRolloutWorkflow")] + public class PatchActivationRolloutWorkflow + { + private bool released; + + [WorkflowRun] + public async Task RunAsync() + { + Workflow.Patched("rollout-patch"); + await Workflow.WaitConditionAsync(() => released); + return Workflow.Patched("rollout-patch") ? "new" : "old"; + } + + [WorkflowQuery] + public string State() => "new"; + + [WorkflowSignal] + public Task ReleaseAsync() + { + released = true; + return Task.CompletedTask; + } + } + + [Workflow("PatchActivationRolloutWorkflow")] + public class PatchActivationOldRolloutWorkflow + { + private bool released; + + [WorkflowRun] + public async Task RunAsync() + { + await Workflow.WaitConditionAsync(() => released); + return "old"; + } + + [WorkflowQuery] + public string State() => "old"; + + [WorkflowSignal] + public Task ReleaseAsync() + { + released = true; + return Task.CompletedTask; + } + } + + [Fact] + public async Task ExecuteWorkflowAsync_PatchActivationCallback_ActivatesAndMemoizes() + { + var calls = new ConcurrentQueue(); + var workflowId = $"workflow-{Guid.NewGuid()}"; + await ExecuteWorkerAsync( + async worker => + { + var result = await Client.ExecuteWorkflowAsync( + (PatchActivationWorkflow wf) => wf.RunAsync("my-patch", false), + new(workflowId, worker.Options.TaskQueue!)); + Assert.Equal([true, true], result); + }, + new TemporalWorkerOptions + { + PatchActivationCallback = input => + { + calls.Enqueue(input); + return true; + }, + }); + var call = Assert.Single(calls); + Assert.Equal(workflowId, call.WorkflowInfo.WorkflowId); + Assert.Equal("my-patch", call.PatchId); + } + + [Fact] + public async Task ExecuteWorkflowAsync_PatchActivationCallback_DeclinesWithoutMarker() + { + var calls = 0; + await ExecuteWorkerAsync( + async worker => + { + var handle = await Client.StartWorkflowAsync( + (PatchActivationWorkflow wf) => wf.RunAsync("my-patch", false), + new($"workflow-{Guid.NewGuid()}", worker.Options.TaskQueue!)); + Assert.Equal([false, false], await handle.GetResultAsync()); + Assert.DoesNotContain( + (await handle.FetchHistoryAsync()).Events, + evt => evt.MarkerRecordedEventAttributes != null); + }, + new TemporalWorkerOptions + { + PatchActivationCallback = _ => + { + Interlocked.Increment(ref calls); + return false; + }, + }); + Assert.Equal(1, calls); + } + + [Fact] + public async Task ExecuteWorkflowAsync_WithoutPatchActivationCallback_ActivatesWithMarker() + { + await ExecuteWorkerAsync(async worker => + { + var handle = await Client.StartWorkflowAsync( + (PatchActivationWorkflow wf) => wf.RunAsync("my-patch", false), + new($"workflow-{Guid.NewGuid()}", worker.Options.TaskQueue!)); + Assert.Equal([true, true], await handle.GetResultAsync()); + Assert.Contains( + (await handle.FetchHistoryAsync()).Events, + evt => evt.MarkerRecordedEventAttributes != null); + }); + } + + [Fact] + public async Task ExecuteWorkflowAsync_PatchActivationCallback_NotCalledOnUncachedReplay() + { + var calls = 0; + await ExecuteWorkerAsync( + async worker => + { + var handle = await Client.StartWorkflowAsync( + (PatchActivationWorkflow wf) => wf.RunAsync("my-patch", true), + new($"workflow-{Guid.NewGuid()}", worker.Options.TaskQueue!)); + Assert.False(await handle.QueryAsync(wf => wf.FirstPatchResult())); + await handle.SignalAsync(wf => wf.ReleaseAsync()); + Assert.Equal([false, false], await handle.GetResultAsync()); + }, + new TemporalWorkerOptions + { + MaxCachedWorkflows = 0, + PatchActivationCallback = _ => + { + Interlocked.Increment(ref calls); + return false; + }, + }); + Assert.Equal(1, calls); + } + + [Fact] + public async Task ExecuteWorkflowAsync_PatchActivationCallback_NotCalledForDeprecatePatch() + { + await ExecuteWorkerAsync( + async worker => Assert.True(await Client.ExecuteWorkflowAsync( + (PatchActivationDeprecateWorkflow wf) => wf.RunAsync("my-patch"), + new($"workflow-{Guid.NewGuid()}", worker.Options.TaskQueue!))), + new TemporalWorkerOptions + { + PatchActivationCallback = _ => throw new InvalidOperationException("Unexpected callback"), + }); + } + + [Fact] + public async Task ExecuteWorkflowAsync_DeclinedPatch_RollsOutToOldWorker() + { + var taskQueue = $"tq-{Guid.NewGuid()}"; + var calls = 0; + WorkflowHandle? handle = null; + using (var worker = new TemporalWorker( + Client, + new TemporalWorkerOptions(taskQueue) + { + MaxCachedWorkflows = 0, + PatchActivationCallback = _ => + { + Interlocked.Increment(ref calls); + return false; + }, + }.AddWorkflow())) + { + await worker.ExecuteAsync(async () => + { + handle = await Client.StartWorkflowAsync( + (PatchActivationRolloutWorkflow wf) => wf.RunAsync(), + new($"workflow-{Guid.NewGuid()}", taskQueue)); + Assert.Equal("new", await handle.QueryAsync(wf => wf.State())); + }); + } + Assert.Equal(1, calls); + Assert.NotNull(handle); + + using var oldWorker = new TemporalWorker( + Client, + new TemporalWorkerOptions(taskQueue) { MaxCachedWorkflows = 0 }. + AddWorkflow()); + await oldWorker.ExecuteAsync(async () => + { + await handle.SignalAsync(wf => wf.ReleaseAsync()); + Assert.Equal("old", await handle.GetResultAsync()); + }); + } + + [Fact] + public async Task ExecuteWorkflowAsync_ActivatedPatch_ReplaysWithoutCallback() + { + var taskQueue = $"tq-{Guid.NewGuid()}"; + var activatedCalls = 0; + WorkflowHandle? handle = null; + using (var worker = new TemporalWorker( + Client, + new TemporalWorkerOptions(taskQueue) + { + MaxCachedWorkflows = 0, + PatchActivationCallback = _ => + { + Interlocked.Increment(ref activatedCalls); + return true; + }, + }.AddWorkflow())) + { + await worker.ExecuteAsync(async () => + { + handle = await Client.StartWorkflowAsync( + (PatchActivationRolloutWorkflow wf) => wf.RunAsync(), + new($"workflow-{Guid.NewGuid()}", taskQueue)); + Assert.Equal("new", await handle.QueryAsync(wf => wf.State())); + }); + } + Assert.Equal(1, activatedCalls); + Assert.NotNull(handle); + + var declinedCalls = 0; + using var declinedWorker = new TemporalWorker( + Client, + new TemporalWorkerOptions(taskQueue) + { + MaxCachedWorkflows = 0, + PatchActivationCallback = _ => + { + Interlocked.Increment(ref declinedCalls); + return false; + }, + }.AddWorkflow()); + await declinedWorker.ExecuteAsync(async () => + { + await handle.SignalAsync(wf => wf.ReleaseAsync()); + Assert.Equal("new", await handle.GetResultAsync()); + }); + Assert.Equal(0, declinedCalls); + } + [Workflow] public class HeadersWithCodecWorkflow { From 1b6a027eb5d50c32de595ead19f8396cde6ac397 Mon Sep 17 00:00:00 2001 From: Spencer Judge Date: Tue, 14 Jul 2026 15:00:46 -0700 Subject: [PATCH 2/3] Harden read-only workflow contexts --- .../Worker/NotifyOnSetDictionary.cs | 31 +++-- .../Worker/TemporalWorkerOptions.cs | 4 +- src/Temporalio/Worker/WorkflowInstance.cs | 111 +++++++++++++++--- src/Temporalio/Workflows/IWorkflowContext.cs | 6 + src/Temporalio/Workflows/Workflow.cs | 14 ++- .../Worker/NotifyOnSetDictionaryTests.cs | 34 ++++++ .../Worker/WorkflowWorkerTests.cs | 111 ++++++++++++++++++ 7 files changed, 282 insertions(+), 29 deletions(-) create mode 100644 tests/Temporalio.Tests/Worker/NotifyOnSetDictionaryTests.cs diff --git a/src/Temporalio/Worker/NotifyOnSetDictionary.cs b/src/Temporalio/Worker/NotifyOnSetDictionary.cs index 3fb29bea..934c9752 100644 --- a/src/Temporalio/Worker/NotifyOnSetDictionary.cs +++ b/src/Temporalio/Worker/NotifyOnSetDictionary.cs @@ -6,7 +6,7 @@ namespace Temporalio.Worker { /// - /// Specialized dictionary that invokes a callback on each value set (but not value delete). + /// Specialized dictionary that invokes callbacks before mutation and after each value set. /// /// Dictionary key type. /// Dictionary value type. @@ -18,18 +18,23 @@ internal class NotifyOnSetDictionary : IDictionary, { private readonly Dictionary dict; private readonly Action callback; + private readonly Action beforeMutation; /// /// Initializes a new instance of the class. /// /// Dictionary to copy existing values from. /// Callback to invoke on value set. + /// Callback to invoke before any mutation. public NotifyOnSetDictionary( - IEnumerable> copyFrom, Action callback) + IEnumerable> copyFrom, + Action callback, + Action beforeMutation) { // Can't use kvp enumerable constructor on Dictionary, too new dict = copyFrom.ToDictionary(kvp => kvp.Key, kvp => kvp.Value); this.callback = callback; + this.beforeMutation = beforeMutation; } /// @@ -71,7 +76,11 @@ public TValue this[TKey key] public void Add(KeyValuePair item) => Add(item.Key, item.Value); /// - public void Clear() => dict.Clear(); + public void Clear() + { + beforeMutation(); + dict.Clear(); + } /// public bool Contains(KeyValuePair item) => @@ -88,11 +97,18 @@ public void CopyTo(KeyValuePair[] array, int arrayIndex) => public IEnumerator> GetEnumerator() => dict.GetEnumerator(); /// - public bool Remove(TKey key) => dict.Remove(key); + public bool Remove(TKey key) + { + beforeMutation(); + return dict.Remove(key); + } /// - public bool Remove(KeyValuePair item) => - ((ICollection>)dict).Remove(item); + public bool Remove(KeyValuePair item) + { + beforeMutation(); + return ((ICollection>)dict).Remove(item); + } /// public bool TryGetValue(TKey key, out TValue value) => @@ -105,6 +121,7 @@ public bool TryGetValue(TKey key, out TValue value) => private void Set(TKey key, TValue value, bool overwrite) { + beforeMutation(); if (!overwrite && ContainsKey(key)) { throw new ArgumentException("Key already exists"); @@ -113,4 +130,4 @@ private void Set(TKey key, TValue value, bool overwrite) callback(key, value); } } -} \ No newline at end of file +} diff --git a/src/Temporalio/Worker/TemporalWorkerOptions.cs b/src/Temporalio/Worker/TemporalWorkerOptions.cs index ca990fc5..8e7a3fbc 100644 --- a/src/Temporalio/Worker/TemporalWorkerOptions.cs +++ b/src/Temporalio/Worker/TemporalWorkerOptions.cs @@ -295,7 +295,9 @@ public TemporalWorkerOptions() /// /// The callback is only invoked for a newly encountered patch. Existing history markers, /// replay, and bypass the callback. Returning false - /// leaves the patch inactive and does not record a patch marker. + /// leaves the patch inactive and does not record a patch marker. The callback runs in a + /// read-only workflow context and therefore cannot use workflow randomness, wait, or + /// schedule workflow commands. /// /// WARNING: This property is experimental and may change in the future. public Func? PatchActivationCallback { get; set; } diff --git a/src/Temporalio/Worker/WorkflowInstance.cs b/src/Temporalio/Worker/WorkflowInstance.cs index 6ba916a9..84fa96dd 100644 --- a/src/Temporalio/Worker/WorkflowInstance.cs +++ b/src/Temporalio/Worker/WorkflowInstance.cs @@ -79,6 +79,7 @@ internal class WorkflowInstance : TaskScheduler, IWorkflowInstance, IWorkflowCon private readonly Func? patchActivationCallback; private readonly Handlers inProgressHandlers = new(); private readonly WorkflowDefinitionOptions definitionOptions; + private DeterministicRandom random; private WorkflowActivationCompletion? completion; // Will be set to null after last use (i.e. when workflow actually started) private Lazy? startArgs; @@ -97,6 +98,8 @@ internal class WorkflowInstance : TaskScheduler, IWorkflowInstance, IWorkflowCon private bool applyModernEventLoopLogic; private bool dynamicOptionsGetterInvoked; private bool inQueryOrValidator; + private bool contextFrozen; + private string currentDetails = string.Empty; /// /// Initializes a new instance of the class. @@ -133,9 +136,24 @@ public WorkflowInstance(WorkflowInstanceDetails details) return rootInbound.Outbound!; }, false); - mutableQueries = new(() => new(Definition.Queries, OnQueryDefinitionAdded), false); - mutableSignals = new(() => new(Definition.Signals, OnSignalDefinitionAdded), false); - mutableUpdates = new(() => new(Definition.Updates, OnUpdateDefinitionAdded), false); + mutableQueries = new( + () => new( + Definition.Queries, + OnQueryDefinitionAdded, + () => AssertNotReadOnly("modify workflow handlers")), + false); + mutableSignals = new( + () => new( + Definition.Signals, + OnSignalDefinitionAdded, + () => AssertNotReadOnly("modify workflow handlers")), + false); + mutableUpdates = new( + () => new( + Definition.Updates, + OnUpdateDefinitionAdded, + () => AssertNotReadOnly("modify workflow handlers")), + false); var initialMemo = details.Init.Memo; memo = new( () => initialMemo == null ? new Dictionary(0) : @@ -218,7 +236,7 @@ public WorkflowInstance(WorkflowInstanceDetails details) replaySafeLogger = new(logger); onTaskStarting = details.OnTaskStarting; onTaskCompleted = details.OnTaskCompleted; - Random = new(details.Init.RandomnessSeed); + random = new(details.Init.RandomnessSeed); TracingEventsEnabled = !details.DisableTracingEvents; workerLevelFailureExceptionTypes = details.WorkerLevelFailureExceptionTypes; disableEagerActivityExecution = details.DisableEagerActivityExecution; @@ -260,7 +278,15 @@ public WorkflowInstance(WorkflowInstanceDetails details) public WorkerDeploymentVersion? CurrentDeploymentVersion { get; private set; } /// - public string CurrentDetails { get; set; } = string.Empty; + public string CurrentDetails + { + get => currentDetails; + set + { + AssertNotReadOnly("set current details"); + currentDetails = value; + } + } /// public int CurrentHistoryLength { get; private set; } @@ -284,6 +310,7 @@ public WorkflowQueryDefinition? DynamicQuery get => dynamicQuery; set { + AssertNotReadOnly("modify workflow handlers"); if (value != null && !value.Dynamic) { throw new ArgumentException("Query is not dynamic"); @@ -298,6 +325,7 @@ public WorkflowSignalDefinition? DynamicSignal get => dynamicSignal; set { + AssertNotReadOnly("modify workflow handlers"); if (value != null && !value.Dynamic) { throw new ArgumentException("Signal is not dynamic"); @@ -320,6 +348,7 @@ public WorkflowUpdateDefinition? DynamicUpdate get => dynamicUpdate; set { + AssertNotReadOnly("modify workflow handlers"); if (value != null && !value.Dynamic) { throw new ArgumentException("Update is not dynamic"); @@ -381,7 +410,14 @@ public object Instance public IDictionary Queries => mutableQueries.Value; /// - public DeterministicRandom Random { get; private set; } + public DeterministicRandom Random + { + get + { + ThrowIfContextFrozen("use workflow randomness"); + return random; + } + } /// public IDictionary Signals => mutableSignals.Value; @@ -405,14 +441,26 @@ public object Instance /// internal Action AssertValidLocalActivity { get; private init; } + /// + public void AssertNotReadOnly(string operation) + { + if (contextFrozen || inQueryOrValidator) + { + throw new InvalidOperationException($"Cannot {operation} in this context"); + } + } + /// public ContinueAsNewException CreateContinueAsNewException( - string workflow, IReadOnlyCollection args, ContinueAsNewOptions? options) => - outbound.Value.CreateContinueAsNewException(new( + string workflow, IReadOnlyCollection args, ContinueAsNewOptions? options) + { + ThrowIfContextFrozen("continue as new"); + return outbound.Value.CreateContinueAsNewException(new( Workflow: workflow, Args: args, Options: options, Headers: null)); + } /// public NexusWorkflowClient CreateNexusWorkflowClient(string service, NexusWorkflowClientOptions options) => @@ -424,19 +472,19 @@ public NexusWorkflowClient CreateNexusWorkflowClient(NexusWo /// public Task DelayWithOptionsAsync(DelayOptions options) => - outbound.Value.DelayAsync(new(options)); + ContextFrozenChecked(() => outbound.Value.DelayAsync(new(options))); /// public Task ExecuteActivityAsync( string activity, IReadOnlyCollection args, ActivityOptions options) => - outbound.Value.ScheduleActivityAsync( - new(Activity: activity, Args: args, Options: options, Headers: null)); + ContextFrozenChecked(() => outbound.Value.ScheduleActivityAsync( + new(Activity: activity, Args: args, Options: options, Headers: null))); /// public Task ExecuteLocalActivityAsync( string activity, IReadOnlyCollection args, LocalActivityOptions options) => - outbound.Value.ScheduleLocalActivityAsync( - new(Activity: activity, Args: args, Options: options, Headers: null)); + ContextFrozenChecked(() => outbound.Value.ScheduleLocalActivityAsync( + new(Activity: activity, Args: args, Options: options, Headers: null))); /// public ExternalWorkflowHandle GetExternalWorkflowHandle( @@ -446,6 +494,7 @@ public ExternalWorkflowHandle GetExternalWorkflowHandle( /// public bool Patch(string patchId, bool deprecated) { + AssertNotReadOnly("patch"); // Use memoized result if present. If this is being deprecated, we can still use // memoized result and skip the command. if (patchesMemoized.TryGetValue(patchId, out var patched)) @@ -457,7 +506,16 @@ public bool Patch(string patchId, bool deprecated) if (!deprecated && !IsReplaying && !patchesNotified.Contains(patchId) && patchActivationCallback != null) { - patched = patchActivationCallback(new(Info, patchId)); + var previousContextFrozen = contextFrozen; + try + { + contextFrozen = true; + patched = patchActivationCallback(new(Info, patchId)); + } + finally + { + contextFrozen = previousContextFrozen; + } } else { @@ -477,12 +535,13 @@ public bool Patch(string patchId, bool deprecated) /// public Task> StartChildWorkflowAsync( string workflow, IReadOnlyCollection args, ChildWorkflowOptions options) => - outbound.Value.StartChildWorkflowAsync( - new(Workflow: workflow, Args: args, Options: options, Headers: null)); + ContextFrozenChecked(() => outbound.Value.StartChildWorkflowAsync( + new(Workflow: workflow, Args: args, Options: options, Headers: null))); /// public void UpsertMemo(IReadOnlyCollection updates) { + ThrowIfContextFrozen("issue workflow commands"); if (updates.Count == 0) { throw new ArgumentException("At least one update required", nameof(updates)); @@ -529,6 +588,7 @@ public void UpsertMemo(IReadOnlyCollection updates) /// public void UpsertTypedSearchAttributes(IReadOnlyCollection updates) { + ThrowIfContextFrozen("issue workflow commands"); if (updates.Count == 0) { throw new ArgumentException("At least one update required", nameof(updates)); @@ -556,6 +616,7 @@ public void UpsertTypedSearchAttributes(IReadOnlyCollection public Task WaitConditionWithOptionsAsync(WaitConditionOptions options) { + AssertNotReadOnly("wait or schedule workflow work"); var source = new TaskCompletionSource(); var node = conditions.AddLast(Tuple.Create(options.ConditionCheck, source)); var token = options.CancellationToken ?? CancellationToken; @@ -860,6 +921,7 @@ protected override bool TryExecuteTaskInline(Task task, bool taskWasPreviouslyQu /// protected override void QueueTask(Task task) { + ThrowIfContextFrozen("wait or schedule workflow work"); // Only queue if not already done if (!scheduledTaskNodes.ContainsKey(task)) { @@ -943,6 +1005,7 @@ private void RunOnce(bool checkConditions) private void AddCommand(WorkflowCommand cmd) { + ThrowIfContextFrozen("issue workflow commands"); if (completion == null) { throw new InvalidOperationException("No completion available"); @@ -951,6 +1014,20 @@ private void AddCommand(WorkflowCommand cmd) completion.Successful?.Commands.Add(cmd); } + private T ContextFrozenChecked(Func func) + { + ThrowIfContextFrozen("wait or schedule workflow work"); + return func(); + } + + private void ThrowIfContextFrozen(string operation) + { + if (contextFrozen) + { + throw new InvalidOperationException($"Cannot {operation} in this context"); + } + } + #pragma warning disable CA2008 // We don't have to pass a scheduler, factory already implies one #pragma warning disable VSTHRD003 // We know it's our own task we're waiting on private Task QueueNewTaskAsync(Func func) @@ -1646,7 +1723,7 @@ await inbound.Value.HandleSignalAsync(new( } private void ApplyUpdateRandomSeed(UpdateRandomSeed update) => - Random = new(update.RandomnessSeed); + random = new(update.RandomnessSeed); private void InitializeWorkflow() { diff --git a/src/Temporalio/Workflows/IWorkflowContext.cs b/src/Temporalio/Workflows/IWorkflowContext.cs index 3660c706..ee5fb837 100644 --- a/src/Temporalio/Workflows/IWorkflowContext.cs +++ b/src/Temporalio/Workflows/IWorkflowContext.cs @@ -153,6 +153,12 @@ internal interface IWorkflowContext /// DateTime UtcNow { get; } + /// + /// Throws if the workflow context is currently read-only. + /// + /// Operation being attempted. + void AssertNotReadOnly(string operation); + /// /// Backing call for /// . diff --git a/src/Temporalio/Workflows/Workflow.cs b/src/Temporalio/Workflows/Workflow.cs index d1301d10..3c0dc119 100644 --- a/src/Temporalio/Workflows/Workflow.cs +++ b/src/Temporalio/Workflows/Workflow.cs @@ -1139,12 +1139,15 @@ public static Guid NewGuid() /// A task for the running task (but not necessarily the task that is returned /// from the function). public static Task RunTaskAsync( - Func function, CancellationToken? cancellationToken = null) => - Task.Factory.StartNew( + Func function, CancellationToken? cancellationToken = null) + { + Context.AssertNotReadOnly("wait or schedule workflow work"); + return Task.Factory.StartNew( function, cancellationToken ?? CancellationToken, TaskCreationOptions.None, TaskScheduler.Current).Unwrap(); + } /// /// Workflow-safe form of . @@ -1156,12 +1159,15 @@ public static Task RunTaskAsync( /// A task for the running task (but not necessarily the task that is returned /// from the function). public static Task RunTaskAsync( - Func> function, CancellationToken? cancellationToken = null) => - Task.Factory.StartNew( + Func> function, CancellationToken? cancellationToken = null) + { + Context.AssertNotReadOnly("wait or schedule workflow work"); + return Task.Factory.StartNew( function, cancellationToken ?? CancellationToken, TaskCreationOptions.None, TaskScheduler.Current).Unwrap(); + } /// /// Start a child workflow via lambda invoking the run method. diff --git a/tests/Temporalio.Tests/Worker/NotifyOnSetDictionaryTests.cs b/tests/Temporalio.Tests/Worker/NotifyOnSetDictionaryTests.cs new file mode 100644 index 00000000..50af44a7 --- /dev/null +++ b/tests/Temporalio.Tests/Worker/NotifyOnSetDictionaryTests.cs @@ -0,0 +1,34 @@ +namespace Temporalio.Tests.Worker; + +using Temporalio.Worker; +using Xunit; + +public class NotifyOnSetDictionaryTests +{ + [Fact] + public void MutationGuard_AppliesToEveryMutation() + { + var mutationAllowed = true; + var dictionary = new NotifyOnSetDictionary( + new Dictionary { ["existing"] = "value" }, + (_, _) => { }, + () => + { + if (!mutationAllowed) + { + throw new InvalidOperationException("Mutation not allowed"); + } + }); + mutationAllowed = false; + + Assert.Throws(() => dictionary["existing"] = "new-value"); + Assert.Throws(() => dictionary.Add("new", "value")); + Assert.Throws(() => + dictionary.Add(new KeyValuePair("new", "value"))); + Assert.Throws(() => dictionary.Remove("existing")); + Assert.Throws(() => + dictionary.Remove(new KeyValuePair("existing", "value"))); + Assert.Throws(() => dictionary.Clear()); + Assert.Equal("value", dictionary["existing"]); + } +} diff --git a/tests/Temporalio.Tests/Worker/WorkflowWorkerTests.cs b/tests/Temporalio.Tests/Worker/WorkflowWorkerTests.cs index d840a941..0231ac4e 100644 --- a/tests/Temporalio.Tests/Worker/WorkflowWorkerTests.cs +++ b/tests/Temporalio.Tests/Worker/WorkflowWorkerTests.cs @@ -2969,12 +2969,21 @@ public async Task> RunAsync(string patchId, bool waitF [WorkflowQuery] public bool FirstPatchResult() => firstPatchResult; + [WorkflowQuery] + public bool PatchFromQuery(string patchId) => Workflow.Patched(patchId); + [WorkflowSignal] public Task ReleaseAsync() { released = true; return Task.CompletedTask; } + + [WorkflowUpdate] + public Task PatchFromUpdateAsync(string patchId) => Task.CompletedTask; + + [WorkflowUpdateValidator(nameof(PatchFromUpdateAsync))] + public void ValidatePatchFromUpdate(string patchId) => _ = Workflow.Patched(patchId); } [Workflow] @@ -3141,6 +3150,108 @@ await ExecuteWorkerAsync( }); } + [Theory] + [InlineData("command", "Cannot issue workflow commands in this context")] + [InlineData("wait", "Cannot wait or schedule workflow work in this context")] + [InlineData("run-task", "Cannot wait or schedule workflow work in this context")] + [InlineData("random", "Cannot use workflow randomness in this context")] + [InlineData("patch", "Cannot patch in this context")] + [InlineData("current-details", "Cannot set current details in this context")] + [InlineData("handler-add", "Cannot modify workflow handlers in this context")] + [InlineData("handler-remove", "Cannot modify workflow handlers in this context")] + [InlineData("dynamic-handler", "Cannot modify workflow handlers in this context")] + [InlineData("continue-as-new", "Cannot continue as new in this context")] + public async Task ExecuteWorkflowAsync_PatchActivationCallback_RunsReadOnly( + string operation, + string expectedError) + { + await ExecuteWorkerAsync( + async worker => + { + var handle = await Client.StartWorkflowAsync( + (PatchActivationWorkflow wf) => wf.RunAsync(operation, false), + new($"workflow-{Guid.NewGuid()}", worker.Options.TaskQueue!)); + await AssertTaskFailureContainsEventuallyAsync(handle, expectedError); + }, + new TemporalWorkerOptions + { + PatchActivationCallback = input => + { + switch (input.PatchId) + { + case "command": + Workflow.UpsertMemo(MemoUpdate.ValueSet("key", "value")); + break; + case "wait": + _ = Workflow.WaitConditionAsync(() => true); + break; + case "run-task": + _ = Workflow.RunTaskAsync(() => Task.CompletedTask); + break; + case "random": + _ = Workflow.Random.Next(); + break; + case "patch": + _ = Workflow.Patched("nested-patch"); + break; + case "current-details": + Workflow.CurrentDetails = "changed in callback"; + break; + case "handler-add": + Workflow.Queries["callback-query"] = + WorkflowQueryDefinition.CreateWithoutAttribute( + "callback-query", () => true); + break; + case "handler-remove": + Workflow.Queries.Remove(nameof(PatchActivationWorkflow.FirstPatchResult)); + break; + case "dynamic-handler": + Workflow.DynamicQuery = WorkflowQueryDefinition.CreateWithoutAttribute( + null, (string _queryName, IRawValue[] _args) => true); + break; + case "continue-as-new": + throw Workflow.CreateContinueAsNewException( + (PatchActivationWorkflow wf) => wf.RunAsync("continued", false)); + } + return true; + }, + }); + } + + [Fact] + public async Task ExecuteWorkflowAsync_PatchActivationCallback_RejectsReadOnlyCallsFromQueryAndValidator() + { + var calls = 0; + await ExecuteWorkerAsync( + async worker => + { + var handle = await Client.StartWorkflowAsync( + (PatchActivationWorkflow wf) => wf.RunAsync("main-patch", true), + new($"workflow-{Guid.NewGuid()}", worker.Options.TaskQueue!)); + Assert.False(await handle.QueryAsync(wf => wf.FirstPatchResult())); + + var queryExc = await Assert.ThrowsAsync( + () => handle.QueryAsync(wf => wf.PatchFromQuery("query-patch"))); + Assert.Contains("Cannot patch in this context", queryExc.Message); + + var updateExc = await Assert.ThrowsAsync( + () => handle.ExecuteUpdateAsync(wf => wf.PatchFromUpdateAsync("validator-patch"))); + Assert.Contains("Cannot patch in this context", updateExc.InnerException?.Message); + + await handle.SignalAsync(wf => wf.ReleaseAsync()); + Assert.Equal([false, false], await handle.GetResultAsync()); + }, + new TemporalWorkerOptions + { + PatchActivationCallback = _ => + { + Interlocked.Increment(ref calls); + return false; + }, + }); + Assert.Equal(1, calls); + } + [Fact] public async Task ExecuteWorkflowAsync_DeclinedPatch_RollsOutToOldWorker() { From ea7e71aca86bda053acaffa126a5d2b311a708ab Mon Sep 17 00:00:00 2001 From: Spencer Judge Date: Fri, 17 Jul 2026 16:05:04 -0700 Subject: [PATCH 3/3] Review feedback --- CHANGELOG.md | 16 ++++++++++++---- src/Temporalio/Worker/WorkflowInstance.cs | 2 +- .../Worker/WorkflowWorkerTests.cs | 6 +++--- 3 files changed, 16 insertions(+), 8 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index cce69425..e960dff8 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -19,6 +19,18 @@ to docs, or any other relevant information. ## [Unreleased] +### Added + +- Added the experimental `TemporalWorkerOptions.PatchActivationCallback`, allowing workers to + decide whether a first non-replay `Workflow.Patched` call should activate a patch during rolling + deployments. + +### Changed + +- Hardened read-only workflow context enforcement so queries, update validators, and patch activation + callbacks cannot mutate handlers or workflow details, invoke patches, or schedule workflow work. + Patch activation callbacks also cannot use workflow randomness or issue workflow commands. + ### [1.17.0] - 2026-07-13 ### Added @@ -32,10 +44,6 @@ to docs, or any other relevant information. connections now compress outbound requests and accept gzip-compressed responses by default. If the remote service does not support gzip compression, the connection is downgraded to uncompressed requests. Set it to `GrpcCompression.None` to opt out. -- Added AWS Lambda worker support packages, including OpenTelemetry helpers for Lambda workers. -- Added the experimental `TemporalWorkerOptions.PatchActivationCallback`, allowing workers to - decide whether a first non-replay `Workflow.Patched` call should activate a patch during rolling - deployments. ### Fixed diff --git a/src/Temporalio/Worker/WorkflowInstance.cs b/src/Temporalio/Worker/WorkflowInstance.cs index 84fa96dd..34b7f066 100644 --- a/src/Temporalio/Worker/WorkflowInstance.cs +++ b/src/Temporalio/Worker/WorkflowInstance.cs @@ -494,7 +494,7 @@ public ExternalWorkflowHandle GetExternalWorkflowHandle( /// public bool Patch(string patchId, bool deprecated) { - AssertNotReadOnly("patch"); + AssertNotReadOnly(deprecated ? "deprecate patch" : "create patch"); // Use memoized result if present. If this is being deprecated, we can still use // memoized result and skip the command. if (patchesMemoized.TryGetValue(patchId, out var patched)) diff --git a/tests/Temporalio.Tests/Worker/WorkflowWorkerTests.cs b/tests/Temporalio.Tests/Worker/WorkflowWorkerTests.cs index 0231ac4e..f2961489 100644 --- a/tests/Temporalio.Tests/Worker/WorkflowWorkerTests.cs +++ b/tests/Temporalio.Tests/Worker/WorkflowWorkerTests.cs @@ -3155,7 +3155,7 @@ await ExecuteWorkerAsync( [InlineData("wait", "Cannot wait or schedule workflow work in this context")] [InlineData("run-task", "Cannot wait or schedule workflow work in this context")] [InlineData("random", "Cannot use workflow randomness in this context")] - [InlineData("patch", "Cannot patch in this context")] + [InlineData("patch", "Cannot create patch in this context")] [InlineData("current-details", "Cannot set current details in this context")] [InlineData("handler-add", "Cannot modify workflow handlers in this context")] [InlineData("handler-remove", "Cannot modify workflow handlers in this context")] @@ -3232,11 +3232,11 @@ await ExecuteWorkerAsync( var queryExc = await Assert.ThrowsAsync( () => handle.QueryAsync(wf => wf.PatchFromQuery("query-patch"))); - Assert.Contains("Cannot patch in this context", queryExc.Message); + Assert.Contains("Cannot create patch in this context", queryExc.Message); var updateExc = await Assert.ThrowsAsync( () => handle.ExecuteUpdateAsync(wf => wf.PatchFromUpdateAsync("validator-patch"))); - Assert.Contains("Cannot patch in this context", updateExc.InnerException?.Message); + Assert.Contains("Cannot create patch in this context", updateExc.InnerException?.Message); await handle.SignalAsync(wf => wf.ReleaseAsync()); Assert.Equal([false, false], await handle.GetResultAsync());