From c6ef936bc02ea52a720f6014cab90ec70c2c8cf4 Mon Sep 17 00:00:00 2001 From: wangbill Date: Tue, 6 Oct 2026 17:59:20 -0400 Subject: [PATCH 1/2] Propagate tags from entity-started orchestrations Snapshot and validate entity scheduling tags, carry them through both gRPC action conversions, and import the pinned canonical field8 schema. Cover both overloads, mutation, rollback, cached execution, and wire round trips with the explicit unpublished Core PR dependency. Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com> Copilot-Session: 3d59e0dc-efdb-4b0a-b588-6dbd83416193 --- Directory.Packages.props | 2 +- .../Entities/TaskEntityContext.cs | 8 + src/Grpc/orchestrator_service.proto | 1 + src/Grpc/versions.txt | 8 +- src/Shared/Grpc/ProtoUtils.cs | 11 + src/Worker/Core/Shims/TaskEntityShim.cs | 1 + .../Core.Tests/Shims/TaskEntityShimTests.cs | 306 ++++++++++++++++++ .../Grpc.Tests/GrpcEntityRunnerTagTests.cs | 302 +++++++++++++++++ .../Grpc.Tests/ProtoUtilsEntityActionTests.cs | 151 +++++++++ 9 files changed, 785 insertions(+), 5 deletions(-) create mode 100644 test/Worker/Core.Tests/Shims/TaskEntityShimTests.cs create mode 100644 test/Worker/Grpc.Tests/GrpcEntityRunnerTagTests.cs create mode 100644 test/Worker/Grpc.Tests/ProtoUtilsEntityActionTests.cs diff --git a/Directory.Packages.props b/Directory.Packages.props index 7b634b8fc..678ad3308 100644 --- a/Directory.Packages.props +++ b/Directory.Packages.props @@ -39,7 +39,7 @@ - + diff --git a/src/Abstractions/Entities/TaskEntityContext.cs b/src/Abstractions/Entities/TaskEntityContext.cs index 9de1ff46e..7a5e087e1 100644 --- a/src/Abstractions/Entities/TaskEntityContext.cs +++ b/src/Abstractions/Entities/TaskEntityContext.cs @@ -41,6 +41,10 @@ public abstract void SignalEntity( /// The name of the orchestration to start. /// The options for starting the orchestration. /// The instance id for the new orchestration. + /// + /// Tags are copied when the orchestration is scheduled. Later changes to the tags dictionary do not affect it. + /// + /// A supplied tag key or value is null. public virtual string ScheduleNewOrchestration(TaskName name, StartOrchestrationOptions options) => this.ScheduleNewOrchestration(name, null, options); @@ -51,6 +55,10 @@ public virtual string ScheduleNewOrchestration(TaskName name, StartOrchestration /// The input for the orchestration. /// The options for starting the orchestration. /// The instance id for the new orchestration. + /// + /// Tags are copied when the orchestration is scheduled. Later changes to the tags dictionary do not affect it. + /// + /// A supplied tag key or value is null. public abstract string ScheduleNewOrchestration( TaskName name, object? input = null, StartOrchestrationOptions? options = null); } diff --git a/src/Grpc/orchestrator_service.proto b/src/Grpc/orchestrator_service.proto index 3d9194acc..2429fff46 100644 --- a/src/Grpc/orchestrator_service.proto +++ b/src/Grpc/orchestrator_service.proto @@ -706,6 +706,7 @@ message StartNewOrchestrationAction { google.protobuf.Timestamp scheduledTime = 5; google.protobuf.Timestamp requestTime = 6; TraceContext parentTraceContext = 7; + map tags = 8; } message AbandonActivityTaskRequest { diff --git a/src/Grpc/versions.txt b/src/Grpc/versions.txt index eca34717f..8902a6399 100644 --- a/src/Grpc/versions.txt +++ b/src/Grpc/versions.txt @@ -1,4 +1,4 @@ -# The following files were downloaded from branch main at 2026-09-17 15:24:44 UTC -https://raw.githubusercontent.com/microsoft/durabletask-protobuf/882583c1177706e148c3b146d4b2ba1c1d242e4c/protos/orchestrator_service.proto -https://raw.githubusercontent.com/microsoft/durabletask-protobuf/882583c1177706e148c3b146d4b2ba1c1d242e4c/protos/durable-task-scheduler/sandbox_service.proto -https://raw.githubusercontent.com/microsoft/durabletask-protobuf/882583c1177706e148c3b146d4b2ba1c1d242e4c/protos/durable-task-scheduler/large_payload_purge.proto +# The following files were downloaded from branch 8188ba772b1c9e2b47e2ce51ae528149867f0f09 at 2026-10-06 21:42:58 UTC +https://raw.githubusercontent.com/microsoft/durabletask-protobuf/8188ba772b1c9e2b47e2ce51ae528149867f0f09/protos/orchestrator_service.proto +https://raw.githubusercontent.com/microsoft/durabletask-protobuf/8188ba772b1c9e2b47e2ce51ae528149867f0f09/protos/durable-task-scheduler/sandbox_service.proto +https://raw.githubusercontent.com/microsoft/durabletask-protobuf/8188ba772b1c9e2b47e2ce51ae528149867f0f09/protos/durable-task-scheduler/large_payload_purge.proto diff --git a/src/Shared/Grpc/ProtoUtils.cs b/src/Shared/Grpc/ProtoUtils.cs index e3691664d..45779e676 100644 --- a/src/Shared/Grpc/ProtoUtils.cs +++ b/src/Shared/Grpc/ProtoUtils.cs @@ -813,6 +813,9 @@ internal static void ToEntityBatchRequest( Input = operationAction.StartNewOrchestration.Input, InstanceId = operationAction.StartNewOrchestration.InstanceId, Version = operationAction.StartNewOrchestration.Version, + Tags = operationAction.StartNewOrchestration.Tags.Count > 0 + ? new Dictionary(operationAction.StartNewOrchestration.Tags, StringComparer.Ordinal) + : null, ScheduledStartTime = operationAction.StartNewOrchestration.ScheduledTime?.ToDateTime(), RequestTime = operationAction.StartNewOrchestration.RequestTime?.ToDateTimeOffset(), ParentTraceContext = operationAction.StartNewOrchestration.ParentTraceContext != null ? @@ -879,6 +882,14 @@ internal static void ToEntityBatchRequest( } : null, }; + if (startNewOrchestrationAction.Tags != null) + { + foreach (KeyValuePair tag in startNewOrchestrationAction.Tags) + { + action.StartNewOrchestration.Tags[tag.Key] = tag.Value; + } + } + break; } diff --git a/src/Worker/Core/Shims/TaskEntityShim.cs b/src/Worker/Core/Shims/TaskEntityShim.cs index bad6c7f08..14c1d3b8e 100644 --- a/src/Worker/Core/Shims/TaskEntityShim.cs +++ b/src/Worker/Core/Shims/TaskEntityShim.cs @@ -248,6 +248,7 @@ public override string ScheduleNewOrchestration(TaskName name, object? input = n Version = options?.Version ?? string.Empty, InstanceId = instanceId, Input = this.dataConverter.Serialize(input), + Tags = options?.Tags?.ToDictionary(tag => tag.Key, tag => Check.NotNull(tag.Value)), ScheduledStartTime = options?.StartAt?.UtcDateTime, RequestTime = DateTimeOffset.UtcNow, ParentTraceContext = this.parentTraceContext, diff --git a/test/Worker/Core.Tests/Shims/TaskEntityShimTests.cs b/test/Worker/Core.Tests/Shims/TaskEntityShimTests.cs new file mode 100644 index 000000000..d7ac9de21 --- /dev/null +++ b/test/Worker/Core.Tests/Shims/TaskEntityShimTests.cs @@ -0,0 +1,306 @@ +// Copyright (c) Microsoft Corporation. +// Licensed under the MIT License. + +using DurableTask.Core.Entities; +using DurableTask.Core.Entities.OperationFormat; +using DurableTask.Core.Tracing; +using Microsoft.DurableTask.Converters; +using Microsoft.DurableTask.Entities; +using Microsoft.Extensions.Logging.Abstractions; + +namespace Microsoft.DurableTask.Worker.Shims; + +public class TaskEntityShimTests +{ + static readonly DistributedTraceContext TraceContext = new( + "00-0123456789abcdef0123456789abcdef-0123456789abcdef-01", + "vendor=value"); + + [Theory] + [InlineData(false)] + [InlineData(true)] + public async Task ScheduleNewOrchestration_WithTags_PreservesStartOptionsAsync(bool optionsOnly) + { + // Arrange + Dictionary tags = new() + { + ["owner"] = "entity", + ["\u6807\u7b7e"] = "\u4f60\u597d \ud83c\udf0d", + ["empty"] = string.Empty, + [string.Empty] = "empty-key", + }; + DateTimeOffset startAt = new(2030, 1, 2, 3, 4, 5, TimeSpan.FromHours(2)); + StartOrchestrationOptions options = new("child", startAt) + { + Version = "v2", + Tags = tags, + }; + DateTimeOffset before = DateTimeOffset.UtcNow; + + // Act + EntityBatchResult result = await RunAsync( + operation => optionsOnly + ? operation.Context.ScheduleNewOrchestration("Child", options) + : operation.Context.ScheduleNewOrchestration("Child", "input", options), + "start"); + DateTimeOffset after = DateTimeOffset.UtcNow; + + // Assert + Assert.NotNull(result.Actions); + Assert.NotNull(result.Results); + StartNewOrchestrationOperationAction action = Assert.IsType( + Assert.Single(result.Actions)); + action.Tags.Should().BeEquivalentTo(tags).And.NotBeSameAs(tags); + action.Name.Should().Be("Child"); + action.InstanceId.Should().Be("child"); + action.Version.Should().Be("v2"); + action.Input.Should().Be(optionsOnly ? null : "\"input\""); + action.ScheduledStartTime.Should().Be(startAt.UtcDateTime); + Assert.InRange(action.RequestTime!.Value, before, after); + action.ParentTraceContext.Should().BeSameAs(TraceContext); + Assert.Single(result.Results).Result.Should().Be("\"child\""); + } + + [Theory] + [InlineData(false, false)] + [InlineData(false, true)] + [InlineData(true, false)] + [InlineData(true, true)] + public async Task ScheduleNewOrchestration_NullOrEmptyTags_PreservesAbsenceAsync(bool optionsOnly, bool empty) + { + // Arrange + StartOrchestrationOptions options = new("child") + { + Tags = empty ? new Dictionary() : null!, + }; + + // Act + EntityBatchResult result = await RunAsync( + operation => optionsOnly + ? operation.Context.ScheduleNewOrchestration("Child", options) + : operation.Context.ScheduleNewOrchestration("Child", input: null, options), + "start"); + + // Assert + Assert.NotNull(result.Actions); + StartNewOrchestrationOperationAction action = Assert.IsType( + Assert.Single(result.Actions)); + if (empty) + { + Assert.NotNull(action.Tags); + Assert.Empty(action.Tags); + Assert.NotSame(options.Tags, action.Tags); + } + else + { + Assert.Null(action.Tags); + } + } + + [Fact] + public async Task ScheduleNewOrchestration_WithoutOptions_PreservesDefaultsAsync() + { + // Arrange + TaskName name = "Child"; + + // Act + EntityBatchResult result = await RunAsync( + operation => operation.Context.ScheduleNewOrchestration(name), + "start"); + + // Assert + Assert.NotNull(result.Actions); + Assert.NotNull(result.Results); + StartNewOrchestrationOperationAction action = Assert.IsType( + Assert.Single(result.Actions)); + Assert.Null(action.Tags); + Assert.Null(action.Input); + Assert.Null(action.ScheduledStartTime); + Assert.Equal(string.Empty, action.Version); + Assert.True(Guid.TryParseExact(action.InstanceId, "N", out _)); + Assert.Single(result.Results).Result.Should().Be(JsonDataConverter.Default.Serialize(action.InstanceId)); + } + + [Theory] + [InlineData(false)] + [InlineData(true)] + public async Task ScheduleNewOrchestration_ReusedDictionary_SnapshotsEachStartAsync(bool optionsOnly) + { + // Arrange + Dictionary tags = new() { ["owner"] = "before", ["remove"] = "original" }; + StartOrchestrationOptions options = new("first") { Tags = tags }; + + // Act + EntityBatchResult result = await RunAsync(operation => + { + if (optionsOnly) + { + operation.Context.ScheduleNewOrchestration("Child", options); + } + else + { + operation.Context.ScheduleNewOrchestration("Child", "first-input", options); + } + + tags["owner"] = "after"; + tags.Remove("remove"); + tags["added"] = "second"; + options = options with { InstanceId = "second" }; + if (optionsOnly) + { + operation.Context.ScheduleNewOrchestration("Child", options); + } + else + { + operation.Context.ScheduleNewOrchestration("Child", "second-input", options); + } + + tags.Clear(); + return null; + }, "start"); + + // Assert + Assert.NotNull(result.Actions); + Assert.Collection( + result.Actions, + first => Assert.IsType(first).Tags.Should().BeEquivalentTo( + new Dictionary { ["owner"] = "before", ["remove"] = "original" }), + second => Assert.IsType(second).Tags.Should().BeEquivalentTo( + new Dictionary { ["owner"] = "after", ["added"] = "second" })); + } + + [Theory] + [InlineData(false)] + [InlineData(true)] + public async Task ExecuteOperationBatchAsync_FailedStart_RollsBackOnlyFailedOperationAsync(bool scheduleThrows) + { + // Arrange + string? stateAfterFailure = null; + + // Act + EntityBatchResult result = await RunAsync(operation => + { + if (operation.Name == "after") + { + stateAfterFailure = operation.State.GetState(); + } + + operation.State.SetState(operation.Name); + operation.Context.ScheduleNewOrchestration("Child", new StartOrchestrationOptions(operation.Name) + { + Tags = new Dictionary { ["operation"] = operation.Name }, + }); + if (operation.Name == "fail") + { + if (scheduleThrows) + { + operation.Context.ScheduleNewOrchestration("Child", new StartOrchestrationOptions("@entity@key")); + } + + throw new InvalidOperationException("Failure after scheduling."); + } + + return null; + }, "before", "fail", "after"); + + // Assert + Assert.Null(result.FailureDetails); + Assert.NotNull(result.Results); + Assert.NotNull(result.Actions); + Assert.Equal(3, result.Results.Count); + Assert.Null(result.Results[0].FailureDetails); + Assert.NotNull(result.Results[1].FailureDetails); + Assert.Equal( + scheduleThrows ? typeof(ArgumentException).FullName : typeof(InvalidOperationException).FullName, + result.Results[1].FailureDetails?.ErrorType); + Assert.Null(result.Results[2].FailureDetails); + Assert.Equal("before", stateAfterFailure); + Assert.Equal("\"after\"", result.EntityState); + Assert.Collection( + result.Actions, + before => + { + StartNewOrchestrationOperationAction action = Assert.IsType(before); + Assert.Equal("before", action.InstanceId); + action.Tags.Should().BeEquivalentTo(new Dictionary { ["operation"] = "before" }); + }, + after => + { + StartNewOrchestrationOperationAction action = Assert.IsType(after); + Assert.Equal("after", action.InstanceId); + action.Tags.Should().BeEquivalentTo(new Dictionary { ["operation"] = "after" }); + }); + } + + [Fact] + public async Task ScheduleNewOrchestration_NullTagValue_RollsBackWithinOperationAsync() + { + // Arrange + string? stateAfterFailure = null; + + // Act + EntityBatchResult result = await RunAsync(operation => + { + if (operation.Name == "after") + { + stateAfterFailure = operation.State.GetState(); + } + + operation.State.SetState(operation.Name); + operation.Context.ScheduleNewOrchestration("Child", new StartOrchestrationOptions(operation.Name) + { + Tags = new Dictionary { ["operation"] = operation.Name }, + }); + if (operation.Name == "invalid") + { + operation.Context.ScheduleNewOrchestration("Child", new StartOrchestrationOptions("invalid-tags") + { + Tags = new Dictionary { ["valid"] = string.Empty, ["invalid"] = null! }, + }); + } + + return null; + }, "before", "invalid", "after"); + + // Assert + Assert.NotNull(result.Results); + Assert.NotNull(result.Actions); + Assert.Null(result.FailureDetails); + Assert.Equal(3, result.Results.Count); + Assert.Null(result.Results[0].FailureDetails); + Assert.NotNull(result.Results[1].FailureDetails); + Assert.Equal(typeof(ArgumentNullException).FullName, result.Results[1].FailureDetails?.ErrorType); + Assert.Null(result.Results[2].FailureDetails); + Assert.Equal("before", stateAfterFailure); + Assert.Equal("\"after\"", result.EntityState); + result.Actions.Select(action => Assert.IsType(action).InstanceId) + .Should().Equal("before", "after"); + } + + static Task RunAsync(Func run, params string[] operations) + { + TaskEntityShim shim = new( + JsonDataConverter.Default, + new TestEntity(run), + new EntityId("scheduler", "key"), + NullLogger.Instance); + return shim.ExecuteOperationBatchAsync(new EntityBatchRequest + { + InstanceId = "@scheduler@key", + EntityState = "\"initial\"", + Operations = operations.Select(name => new OperationRequest + { + Id = Guid.NewGuid(), + Operation = name, + TraceContext = TraceContext, + }).ToList(), + }); + } + + sealed class TestEntity(Func run) : ITaskEntity + { + readonly Func run = run; + + public ValueTask RunAsync(TaskEntityOperation operation) => new(this.run(operation)); + } +} diff --git a/test/Worker/Grpc.Tests/GrpcEntityRunnerTagTests.cs b/test/Worker/Grpc.Tests/GrpcEntityRunnerTagTests.cs new file mode 100644 index 000000000..57ca3fa83 --- /dev/null +++ b/test/Worker/Grpc.Tests/GrpcEntityRunnerTagTests.cs @@ -0,0 +1,302 @@ +// Copyright (c) Microsoft Corporation. +// Licensed under the MIT License. + +using Google.Protobuf; +using Google.Protobuf.WellKnownTypes; +using Microsoft.DurableTask.Entities; +using Microsoft.Extensions.Caching.Memory; +using P = Microsoft.DurableTask.Protobuf; + +namespace Microsoft.DurableTask.Worker.Grpc.Tests; + +public class GrpcEntityRunnerTagTests +{ + static readonly P.TraceContext TraceContext = new() + { + TraceParent = "00-0123456789abcdef0123456789abcdef-0123456789abcdef-01", + TraceState = "vendor=value", + }; + + [Theory] + [InlineData(false)] + [InlineData(true)] + public async Task LoadAndRunAsync_TaggedStart_PreservesOptionsInResponseAsync(bool optionsOnly) + { + // Arrange + Dictionary tags = new() + { + ["owner"] = "entity", + ["\u6807\u7b7e"] = "\u4f60\u597d \ud83c\udf0d", + ["empty"] = string.Empty, + [string.Empty] = "empty-key", + }; + DateTimeOffset startAt = new(2030, 1, 2, 3, 4, 5, TimeSpan.FromHours(2)); + StartOrchestrationOptions options = new("child", startAt) + { + Tags = tags, + Version = "v2", + }; + DateTimeOffset before = DateTimeOffset.UtcNow; + + // Act + P.EntityBatchResult result = await RunAsync( + operation => optionsOnly + ? operation.Context.ScheduleNewOrchestration("Child", options) + : operation.Context.ScheduleNewOrchestration("Child", "input", options), + "start"); + DateTimeOffset after = DateTimeOffset.UtcNow; + + // Assert + Assert.Null(result.FailureDetails); + P.StartNewOrchestrationAction action = Assert.Single(result.Actions).StartNewOrchestration; + action.Tags.Should().BeEquivalentTo(tags); + Assert.Equal("Child", action.Name); + Assert.Equal("child", action.InstanceId); + Assert.Equal("v2", action.Version); + Assert.Equal(optionsOnly ? null : "\"input\"", action.Input); + Assert.Equal(startAt.UtcDateTime, action.ScheduledTime.ToDateTime()); + Assert.InRange(action.RequestTime.ToDateTimeOffset(), before, after); + Assert.Equal(TraceContext, action.ParentTraceContext); + Assert.Equal("\"child\"", Assert.Single(result.Results).Success.Result); + } + + [Theory] + [InlineData(false, false)] + [InlineData(false, true)] + [InlineData(true, false)] + [InlineData(true, true)] + public async Task LoadAndRunAsync_NullOrEmptyTags_EmitsEmptyMapAsync(bool optionsOnly, bool empty) + { + // Arrange + StartOrchestrationOptions options = new("child") + { + Tags = empty ? new Dictionary() : null!, + }; + + // Act + P.EntityBatchResult result = await RunAsync( + operation => optionsOnly + ? operation.Context.ScheduleNewOrchestration("Child", options) + : operation.Context.ScheduleNewOrchestration("Child", input: null, options), + "start"); + + // Assert + Assert.Empty(Assert.Single(result.Actions).StartNewOrchestration.Tags); + Assert.NotNull(Assert.Single(result.Results).Success); + } + + [Fact] + public async Task LoadAndRunAsync_WithoutOptions_PreservesDefaultsAsync() + { + // Arrange + TaskName name = "Child"; + + // Act + P.EntityBatchResult result = await RunAsync( + operation => operation.Context.ScheduleNewOrchestration(name), + "start"); + + // Assert + P.StartNewOrchestrationAction action = Assert.Single(result.Actions).StartNewOrchestration; + Assert.Empty(action.Tags); + Assert.Null(action.Input); + Assert.Null(action.ScheduledTime); + Assert.Equal(string.Empty, action.Version); + Assert.True(Guid.TryParseExact(action.InstanceId, "N", out _)); + } + + [Theory] + [InlineData(false)] + [InlineData(true)] + public async Task LoadAndRunAsync_ReusedDictionary_PreservesEachSnapshotAsync(bool optionsOnly) + { + // Arrange + Dictionary tags = new() { ["owner"] = "before", ["remove"] = "original" }; + StartOrchestrationOptions options = new("first") { Tags = tags }; + + // Act + P.EntityBatchResult result = await RunAsync(operation => + { + if (optionsOnly) + { + operation.Context.ScheduleNewOrchestration("Child", options); + } + else + { + operation.Context.ScheduleNewOrchestration("Child", "first-input", options); + } + + tags["owner"] = "after"; + tags.Remove("remove"); + tags["added"] = "second"; + options = options with { InstanceId = "second" }; + if (optionsOnly) + { + operation.Context.ScheduleNewOrchestration("Child", options); + } + else + { + operation.Context.ScheduleNewOrchestration("Child", "second-input", options); + } + + tags.Clear(); + return null; + }, "start"); + + // Assert + Assert.Collection( + result.Actions, + first => first.StartNewOrchestration.Tags.Should().BeEquivalentTo( + new Dictionary { ["owner"] = "before", ["remove"] = "original" }), + second => second.StartNewOrchestration.Tags.Should().BeEquivalentTo( + new Dictionary { ["owner"] = "after", ["added"] = "second" })); + } + + [Theory] + [InlineData(false)] + [InlineData(true)] + public async Task LoadAndRunAsync_FailedStart_DiscardsOnlyFailedOperationActionsAsync(bool scheduleThrows) + { + // Arrange + string? stateAfterFailure = null; + + // Act + P.EntityBatchResult result = await RunAsync(operation => + { + if (operation.Name == "after") + { + stateAfterFailure = operation.State.GetState(); + } + + operation.State.SetState(operation.Name); + operation.Context.ScheduleNewOrchestration("Child", new StartOrchestrationOptions(operation.Name) + { + Tags = new Dictionary { ["operation"] = operation.Name }, + }); + if (operation.Name == "fail") + { + if (scheduleThrows) + { + operation.Context.ScheduleNewOrchestration("Child", new StartOrchestrationOptions("@entity@key")); + } + + throw new InvalidOperationException("Failure after scheduling."); + } + + return null; + }, "before", "fail", "after"); + + // Assert + Assert.Null(result.FailureDetails); + Assert.Equal(3, result.Results.Count); + Assert.NotNull(result.Results[0].Success); + Assert.NotNull(result.Results[1].Failure); + Assert.Equal( + scheduleThrows ? typeof(ArgumentException).FullName : typeof(InvalidOperationException).FullName, + result.Results[1].Failure.FailureDetails.ErrorType); + Assert.NotNull(result.Results[2].Success); + Assert.Equal("before", stateAfterFailure); + Assert.Equal("\"after\"", result.EntityState); + Assert.Collection( + result.Actions, + before => + { + Assert.Equal("before", before.StartNewOrchestration.InstanceId); + before.StartNewOrchestration.Tags.Should().BeEquivalentTo( + new Dictionary { ["operation"] = "before" }); + }, + after => + { + Assert.Equal("after", after.StartNewOrchestration.InstanceId); + after.StartNewOrchestration.Tags.Should().BeEquivalentTo( + new Dictionary { ["operation"] = "after" }); + }); + } + + [Fact] + public async Task LoadAndRunAsync_NullTagValue_RollsBackBeforeSerializationAndCachingAsync() + { + // Arrange + using ExtendedSessionsCache cache = new(); + string? stateAfterFailure = null; + + // Act + P.EntityBatchResult result = await RunAsync(operation => + { + if (operation.Name == "after") + { + stateAfterFailure = operation.State.GetState(); + } + + operation.State.SetState(operation.Name); + operation.Context.ScheduleNewOrchestration("Child", new StartOrchestrationOptions(operation.Name) + { + Tags = new Dictionary { ["operation"] = operation.Name }, + }); + if (operation.Name == "invalid") + { + operation.Context.ScheduleNewOrchestration("Child", new StartOrchestrationOptions("invalid-tags") + { + Tags = new Dictionary { ["valid"] = string.Empty, ["invalid"] = null! }, + }); + } + + return null; + }, cache, "before", "invalid", "after"); + + // Assert + Assert.Null(result.FailureDetails); + Assert.Equal(3, result.Results.Count); + Assert.NotNull(result.Results[0].Success); + Assert.NotNull(result.Results[1].Failure); + Assert.Equal(typeof(ArgumentNullException).FullName, result.Results[1].Failure.FailureDetails.ErrorType); + Assert.NotNull(result.Results[2].Success); + Assert.Equal("before", stateAfterFailure); + Assert.Equal("\"after\"", result.EntityState); + result.Actions.Select(action => action.StartNewOrchestration.InstanceId).Should().Equal("before", "after"); + Assert.True(cache.GetOrInitializeCache(30).TryGetValue("@scheduler@key", out string? cachedState)); + Assert.Equal("\"after\"", cachedState); + } + + static Task RunAsync(Func run, params string[] operations) + => RunAsync(run, null, operations); + + static async Task RunAsync( + Func run, ExtendedSessionsCache? cache, params string[] operations) + { + P.EntityBatchRequest request = new() + { + InstanceId = "@scheduler@key", + EntityState = "\"initial\"", + Operations = + { + operations.Select(name => new P.OperationRequest + { + RequestId = Guid.NewGuid().ToString(), + Operation = name, + TraceContext = TraceContext, + }), + }, + }; + request.Properties.Add("IncludeState", Value.ForBool(true)); + if (cache is not null) + { + request.Properties.Add("IsExtendedSession", Value.ForBool(true)); + request.Properties.Add("ExtendedSessionIdleTimeoutInSeconds", Value.ForNumber(30)); + } + + string encodedRequest = Convert.ToBase64String(request.ToByteArray()); + TestEntity entity = new(run); + string response = cache is null + ? await GrpcEntityRunner.LoadAndRunAsync(encodedRequest, entity) + : await GrpcEntityRunner.LoadAndRunAsync(encodedRequest, entity, cache); + return P.EntityBatchResult.Parser.ParseFrom(Convert.FromBase64String(response)); + } + + sealed class TestEntity(Func run) : ITaskEntity + { + readonly Func run = run; + + public ValueTask RunAsync(TaskEntityOperation operation) => new(this.run(operation)); + } +} diff --git a/test/Worker/Grpc.Tests/ProtoUtilsEntityActionTests.cs b/test/Worker/Grpc.Tests/ProtoUtilsEntityActionTests.cs new file mode 100644 index 000000000..f365a8c3b --- /dev/null +++ b/test/Worker/Grpc.Tests/ProtoUtilsEntityActionTests.cs @@ -0,0 +1,151 @@ +// Copyright (c) Microsoft Corporation. +// Licensed under the MIT License. + +using DurableTask.Core.Entities.OperationFormat; +using DurableTask.Core.Tracing; +using Google.Protobuf; +using Google.Protobuf.WellKnownTypes; +using P = Microsoft.DurableTask.Protobuf; + +namespace Microsoft.DurableTask.Worker.Grpc.Tests; + +public class ProtoUtilsEntityActionTests +{ + static readonly DateTime ScheduledTime = new(2030, 1, 2, 3, 4, 5, DateTimeKind.Utc); + static readonly DateTimeOffset RequestTime = new(2029, 1, 2, 3, 4, 5, TimeSpan.Zero); + const string TraceParent = "00-0123456789abcdef0123456789abcdef-0123456789abcdef-01"; + const string TraceState = "vendor=value"; + + [Fact] + public void ToOperationAction_CoreStart_PreservesTagsAndExistingFields() + { + // Arrange + Dictionary tags = new() + { + ["owner"] = "entity", + ["\u6807\u7b7e"] = "\u4f60\u597d \ud83c\udf0d", + ["empty"] = string.Empty, + [string.Empty] = "empty-key", + }; + StartNewOrchestrationOperationAction core = new() + { + InstanceId = "child", + Name = "Child", + Version = "v2", + Input = "\"input\"", + ScheduledStartTime = ScheduledTime, + RequestTime = RequestTime, + ParentTraceContext = new DistributedTraceContext(TraceParent, TraceState), + Tags = tags, + }; + + // Act + P.OperationAction converted = core.ToOperationAction(); + tags.Clear(); + P.StartNewOrchestrationAction wire = P.OperationAction.Parser.ParseFrom(converted.ToByteArray()).StartNewOrchestration; + + // Assert + wire.Tags.Should().BeEquivalentTo(new Dictionary + { + ["owner"] = "entity", + ["\u6807\u7b7e"] = "\u4f60\u597d \ud83c\udf0d", + ["empty"] = string.Empty, + [string.Empty] = "empty-key", + }); + Assert.Equal("child", wire.InstanceId); + Assert.Equal("Child", wire.Name); + Assert.Equal("v2", wire.Version); + Assert.Equal("\"input\"", wire.Input); + Assert.Equal(ScheduledTime, wire.ScheduledTime.ToDateTime()); + Assert.Equal(RequestTime, wire.RequestTime.ToDateTimeOffset()); + Assert.Equal(TraceParent, wire.ParentTraceContext.TraceParent); + Assert.Equal(TraceState, wire.ParentTraceContext.TraceState); + } + + [Fact] + public void ToOperationAction_ProtoStart_PreservesTagsAndExistingFields() + { + // Arrange + P.OperationAction proto = new() + { + StartNewOrchestration = new P.StartNewOrchestrationAction + { + InstanceId = "child", + Name = "Child", + Version = "v2", + Input = "\"input\"", + ScheduledTime = Timestamp.FromDateTime(ScheduledTime), + RequestTime = Timestamp.FromDateTimeOffset(RequestTime), + ParentTraceContext = new P.TraceContext { TraceParent = TraceParent, TraceState = TraceState }, + Tags = { ["owner"] = "entity", ["\u6807\u7b7e"] = "\u4f60\u597d \ud83c\udf0d", ["empty"] = string.Empty, [string.Empty] = "empty-key" }, + }, + }; + P.OperationAction wire = P.OperationAction.Parser.ParseFrom(proto.ToByteArray()); + + // Act + StartNewOrchestrationOperationAction core = Assert.IsType(wire.ToOperationAction()); + wire.StartNewOrchestration.Tags.Clear(); + + // Assert + core.Tags.Should().BeEquivalentTo(new Dictionary + { + ["owner"] = "entity", + ["\u6807\u7b7e"] = "\u4f60\u597d \ud83c\udf0d", + ["empty"] = string.Empty, + [string.Empty] = "empty-key", + }); + Assert.Equal("child", core.InstanceId); + Assert.Equal("Child", core.Name); + Assert.Equal("v2", core.Version); + Assert.Equal("\"input\"", core.Input); + Assert.Equal(ScheduledTime, core.ScheduledStartTime); + Assert.Equal(RequestTime, core.RequestTime); + Assert.Equal(TraceParent, core.ParentTraceContext!.TraceParent); + Assert.Equal(TraceState, core.ParentTraceContext.TraceState); + } + + [Theory] + [InlineData(false)] + [InlineData(true)] + public void ToOperationAction_NullOrEmptyTags_RoundTripsWithoutTags(bool empty) + { + // Arrange + StartNewOrchestrationOperationAction original = new() + { + InstanceId = "child", + Name = "Child", + Tags = empty ? new Dictionary() : null, + }; + + // Act + P.OperationAction proto = original.ToOperationAction(); + StartNewOrchestrationOperationAction restored = Assert.IsType( + P.OperationAction.Parser.ParseFrom(proto.ToByteArray()).ToOperationAction()); + + // Assert + Assert.Empty(proto.StartNewOrchestration.Tags); + Assert.Null(restored.Tags); + Assert.Null(restored.Input); + Assert.Null(restored.Version); + Assert.Null(restored.ScheduledStartTime); + Assert.Null(restored.RequestTime); + Assert.Null(restored.ParentTraceContext); + } + + [Fact] + public void StartNewOrchestrationAction_Tags_UsesCanonicalOptionalMapField() + { + // Arrange + Google.Protobuf.Reflection.MessageDescriptor descriptor = P.StartNewOrchestrationAction.Descriptor; + + // Act + Google.Protobuf.Reflection.FieldDescriptor field = descriptor.FindFieldByName("tags"); + + // Assert + Assert.Equal(8, field.FieldNumber); + Assert.True(field.IsMap); + Assert.Equal(Google.Protobuf.Reflection.FieldType.String, field.MessageType.FindFieldByNumber(1).FieldType); + Assert.Equal(Google.Protobuf.Reflection.FieldType.String, field.MessageType.FindFieldByNumber(2).FieldType); + Assert.Empty(new P.StartNewOrchestrationAction().ToByteArray()); + } +} From 5c304330dde4a2044f1ef35da288966aa31e9a89 Mon Sep 17 00:00:00 2001 From: wangbill Date: Wed, 7 Oct 2026 15:28:28 -0400 Subject: [PATCH 2/2] Keep released Core dependency until prerequisite release Restore Directory.Packages.props to the exact PR baseline. Validate tag propagation only with an explicit session-local build override; update the SDK dependency to a real published Core version after the prerequisite release. Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com> Copilot-Session: 3d59e0dc-efdb-4b0a-b588-6dbd83416193 --- Directory.Packages.props | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/Directory.Packages.props b/Directory.Packages.props index 678ad3308..7b634b8fc 100644 --- a/Directory.Packages.props +++ b/Directory.Packages.props @@ -39,7 +39,7 @@ - +