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()); + } +}