diff --git a/src/Temporalio.SystemNexus.Generator/Program.cs b/src/Temporalio.SystemNexus.Generator/Program.cs index 728aed10..891414a0 100644 --- a/src/Temporalio.SystemNexus.Generator/Program.cs +++ b/src/Temporalio.SystemNexus.Generator/Program.cs @@ -164,8 +164,6 @@ static void GeneratePayloadVisitor( builder.AppendLine(" [GeneratedCode(\"Temporalio.SystemNexus.Generator\", null)]"); builder.AppendLine(" internal static partial class SystemNexusPayloadVisitor"); builder.AppendLine(" {"); - builder.AppendLine(" private const string TemporalSystemEndpoint = \"__temporal_system\";"); - builder.AppendLine(); builder.AppendLine(" private static readonly IReadOnlyDictionary> EnvelopeVisitors ="); builder.AppendLine(" new Dictionary>"); builder.AppendLine(" {"); @@ -178,14 +176,13 @@ static void GeneratePayloadVisitor( builder.AppendLine(" };"); builder.AppendLine(); - builder.AppendLine(" private static async Task TryVisitAsync("); - builder.AppendLine(" string? endpoint,"); + builder.AppendLine(" internal static async Task TryVisitAsync("); builder.AppendLine(" Payload payload,"); builder.AppendLine(" PayloadVisitor visitPayload,"); builder.AppendLine(" PayloadsVisitor visitPayloads,"); - builder.AppendLine(" EnvelopeVisitor? visitEnvelope)"); + builder.AppendLine(" EnvelopeVisitor? visitEnvelope = null)"); builder.AppendLine(" {"); - builder.AppendLine(" if (!IsSystemNexusEndpoint(endpoint))"); + builder.AppendLine(" if (!IsSystemPayload(payload))"); builder.AppendLine(" {"); builder.AppendLine(" return false;"); builder.AppendLine(" }"); @@ -200,9 +197,6 @@ static void GeneratePayloadVisitor( builder.AppendLine(" return true;"); builder.AppendLine(" }"); builder.AppendLine(); - builder.AppendLine(" internal static bool IsSystemNexusEndpoint(string? endpoint) => endpoint == TemporalSystemEndpoint;"); - builder.AppendLine(); - foreach (var operation in operationMessages) { EmitVisitMethod(builder, operation.Input, messages, containsPayloadMemo, emittedMethods); diff --git a/src/Temporalio/Nexus/SystemNexusPayloadConverter.cs b/src/Temporalio/Nexus/SystemNexusPayloadConverter.cs index bd7dab3a..954fdbd0 100644 --- a/src/Temporalio/Nexus/SystemNexusPayloadConverter.cs +++ b/src/Temporalio/Nexus/SystemNexusPayloadConverter.cs @@ -1,6 +1,7 @@ using System; using Temporalio.Api.Common.V1; using Temporalio.Converters; +using Temporalio.Worker; namespace Temporalio.Nexus { @@ -21,7 +22,9 @@ public Payload ToPayload(object? value) { // TODO: Scope the generated System Nexus support converter context here once the // generated support file is ingested into the SDK. - return OuterPayloadConverter.ToPayload(value); + var payload = OuterPayloadConverter.ToPayload(value); + SystemNexusPayloadVisitor.MarkSystemPayload(payload); + return payload; } /// diff --git a/src/Temporalio/Worker/Generated/SystemNexusPayloadVisitor.cs b/src/Temporalio/Worker/Generated/SystemNexusPayloadVisitor.cs index f71ffa92..3bab8e5d 100644 --- a/src/Temporalio/Worker/Generated/SystemNexusPayloadVisitor.cs +++ b/src/Temporalio/Worker/Generated/SystemNexusPayloadVisitor.cs @@ -13,8 +13,6 @@ namespace Temporalio.Worker [GeneratedCode("Temporalio.SystemNexus.Generator", null)] internal static partial class SystemNexusPayloadVisitor { - private const string TemporalSystemEndpoint = "__temporal_system"; - private static readonly IReadOnlyDictionary> EnvelopeVisitors = new Dictionary> { @@ -34,14 +32,13 @@ internal static partial class SystemNexusPayloadVisitor visitEnvelope), }; - private static async Task TryVisitAsync( - string? endpoint, + internal static async Task TryVisitAsync( Payload payload, PayloadVisitor visitPayload, PayloadsVisitor visitPayloads, - EnvelopeVisitor? visitEnvelope) + EnvelopeVisitor? visitEnvelope = null) { - if (!IsSystemNexusEndpoint(endpoint)) + if (!IsSystemPayload(payload)) { return false; } @@ -56,8 +53,6 @@ private static async Task TryVisitAsync( return true; } - internal static bool IsSystemNexusEndpoint(string? endpoint) => endpoint == TemporalSystemEndpoint; - private static async Task Visit_temporal_api_common_v1_Memo( global::Temporalio.Api.Common.V1.Memo value, PayloadVisitor visitPayload, diff --git a/src/Temporalio/Worker/IWorkflowCodecHelperInstance.cs b/src/Temporalio/Worker/IWorkflowCodecHelperInstance.cs index 31a51ad3..3b891241 100644 --- a/src/Temporalio/Worker/IWorkflowCodecHelperInstance.cs +++ b/src/Temporalio/Worker/IWorkflowCodecHelperInstance.cs @@ -40,18 +40,5 @@ internal interface IWorkflowCodecHelperInstance /// Sequence. /// Context. ISerializationContext.Workflow? GetPendingExternalSignalSerializationContext(uint seq); - - /// - /// Gets the pending Nexus operation info for the given sequence. - /// - /// Sequence. - /// Info. - NexusOperationInfo? GetPendingNexusOperationInfo(uint seq); - - /// - /// Pending Nexus operation info. - /// - /// Endpoint name. - internal record NexusOperationInfo(string? Endpoint); } } diff --git a/src/Temporalio/Worker/SystemNexusPayloadVisitor.cs b/src/Temporalio/Worker/SystemNexusPayloadVisitor.cs index 9377f661..6881a256 100644 --- a/src/Temporalio/Worker/SystemNexusPayloadVisitor.cs +++ b/src/Temporalio/Worker/SystemNexusPayloadVisitor.cs @@ -11,27 +11,22 @@ namespace Temporalio.Worker { internal static partial class SystemNexusPayloadVisitor { + private const string TemporalSystemEndpoint = "__temporal_system"; + + private const string SystemPayloadMetadataKey = "__temporal_system_payload"; + + private static readonly ByteString SystemPayloadMetadataValue = ByteString.CopyFromUtf8("true"); + internal delegate Task PayloadVisitor(Payload payload); internal delegate Task PayloadsVisitor(RepeatedField payloads); internal delegate Task EnvelopeVisitor(Payload payload); - internal static Task TryVisitInputAsync( - string? endpoint, - Payload payload, - PayloadVisitor visitPayload, - PayloadsVisitor visitPayloads, - EnvelopeVisitor? visitEnvelope = null) => - TryVisitAsync(endpoint, payload, visitPayload, visitPayloads, visitEnvelope); + internal static bool IsSystemNexusEndpoint(string? endpoint) => endpoint == TemporalSystemEndpoint; - internal static Task TryVisitOutputAsync( - string? endpoint, - Payload payload, - PayloadVisitor visitPayload, - PayloadsVisitor visitPayloads, - EnvelopeVisitor? visitEnvelope = null) => - TryVisitAsync(endpoint, payload, visitPayload, visitPayloads, visitEnvelope); + internal static void MarkSystemPayload(Payload payload) => + payload.Metadata[SystemPayloadMetadataKey] = SystemPayloadMetadataValue; private static async Task VisitEnvelopeAsync( Payload payload, @@ -48,11 +43,16 @@ private static async Task VisitEnvelopeAsync( payload.Metadata.Clear(); payload.Metadata["encoding"] = ByteString.CopyFromUtf8("binary/protobuf"); payload.Metadata["messageType"] = ByteString.CopyFromUtf8(message.Descriptor.FullName); + MarkSystemPayload(payload); payload.Data = message.ToByteString(); if (visitEnvelope != null) { await visitEnvelope(payload).ConfigureAwait(false); } } + + private static bool IsSystemPayload(Payload payload) => + payload.Metadata.TryGetValue(SystemPayloadMetadataKey, out var value) && + value.Equals(SystemPayloadMetadataValue); } } diff --git a/src/Temporalio/Worker/WorkflowCodecHelper.cs b/src/Temporalio/Worker/WorkflowCodecHelper.cs index 95804a39..61680f72 100644 --- a/src/Temporalio/Worker/WorkflowCodecHelper.cs +++ b/src/Temporalio/Worker/WorkflowCodecHelper.cs @@ -1,3 +1,4 @@ +using System.Collections.Generic; using System.Linq; using System.Threading.Tasks; using Google.Protobuf; @@ -125,20 +126,9 @@ await childCodec2.DecodeFailureAsync( } if (job.ResolveNexusOperation.Result.Completed != null) { - var operationInfo = context.Instance?.GetPendingNexusOperationInfo( - job.ResolveNexusOperation.Seq); - if (operationInfo == null || - !await SystemNexusPayloadVisitor.TryVisitOutputAsync( - operationInfo.Endpoint, - job.ResolveNexusOperation.Result.Completed, - payload => DecodeAsync(nexusCodec, payload), - payloads => DecodeAsync(nexusCodec, payloads)). - ConfigureAwait(false)) - { - await DecodeAsync( - nexusCodec, job.ResolveNexusOperation.Result.Completed). - ConfigureAwait(false); - } + await DecodeAsync( + nexusCodec, job.ResolveNexusOperation.Result.Completed). + ConfigureAwait(false); } else if (job.ResolveNexusOperation.Result.Failed != null) { @@ -347,16 +337,7 @@ await EncodeAsync( codec = context.CodecNoContext; if (cmd.ScheduleNexusOperation.Input != null && codec != null) { - if (!await SystemNexusPayloadVisitor.TryVisitInputAsync( - cmd.ScheduleNexusOperation.Endpoint, - cmd.ScheduleNexusOperation.Input, - payload => EncodeAsync(codec, payload), - payloads => EncodeAsync(codec, payloads)). - ConfigureAwait(false)) - { - await EncodeAsync( - codec, cmd.ScheduleNexusOperation.Input).ConfigureAwait(false); - } + await EncodeAsync(codec, cmd.ScheduleNexusOperation.Input).ConfigureAwait(false); } break; case WorkflowCommand.VariantOneofCase.StartChildWorkflowExecution: @@ -429,15 +410,43 @@ private static async Task EncodeAsync( { return; } - // We have to convert to list here just in case they are based on the underlying list - // and we clear it out (which can happen with Linq selectors) - var newPayloads = (await codec.EncodeAsync(payloads).ConfigureAwait(false)).ToList(); + var newPayloads = new List(); + var codecPayloads = new List(); + foreach (var payload in payloads) + { + if (!await SystemNexusPayloadVisitor.TryVisitAsync( + payload, + payload => EncodeAsync(codec, payload), + nestedPayloads => EncodeAsync(codec, nestedPayloads)).ConfigureAwait(false)) + { + codecPayloads.Add(payload); + continue; + } + + if (codecPayloads.Count > 0) + { + newPayloads.AddRange(await codec.EncodeAsync(codecPayloads).ConfigureAwait(false)); + codecPayloads.Clear(); + } + newPayloads.Add(payload); + } + if (codecPayloads.Count > 0) + { + newPayloads.AddRange(await codec.EncodeAsync(codecPayloads).ConfigureAwait(false)); + } payloads.Clear(); payloads.AddRange(newPayloads); } private static async Task EncodeAsync(IPayloadCodec codec, Payload payload) { + if (await SystemNexusPayloadVisitor.TryVisitAsync( + payload, + nestedPayload => EncodeAsync(codec, nestedPayload), + nestedPayloads => EncodeAsync(codec, nestedPayloads)).ConfigureAwait(false)) + { + return; + } // We are gonna require a single result here. It is important that we do Single() call // before clearing out payload to merge with since underlying enumerable may be lazy. // If the returned payload is literally the same object as the one sent to the codec, @@ -538,15 +547,43 @@ private static async Task DecodeAsync(IPayloadCodec codec, RepeatedField(); + var codecPayloads = new List(); + foreach (var payload in payloads) + { + if (!await SystemNexusPayloadVisitor.TryVisitAsync( + payload, + payload => DecodeAsync(codec, payload), + nestedPayloads => DecodeAsync(codec, nestedPayloads)).ConfigureAwait(false)) + { + codecPayloads.Add(payload); + continue; + } + + if (codecPayloads.Count > 0) + { + newPayloads.AddRange(await codec.DecodeAsync(codecPayloads).ConfigureAwait(false)); + codecPayloads.Clear(); + } + newPayloads.Add(payload); + } + if (codecPayloads.Count > 0) + { + newPayloads.AddRange(await codec.DecodeAsync(codecPayloads).ConfigureAwait(false)); + } payloads.Clear(); payloads.AddRange(newPayloads); } private static async Task DecodeAsync(IPayloadCodec codec, Payload payload) { + if (await SystemNexusPayloadVisitor.TryVisitAsync( + payload, + nestedPayload => DecodeAsync(codec, nestedPayload), + nestedPayloads => DecodeAsync(codec, nestedPayloads)).ConfigureAwait(false)) + { + return; + } // We are gonna require a single result here. // Similarly with encode, we leave the payload alone if it's exactly the same object as the original. var decoded = await codec.DecodeAsync(new Payload[] { payload }).ConfigureAwait(false); diff --git a/src/Temporalio/Worker/WorkflowInstance.cs b/src/Temporalio/Worker/WorkflowInstance.cs index b40d4ca5..e8b82f4a 100644 --- a/src/Temporalio/Worker/WorkflowInstance.cs +++ b/src/Temporalio/Worker/WorkflowInstance.cs @@ -898,13 +898,6 @@ public WorkflowActivationCompletion Activate(WorkflowActivation act) return pending?.SerializationContext; } - /// - public IWorkflowCodecHelperInstance.NexusOperationInfo? GetPendingNexusOperationInfo(uint seq) - { - nexusOperationsPending.TryGetValue(seq, out var pending); - return pending == null ? null : new(pending.Endpoint); - } - /// protected override IEnumerable? GetScheduledTasks() => scheduledTasks; @@ -2735,7 +2728,6 @@ public override Task> ScheduleNexusOperati var handleSource = new TaskCompletionSource>(); var pending = new PendingNexusOperationInfo( - Endpoint: input.ClientOptions.Endpoint, StartCompletionSource: new(), ResultCompletionSource: new()); instance.nexusOperationsPending[seq] = pending; @@ -3210,7 +3202,6 @@ private record PendingExternalCancel( TaskCompletionSource CompletionSource); private record PendingNexusOperationInfo( - string? Endpoint, TaskCompletionSource StartCompletionSource, TaskCompletionSource ResultCompletionSource); diff --git a/tests/Temporalio.Tests/Worker/WorkflowCodecHelperTests.cs b/tests/Temporalio.Tests/Worker/WorkflowCodecHelperTests.cs index 3b727869..38d720f5 100644 --- a/tests/Temporalio.Tests/Worker/WorkflowCodecHelperTests.cs +++ b/tests/Temporalio.Tests/Worker/WorkflowCodecHelperTests.cs @@ -3,7 +3,9 @@ namespace Temporalio.Tests.Worker; using Google.Protobuf; using Google.Protobuf.Collections; using Temporalio.Api.Common.V1; +using Temporalio.Api.WorkflowService.V1; using Temporalio.Bridge.Api.WorkflowActivation; +using Temporalio.Bridge.Api.WorkflowCommands; using Temporalio.Bridge.Api.WorkflowCompletion; using Temporalio.Converters; using Temporalio.Worker; @@ -104,6 +106,202 @@ await CreateAndVisitPayload(new(), comp, async (ctx, payload) => }); } + [Fact] + public async Task EncodeAsync_SystemNexusEnvelopeInGenericPayloadField_EncodesNestedPayload() + { + var request = new SignalWithStartWorkflowExecutionRequest + { + Input = new() { Payloads_ = { new Payload { Data = ByteString.CopyFromUtf8("input") } } }, + }; + var envelope = CreateSystemEnvelope(request); + var completion = new WorkflowActivationCompletion + { + Successful = new() + { + Commands = + { + new WorkflowCommand { UpdateResponse = new() { Completed = envelope } }, + }, + }, + }; + + await WorkflowCodecHelper.EncodeAsync(CreateSimpleCodecContext(new MarkerPayloadCodec()), completion); + + Assert.Equal( + ByteString.CopyFromUtf8("true"), + completion.Successful.Commands[0].UpdateResponse.Completed.Metadata[ + "__temporal_system_payload"]); + Assert.DoesNotContain( + "encoded", + completion.Successful.Commands[0].UpdateResponse.Completed.Metadata.Keys); + var encodedRequest = SignalWithStartWorkflowExecutionRequest.Parser.ParseFrom( + completion.Successful.Commands[0].UpdateResponse.Completed.Data); + Assert.Contains("encoded", encodedRequest.Input.Payloads_[0].Metadata.Keys); + } + + [Fact] + public async Task EncodeAsync_UnmarkedSystemNexusEnvelopeInGenericPayloadField_EncodesEnvelope() + { + var request = new SignalWithStartWorkflowExecutionRequest + { + Input = new() { Payloads_ = { new Payload { Data = ByteString.CopyFromUtf8("input") } } }, + }; + var envelope = CreateSystemEnvelope(request); + envelope.Metadata.Remove("__temporal_system_payload"); + var completion = new WorkflowActivationCompletion + { + Successful = new() + { + Commands = + { + new WorkflowCommand { UpdateResponse = new() { Completed = envelope } }, + }, + }, + }; + + await WorkflowCodecHelper.EncodeAsync(CreateSimpleCodecContext(new MarkerPayloadCodec()), completion); + + Assert.Contains( + "encoded", + completion.Successful.Commands[0].UpdateResponse.Completed.Metadata.Keys); + var encodedRequest = SignalWithStartWorkflowExecutionRequest.Parser.ParseFrom( + completion.Successful.Commands[0].UpdateResponse.Completed.Data); + Assert.DoesNotContain("encoded", encodedRequest.Input.Payloads_[0].Metadata.Keys); + } + + [Fact] + public async Task DecodeAsync_SystemNexusEnvelopeInGenericPayloadField_DecodesNestedPayload() + { + var nestedPayload = new Payload { Data = ByteString.CopyFromUtf8("input") }; + nestedPayload.Metadata["encoded"] = ByteString.Empty; + var request = new SignalWithStartWorkflowExecutionRequest + { + Input = new() { Payloads_ = { nestedPayload } }, + }; + var envelope = CreateSystemEnvelope(request); + var activation = new WorkflowActivation + { + Jobs = + { + new WorkflowActivationJob { DoUpdate = new() { Input = { envelope } } }, + }, + }; + + await WorkflowCodecHelper.DecodeAsync(CreateSimpleCodecContext(new MarkerPayloadCodec()), activation); + + Assert.DoesNotContain("decoded", activation.Jobs[0].DoUpdate.Input[0].Metadata.Keys); + var decodedRequest = SignalWithStartWorkflowExecutionRequest.Parser.ParseFrom( + activation.Jobs[0].DoUpdate.Input[0].Data); + Assert.Contains("decoded", decodedRequest.Input.Payloads_[0].Metadata.Keys); + } + + [Fact] + public async Task EncodeAsync_AllSystemPayloads_EncodesNestedPayloads() + { + var firstEnvelope = CreateSystemEnvelope("first"); + var secondEnvelope = CreateSystemEnvelope("second"); + var completion = new WorkflowActivationCompletion + { + Successful = new() + { + Commands = + { + new WorkflowCommand + { + ScheduleActivity = new() + { + Arguments = { firstEnvelope, secondEnvelope }, + }, + }, + }, + }, + }; + + await WorkflowCodecHelper.EncodeAsync(CreateSimpleCodecContext(new PackingPayloadCodec()), completion); + + var arguments = completion.Successful.Commands[0].ScheduleActivity.Arguments; + Assert.Equal(2, arguments.Count); + Assert.All(arguments, payload => + { + Assert.Equal(ByteString.CopyFromUtf8("true"), payload.Metadata["__temporal_system_payload"]); + var request = SignalWithStartWorkflowExecutionRequest.Parser.ParseFrom(payload.Data); + Assert.Contains("packed", request.Input.Payloads_[0].Metadata.Keys); + }); + } + + [Fact] + public async Task EncodeAndDecodeAsync_MixedPayloads_PreservesSystemPayloadBoundaries() + { + var envelope = CreateSystemEnvelope("system"); + var completion = new WorkflowActivationCompletion + { + Successful = new() + { + Commands = + { + new WorkflowCommand + { + ScheduleActivity = new() + { + Arguments = + { + new Payload { Data = ByteString.CopyFromUtf8("first") }, + new Payload { Data = ByteString.CopyFromUtf8("second") }, + envelope, + new Payload { Data = ByteString.CopyFromUtf8("third") }, + new Payload { Data = ByteString.CopyFromUtf8("fourth") }, + }, + }, + }, + }, + }, + }; + + var codec = new PackingPayloadCodec(); + await WorkflowCodecHelper.EncodeAsync(CreateSimpleCodecContext(codec), completion); + + var encodedArguments = completion.Successful.Commands[0].ScheduleActivity.Arguments; + Assert.Equal(3, encodedArguments.Count); + Assert.Equal("first|second", encodedArguments[0].Data.ToStringUtf8()); + Assert.Equal(ByteString.CopyFromUtf8("true"), encodedArguments[1].Metadata["__temporal_system_payload"]); + Assert.Equal("third|fourth", encodedArguments[2].Data.ToStringUtf8()); + + var activation = new WorkflowActivation + { + Jobs = + { + new WorkflowActivationJob + { + DoUpdate = new() { Input = { encodedArguments } }, + }, + }, + }; + await WorkflowCodecHelper.DecodeAsync(CreateSimpleCodecContext(codec), activation); + + var decodedPayloads = activation.Jobs[0].DoUpdate.Input; + Assert.Equal(5, decodedPayloads.Count); + Assert.Equal("first", decodedPayloads[0].Data.ToStringUtf8()); + Assert.Equal("second", decodedPayloads[1].Data.ToStringUtf8()); + Assert.Equal(ByteString.CopyFromUtf8("true"), decodedPayloads[2].Metadata["__temporal_system_payload"]); + Assert.Equal("third", decodedPayloads[3].Data.ToStringUtf8()); + Assert.Equal("fourth", decodedPayloads[4].Data.ToStringUtf8()); + var decodedRequest = SignalWithStartWorkflowExecutionRequest.Parser.ParseFrom(decodedPayloads[2].Data); + Assert.Equal("system", decodedRequest.Input.Payloads_[0].Data.ToStringUtf8()); + } + + private static Payload CreateSystemEnvelope(string value) => CreateSystemEnvelope( + new SignalWithStartWorkflowExecutionRequest + { + Input = new() { Payloads_ = { new Payload { Data = ByteString.CopyFromUtf8(value) } } }, + }); + + private static Payload CreateSystemEnvelope(IMessage message) + { + Assert.True(new BinaryProtoConverter().TryToPayload(message, out var payload)); + SystemNexusPayloadVisitor.MarkSystemPayload(payload!); + return payload!; + } + private static WorkflowCodecHelper.WorkflowCodecContext CreateSimpleCodecContext(IPayloadCodec codec) => new( CodecNoContext: codec, CodecWorkflowContext: codec, @@ -223,4 +421,22 @@ public Task> DecodeAsync(IReadOnlyCollection> EncodeAsync(IReadOnlyCollection payloads) => + Task.FromResult>(new[] + { + new Payload + { + Data = ByteString.CopyFromUtf8(string.Join("|", payloads.Select(p => p.Data.ToStringUtf8()))), + Metadata = { ["packed"] = ByteString.CopyFromUtf8(payloads.Count.ToString()) }, + }, + }); + + public Task> DecodeAsync(IReadOnlyCollection payloads) => + Task.FromResult>(payloads.SelectMany(payload => + payload.Data.ToStringUtf8().Split('|').Select(value => + new Payload { Data = ByteString.CopyFromUtf8(value) })).ToList()); + } }