diff --git a/src/Dapr.Workflow/Worker/Internal/WorkflowOrchestrationContext.cs b/src/Dapr.Workflow/Worker/Internal/WorkflowOrchestrationContext.cs index 0edae7ce2..badf9f0c2 100644 --- a/src/Dapr.Workflow/Worker/Internal/WorkflowOrchestrationContext.cs +++ b/src/Dapr.Workflow/Worker/Internal/WorkflowOrchestrationContext.cs @@ -115,8 +115,7 @@ public WorkflowOrchestrationContext(string name, string instanceId, DateTime cur /// public override bool IsPatched(string patchName) { - var hasPatchHistory = _versionTracker.AggregatedPatchesOrdered.Count > 0; - return _versionTracker.RequestPatch(patchName, isReplaying: this.IsReplaying && hasPatchHistory); + return _versionTracker.RequestPatch(patchName, isReplaying: this.IsReplaying); } /// diff --git a/src/Dapr.Workflow/Worker/WorkflowWorker.cs b/src/Dapr.Workflow/Worker/WorkflowWorker.cs index 79d17f623..33de41ef0 100644 --- a/src/Dapr.Workflow/Worker/WorkflowWorker.cs +++ b/src/Dapr.Workflow/Worker/WorkflowWorker.cs @@ -632,16 +632,24 @@ public override async Task StopAsync(CancellationToken cancellationToken) return null; var traceState = request.ParentTraceContext?.TraceState; - if (ActivityContext.TryParse(traceParent, traceState, out var parentCtx)) + if (ActivityContext.TryParse(traceParent, traceState, isRemote: true, out var parentCtx)) { - return WorkflowActivitySource.StartActivity( + // Prefer ActivitySource so registered telemetry listeners (e.g. OpenTelemetry) receive the span. + var started = WorkflowActivitySource.StartActivity( name: $"WorkflowActivity {request.Name}", kind: ActivityKind.Internal, parentContext: parentCtx, []); + if (started != null) + return started; + + // ActivitySource.StartActivity returns null when no listener is registered for "Dapr.Workflow". + // Fall through to ensure Activity.Current is always non-null inside user activity code, + // regardless of whether OpenTelemetry or another telemetry listener is configured. } - - // Fall back to raw parent ID if parsing fails + + // Always create an Activity directly (not via ActivitySource) so that Activity.Current is + // non-null in user activity code even when no telemetry listener is registered. var act = new Activity($"WorkflowActivity {request.Name}"); act.SetParentId(traceParent); if (!string.IsNullOrEmpty(traceState)) diff --git a/test/Dapr.IntegrationTest.Workflow/Versioning/Patches/PatchAddedToWorkflowWithNoPriorHistoryTests.cs b/test/Dapr.IntegrationTest.Workflow/Versioning/Patches/PatchAddedToWorkflowWithNoPriorHistoryTests.cs new file mode 100644 index 000000000..97c6bfbca --- /dev/null +++ b/test/Dapr.IntegrationTest.Workflow/Versioning/Patches/PatchAddedToWorkflowWithNoPriorHistoryTests.cs @@ -0,0 +1,186 @@ +// ------------------------------------------------------------------------ +// 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. +// ------------------------------------------------------------------------ + +using Dapr.Testcontainers.Common; +using Dapr.Testcontainers.Common.Testing; +using Dapr.Testcontainers.Harnesses; +using Dapr.Testcontainers.Xunit.Attributes; +using Dapr.Workflow; +using Microsoft.Extensions.Configuration; +using Microsoft.Extensions.DependencyInjection; + +namespace Dapr.IntegrationTest.Workflow.Versioning.Patches; + +/// +/// Regression test: adding an IsPatched guard to a workflow that was originally deployed +/// without any IsPatched calls must not corrupt the task-sequence during replay. +/// +/// Failure mode (before fix): the runtime holds no version / patch history for the in-flight +/// instance. When IsPatched is called during replay, WorkflowOrchestrationContext +/// incorrectly passes isReplaying: false to WorkflowVersionTracker.RequestPatch +/// (because hasPatchHistory == false). RequestPatch therefore returns true, +/// causing the new patched-path activity to consume task-slot 0 (the original email-activity +/// slot). The original email activity is then assigned slot 1, and the TimerFired event +/// (also on slot 1) resolves that TCS. HandleHistoryMatch throws +/// InvalidOperationException: "Unexpected history event type for task ID 1", leaving the +/// workflow in the Failed state. +/// +/// Fix: pass isReplaying: this.IsReplaying unconditionally (without && hasPatchHistory). +/// +public sealed class PatchAddedToWorkflowWithNoPriorHistoryTests +{ + // Environment variable shared between the two "deployments" within a single test run. + // Controlled inside the test; set before starting each app instance. + private const string ModeEnvVar = "DAPR_PATCH_NO_HISTORY_TEST_MODE"; + + /// + /// Timer duration used in the V1 workflow. Long enough that we can stop the app before it + /// fires, short enough that the test completes in a reasonable time. + /// + private static readonly TimeSpan TimerDuration = TimeSpan.FromSeconds(10); + + [MinimumDaprRuntimeFact("1.17.0")] + public async Task Workflow_PatchAddedAfterDeploy_NoPriorPatchHistory_CompletesSuccessfully() + { + var componentsDir = TestDirectoryManager.CreateTestDirectory("patch-no-prior-history"); + var instanceId = Guid.NewGuid().ToString(); + + await using var environment = await DaprTestEnvironment.CreateWithPooledNetworkAsync(needsActorState: true); + await environment.StartAsync(); + + var harness = new DaprHarnessBuilder(componentsDir) + .WithEnvironment(environment) + .BuildWorkflow(); + + // ── Phase 1: deploy V1 (no IsPatched) ──────────────────────────────────────────────── + // The workflow calls EmailActivity and then waits on a durable timer. + // We stop the app while the timer is still in-flight so the Dapr sidecar owns the + // timer state. When V2 starts, the sidecar delivers the TimerFired new-event, which + // is what triggers the bug. + Environment.SetEnvironmentVariable(ModeEnvVar, "v1"); + + await using (var appV1 = await BuildWorkflowAppAsync(harness)) + { + using var scope = appV1.CreateScope(); + var client = scope.ServiceProvider.GetRequiredService(); + + await client.ScheduleNewWorkflowAsync(nameof(NotifyWorkflow), instanceId, "user@example.com"); + + // Wait long enough for the EmailActivity to complete and for the timer to be + // recorded in the workflow history, but short enough that the timer has NOT yet + // fired (timer duration is 10 s, we wait 3 s). + await Task.Delay(TimeSpan.FromSeconds(3)); + } + + // ── Phase 2: deploy V2 (IsPatched added before EmailActivity) ──────────────────────── + // The patched path would call SmsActivity first. For a correctly replaying workflow + // IsPatched("sms") must return false (patch not in prior history), so only EmailActivity + // runs, preserving task-slot 0 for the history's TaskCompleted(EmailActivity) event and + // task-slot 1 for the TimerFired event. + Environment.SetEnvironmentVariable(ModeEnvVar, "v2"); + + await using (var appV2 = await BuildWorkflowAppAsync(harness)) + { + using var scope = appV2.CreateScope(); + var client = scope.ServiceProvider.GetRequiredService(); + + using var cts = new CancellationTokenSource(TimeSpan.FromMinutes(2)); + WorkflowState result; + try + { + result = await client.WaitForWorkflowCompletionAsync(instanceId, cancellation: cts.Token); + } + catch (OperationCanceledException) + { + var state = await client.GetWorkflowStateAsync(instanceId, getInputsAndOutputs: true); + Assert.Fail( + $"Timed out waiting for workflow '{instanceId}' to complete. " + + $"Current status: {state?.RuntimeStatus}."); + return; + } + + Assert.Equal(WorkflowRuntimeStatus.Completed, result.RuntimeStatus); + + // The in-flight workflow was started with V1 code (no IsPatched). On replay, + // IsPatched("sms") must evaluate to false, so only EmailActivity runs. + // The expected output is "email" – not "sms+email" which would indicate the + // patched path was incorrectly entered during replay. + var output = result.ReadOutputAs(); + Assert.Equal("email", output); + } + } + + private static Task BuildWorkflowAppAsync(BaseHarness harness) + { + return DaprHarnessBuilder.ForHarness(harness) + .ConfigureServices(builder => + { + builder.Services.AddDaprWorkflowBuilder( + configureRuntime: opt => + { + opt.RegisterWorkflow(); + opt.RegisterActivity(); + opt.RegisterActivity(); + }, + configureClient: (sp, clientBuilder) => + { + var config = sp.GetRequiredService(); + var grpcEndpoint = config["DAPR_GRPC_ENDPOINT"]; + if (!string.IsNullOrEmpty(grpcEndpoint)) + clientBuilder.UseGrpcEndpoint(grpcEndpoint); + }); + }) + .BuildAndStartAsync(); + } + + /// + /// Workflow under test. + /// + /// V1 behaviour (mode = "v1"): EmailActivity → timer → return "email" + /// V2 behaviour (mode = "v2"): if IsPatched("sms") { SmsActivity } → EmailActivity → timer → return + /// + /// When replaying an in-flight V1 instance under V2 code, IsPatched("sms") must + /// return false so the code path matches the original V1 execution. + /// + private sealed class NotifyWorkflow : Workflow + { + public override async Task RunAsync(WorkflowContext context, string input) + { + var mode = Environment.GetEnvironmentVariable(ModeEnvVar) ?? "v1"; + var result = string.Empty; + + if (mode == "v2" && context.IsPatched("sms")) + { + result += await context.CallActivityAsync(nameof(SmsActivity), input); + } + + result += await context.CallActivityAsync(nameof(EmailActivity), input); + + await context.CreateTimer(TimerDuration); + + return result; + } + } + + private sealed class EmailActivity : WorkflowActivity + { + public override Task RunAsync(WorkflowActivityContext context, string input) + => Task.FromResult("email"); + } + + private sealed class SmsActivity : WorkflowActivity + { + public override Task RunAsync(WorkflowActivityContext context, string input) + => Task.FromResult("sms+"); + } +} diff --git a/test/Dapr.Workflow.Test/Worker/WorkflowWorkerTests.cs b/test/Dapr.Workflow.Test/Worker/WorkflowWorkerTests.cs index 3316ce6d9..24e02fd6b 100644 --- a/test/Dapr.Workflow.Test/Worker/WorkflowWorkerTests.cs +++ b/test/Dapr.Workflow.Test/Worker/WorkflowWorkerTests.cs @@ -268,7 +268,73 @@ public async Task HandleActivityResponseAsync_ShouldFallBackToSetParentId_WhenTr Assert.Null(response.FailureDetails); Assert.Equal(malformedParentId, observedParentId); } - + + [Fact] + public async Task HandleActivityResponseAsync_ShouldPopulateActivityCurrent_WithoutRegisteredListener() + { + // Regression test for issue #1749: Activity.Current must be non-null inside user activity + // code even when NO ActivityListener is registered for "Dapr.Workflow". The original patch + // in #1795 used WorkflowActivitySource.StartActivity(), which returns null when there is + // no listener — leaving Activity.Current null for users who haven't wired up OpenTelemetry. + // Explicitly do NOT register any ActivityListener here to reproduce the real-world scenario. + var sp = new ServiceCollection().BuildServiceProvider(); + var serializer = new JsonDaprSerializer(new JsonSerializerOptions(JsonSerializerDefaults.Web)); + var options = new WorkflowRuntimeOptions(); + + const string expectedTraceId = "0af7651916cd43dd8448eb211c80319c"; + const string parentSpanId = "b7ad6b7169203331"; + const string traceParent = $"00-{expectedTraceId}-{parentSpanId}-01"; + const string traceState = "vendor=value"; + + Activity? observedCurrent = null; + string? observedTraceId = null; + string? observedParentSpanId = null; + string? observedTraceState = null; + + var factory = new StubWorkflowsFactory(); + factory.AddActivity("act", new InlineActivity( + inputType: typeof(int), + run: (_, _) => + { + observedCurrent = Activity.Current; + observedTraceId = Activity.Current?.TraceId.ToHexString(); + observedParentSpanId = Activity.Current?.ParentSpanId.ToHexString(); + observedTraceState = Activity.Current?.TraceStateString; + return Task.FromResult(null); + })); + + var worker = new WorkflowWorker( + CreateGrpcClientMock().Object, + factory, + NullLoggerFactory.Instance, + serializer, + sp, + options); + + Activity.Current = null; + + var request = new ActivityRequest + { + Name = "act", + TaskId = 5, + Input = string.Empty, + OrchestrationInstance = new OrchestrationInstance { InstanceId = "wf-5" }, + ParentTraceContext = new TraceContext + { + TraceParent = traceParent, + TraceState = traceState + } + }; + + var response = await InvokeHandleActivityResponseAsync(worker, request); + + Assert.Null(response.FailureDetails); + Assert.NotNull(observedCurrent); + Assert.Equal(expectedTraceId, observedTraceId); + Assert.Equal(parentSpanId, observedParentSpanId); + Assert.Equal(traceState, observedTraceState); + } + [Fact] public void Constructor_ShouldThrowArgumentNullException_WhenGrpcClientIsNull() {