From 4ec540ef46ede78fad0f48cdcf9e7c9d9f844419 Mon Sep 17 00:00:00 2001 From: Tim Conley Date: Mon, 17 Aug 2026 15:12:18 -0700 Subject: [PATCH 1/2] Add System Nexus transfer converter context --- CHANGELOG.md | 6 + .../Program.cs | 2 +- .../Nexus/SystemNexusConverterContext.cs | 73 +++++++++++ .../Nexus/SystemNexusPayloadConverter.cs | 50 ++++++++ .../Generated/SystemNexusPayloadVisitor.cs | 2 +- .../Worker/SystemNexusPayloadVisitor.cs | 22 ---- src/Temporalio/Worker/WorkflowInstance.cs | 15 +-- .../Nexus/SystemNexusPayloadConverterTests.cs | 94 +++++++++++++++ .../Worker/NexusWorkerTests.cs | 2 +- .../Worker/SystemNexusTests.cs | 114 ++++++++++++++++++ .../Worker/WorkflowWorkerTests.cs | 83 ------------- 11 files changed, 348 insertions(+), 115 deletions(-) create mode 100644 src/Temporalio/Nexus/SystemNexusConverterContext.cs create mode 100644 src/Temporalio/Nexus/SystemNexusPayloadConverter.cs create mode 100644 tests/Temporalio.Tests/Nexus/SystemNexusPayloadConverterTests.cs create mode 100644 tests/Temporalio.Tests/Worker/SystemNexusTests.cs diff --git a/CHANGELOG.md b/CHANGELOG.md index 28a30f9c..0cc4e84c 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -19,6 +19,12 @@ to docs, or any other relevant information. ## [Unreleased] +### Added + +- Added `SystemNexusConverterContext`, which exposes the application's original payload and failure + converters while a System Nexus transfer type converter is executing. This lets generated System + Nexus transfer types serialize nested values without recursively applying their own transfer hooks. + ### Changed - A non-retryable `ApplicationFailureException` with error type `PayloadValidationError` thrown by a diff --git a/src/Temporalio.SystemNexus.Generator/Program.cs b/src/Temporalio.SystemNexus.Generator/Program.cs index d4ec9407..728aed10 100644 --- a/src/Temporalio.SystemNexus.Generator/Program.cs +++ b/src/Temporalio.SystemNexus.Generator/Program.cs @@ -200,7 +200,7 @@ static void GeneratePayloadVisitor( builder.AppendLine(" return true;"); builder.AppendLine(" }"); builder.AppendLine(); - builder.AppendLine(" private static bool IsSystemNexusEndpoint(string? endpoint) => endpoint == TemporalSystemEndpoint;"); + builder.AppendLine(" internal static bool IsSystemNexusEndpoint(string? endpoint) => endpoint == TemporalSystemEndpoint;"); builder.AppendLine(); foreach (var operation in operationMessages) diff --git a/src/Temporalio/Nexus/SystemNexusConverterContext.cs b/src/Temporalio/Nexus/SystemNexusConverterContext.cs new file mode 100644 index 00000000..105c4a66 --- /dev/null +++ b/src/Temporalio/Nexus/SystemNexusConverterContext.cs @@ -0,0 +1,73 @@ +using System; +using System.Threading; +using Temporalio.Converters; + +namespace Temporalio.Nexus +{ + /// + /// Provides the user configured converters while System Nexus transfer conversion runs. + /// + /// + /// This context is only available while a System Nexus workflow operation converts its input + /// or result through a Temporal transfer type converter. + /// + public static class SystemNexusConverterContext + { + private static readonly AsyncLocal CurrentLocal = new(); + + /// + /// Gets the application's payload converter for the current System Nexus conversion. + /// + /// + /// Thrown when called outside a System Nexus transfer conversion. + /// + public static IPayloadConverter PayloadConverter => Current.PayloadConverter; + + /// + /// Gets the application's failure converter for the current System Nexus conversion. + /// + /// + /// Thrown when called outside a System Nexus transfer conversion. + /// + public static IFailureConverter FailureConverter => Current.FailureConverter; + + private static ConverterContext Current => CurrentLocal.Value ?? throw new InvalidOperationException( + "The System Nexus converter context is only available while a System Nexus transfer type converter is executing."); + + /// + /// Sets the converters for the duration of the returned scope. + /// + /// The application's payload converter. + /// The application's failure converter. + /// A scope that restores the preceding converter context. + internal static IDisposable Push( + IPayloadConverter payloadConverter, + IFailureConverter failureConverter) + { + var previous = CurrentLocal.Value; + CurrentLocal.Value = new(payloadConverter, failureConverter); + return new PopOnDispose(previous); + } + + private sealed record ConverterContext( + IPayloadConverter PayloadConverter, + IFailureConverter FailureConverter); + + private sealed class PopOnDispose : IDisposable + { + private readonly ConverterContext? previous; + private bool disposed; + + internal PopOnDispose(ConverterContext? previous) => this.previous = previous; + + public void Dispose() + { + if (!disposed) + { + CurrentLocal.Value = previous; + disposed = true; + } + } + } + } +} diff --git a/src/Temporalio/Nexus/SystemNexusPayloadConverter.cs b/src/Temporalio/Nexus/SystemNexusPayloadConverter.cs new file mode 100644 index 00000000..8bf2f4c6 --- /dev/null +++ b/src/Temporalio/Nexus/SystemNexusPayloadConverter.cs @@ -0,0 +1,50 @@ +using System; +using Temporalio.Converters; + +namespace Temporalio.Nexus +{ + /// + /// Payload converter for System Nexus outer protobuf envelopes. + /// + /// + /// This converter applies transfer type conversion to the outer System Nexus envelope while + /// making the application's converters available to generated transfer types. + /// + internal sealed class SystemNexusPayloadConverter : IPayloadConverter + { + private readonly IPayloadConverter userPayloadConverter; + private readonly IFailureConverter userFailureConverter; + private readonly IPayloadConverter outerPayloadConverter; + + /// + /// Initializes a new instance of the class. + /// + /// The application's payload converter. + /// The application's failure converter. + internal SystemNexusPayloadConverter( + IPayloadConverter userPayloadConverter, + IFailureConverter userFailureConverter) + { + this.userPayloadConverter = userPayloadConverter; + this.userFailureConverter = userFailureConverter; + outerPayloadConverter = TemporalTransferTypePayloadConverter.Wrap( + new DefaultPayloadConverter(new BinaryProtoConverter())); + } + + /// + public Temporalio.Api.Common.V1.Payload ToPayload(object? value) + { + using var context = SystemNexusConverterContext.Push( + userPayloadConverter, userFailureConverter); + return outerPayloadConverter.ToPayload(value); + } + + /// + public object? ToValue(Temporalio.Api.Common.V1.Payload payload, Type type) + { + using var context = SystemNexusConverterContext.Push( + userPayloadConverter, userFailureConverter); + return outerPayloadConverter.ToValue(payload, type); + } + } +} diff --git a/src/Temporalio/Worker/Generated/SystemNexusPayloadVisitor.cs b/src/Temporalio/Worker/Generated/SystemNexusPayloadVisitor.cs index e0479232..f71ffa92 100644 --- a/src/Temporalio/Worker/Generated/SystemNexusPayloadVisitor.cs +++ b/src/Temporalio/Worker/Generated/SystemNexusPayloadVisitor.cs @@ -56,7 +56,7 @@ private static async Task TryVisitAsync( return true; } - private static bool IsSystemNexusEndpoint(string? endpoint) => endpoint == TemporalSystemEndpoint; + 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, diff --git a/src/Temporalio/Worker/SystemNexusPayloadVisitor.cs b/src/Temporalio/Worker/SystemNexusPayloadVisitor.cs index 0ca62f7d..9377f661 100644 --- a/src/Temporalio/Worker/SystemNexusPayloadVisitor.cs +++ b/src/Temporalio/Worker/SystemNexusPayloadVisitor.cs @@ -11,34 +11,12 @@ namespace Temporalio.Worker { internal static partial class SystemNexusPayloadVisitor { - private static readonly BinaryProtoConverter ProtoPayloadConverter = new(); - internal delegate Task PayloadVisitor(Payload payload); internal delegate Task PayloadsVisitor(RepeatedField payloads); internal delegate Task EnvelopeVisitor(Payload payload); - internal static bool TryToInputPayload( - string? endpoint, - object? value, - out Payload payload) - { - payload = null!; - if (!IsSystemNexusEndpoint(endpoint)) - { - return false; - } - - if (!ProtoPayloadConverter.TryToPayload(value, out var converted) || converted == null) - { - return false; - } - - payload = converted; - return true; - } - internal static Task TryVisitInputAsync( string? endpoint, Payload payload, diff --git a/src/Temporalio/Worker/WorkflowInstance.cs b/src/Temporalio/Worker/WorkflowInstance.cs index 34b7f066..d8d547b8 100644 --- a/src/Temporalio/Worker/WorkflowInstance.cs +++ b/src/Temporalio/Worker/WorkflowInstance.cs @@ -22,6 +22,7 @@ using Temporalio.Common; using Temporalio.Converters; using Temporalio.Exceptions; +using Temporalio.Nexus; using Temporalio.Runtime; using Temporalio.Worker.Interceptors; using Temporalio.Workflows; @@ -2688,15 +2689,15 @@ public override Task> ScheduleNexusOperati } // TODO(cretz): Support Nexus serialization context - var payloadConverter = instance.payloadConverterNoContext; + var payloadConverter = SystemNexusPayloadVisitor.IsSystemNexusEndpoint( + input.ClientOptions.Endpoint) ? + new SystemNexusPayloadConverter( + instance.payloadConverterNoContext, + instance.failureConverterNoContext) : + instance.payloadConverterNoContext; var seq = ++instance.nexusOperationCounter; - var inputPayload = SystemNexusPayloadVisitor.TryToInputPayload( - input.ClientOptions.Endpoint, - input.Arg, - out var systemNexusInputPayload) ? - systemNexusInputPayload : - payloadConverter.ToPayload(input.Arg); + var inputPayload = payloadConverter.ToPayload(input.Arg); var cmd = new ScheduleNexusOperation() { Seq = seq, diff --git a/tests/Temporalio.Tests/Nexus/SystemNexusPayloadConverterTests.cs b/tests/Temporalio.Tests/Nexus/SystemNexusPayloadConverterTests.cs new file mode 100644 index 00000000..11923432 --- /dev/null +++ b/tests/Temporalio.Tests/Nexus/SystemNexusPayloadConverterTests.cs @@ -0,0 +1,94 @@ +namespace Temporalio.Tests.Nexus; + +using Google.Protobuf; +using Temporalio.Api.Common.V1; +using Temporalio.Converters; +using Temporalio.Nexus; +using Xunit; +using ApiWorkflowService = Temporalio.Api.WorkflowService.V1; + +public class SystemNexusPayloadConverterTests +{ + [Fact] + public void TransferTypeWithEmbeddedPayload_RoundTrips() + { + var userPayloadConverter = TemporalTransferTypePayloadConverter.Wrap( + new StringPayloadConverter()); + var converter = new SystemNexusPayloadConverter( + userPayloadConverter, new DefaultFailureConverter()); + var value = new SystemNexusRequest("embedded-value"); + + var payload = converter.ToPayload(value); + + Assert.Equal("binary/protobuf", payload.Metadata["encoding"].ToStringUtf8()); + var transferType = ApiWorkflowService.SignalWithStartWorkflowExecutionRequest.Parser.ParseFrom( + payload.Data); + Assert.Equal("test/string", transferType.Input.Payloads_[0].Metadata["encoding"].ToStringUtf8()); + Assert.Equal(value, converter.ToValue(payload, typeof(SystemNexusRequest))); + Assert.Throws(() => + _ = SystemNexusConverterContext.PayloadConverter); + } + + [TemporalTransferTypeConverter(typeof(SystemNexusRequestConverter))] + public sealed record SystemNexusRequest(string EmbeddedValue); + + public sealed class SystemNexusRequestConverter : ITemporalTransferTypeConverter + { + public Type TransferType => typeof(ApiWorkflowService.SignalWithStartWorkflowExecutionRequest); + + public object ToTransferType(object? value) + { + Assert.IsType( + SystemNexusConverterContext.PayloadConverter); + Assert.IsType(SystemNexusConverterContext.FailureConverter); + var request = (SystemNexusRequest)value!; + var input = new Payloads(); + input.Payloads_.Add(SystemNexusConverterContext.PayloadConverter.ToPayload( + new EmbeddedValue(request.EmbeddedValue))); + return new ApiWorkflowService.SignalWithStartWorkflowExecutionRequest { Input = input }; + } + + public object FromTransferType(object? transferType) + { + Assert.IsType( + SystemNexusConverterContext.PayloadConverter); + Assert.IsType(SystemNexusConverterContext.FailureConverter); + var request = (ApiWorkflowService.SignalWithStartWorkflowExecutionRequest)transferType!; + var embedded = SystemNexusConverterContext.PayloadConverter.ToValue( + request.Input.Payloads_[0]); + return new SystemNexusRequest(embedded.Value); + } + } + + [TemporalTransferTypeConverter(typeof(EmbeddedValueConverter))] + public sealed record EmbeddedValue(string Value); + + public sealed class EmbeddedValueConverter : ITemporalTransferTypeConverter + { + public Type TransferType => typeof(string); + + public object ToTransferType(object? value) => ((EmbeddedValue)value!).Value; + + public object FromTransferType(object? transferType) => new EmbeddedValue((string)transferType!); + } + + private sealed class StringPayloadConverter : IPayloadConverter + { + public Payload ToPayload(object? value) + { + var stringValue = Assert.IsType(value); + return new() + { + Metadata = { ["encoding"] = ByteString.CopyFromUtf8("test/string") }, + Data = ByteString.CopyFromUtf8(stringValue), + }; + } + + public object? ToValue(Payload payload, Type type) + { + Assert.Equal(typeof(string), type); + Assert.Equal("test/string", payload.Metadata["encoding"].ToStringUtf8()); + return payload.Data.ToStringUtf8(); + } + } +} diff --git a/tests/Temporalio.Tests/Worker/NexusWorkerTests.cs b/tests/Temporalio.Tests/Worker/NexusWorkerTests.cs index 012989fd..d31e5a4b 100644 --- a/tests/Temporalio.Tests/Worker/NexusWorkerTests.cs +++ b/tests/Temporalio.Tests/Worker/NexusWorkerTests.cs @@ -3115,4 +3115,4 @@ private async Task> RunInWorkflowAsy return handle; }); } -} \ No newline at end of file +} diff --git a/tests/Temporalio.Tests/Worker/SystemNexusTests.cs b/tests/Temporalio.Tests/Worker/SystemNexusTests.cs new file mode 100644 index 00000000..f8775b0c --- /dev/null +++ b/tests/Temporalio.Tests/Worker/SystemNexusTests.cs @@ -0,0 +1,114 @@ +namespace Temporalio.Tests.Worker; + +using Temporalio.Api.Enums.V1; +using Temporalio.Client; +using Temporalio.Converters; +using Temporalio.Tests.Converters; +using Temporalio.Worker; +using Temporalio.Worker.Interceptors; +using Temporalio.Workflows; +using Xunit; +using Xunit.Abstractions; + +public class SystemNexusTests : WorkflowEnvironmentTestBase +{ + public SystemNexusTests(ITestOutputHelper output, WorkflowEnvironment env) + : base(output, env) + { + } + + [Fact] + public async Task ExecuteWorkflowAsync_SignalWithStartFromWorkflow_Succeeds() + { + var newOptions = (TemporalClientOptions)Client.Options.Clone(); + newOptions.DataConverter = DataConverter.Default with + { + PayloadCodec = new Base64PayloadCodec(), + }; + var codecClient = new TemporalClient(Client.Connection, newOptions); + var workerOptions = new TemporalWorkerOptions($"tq-{Guid.NewGuid()}"). + AddWorkflow(); + await ExecuteWorkerAsync( + async worker => + { + var targetWorkflowId = $"workflow-{Guid.NewGuid()}"; + var resultWorkflowId = await codecClient.ExecuteWorkflowAsync( + (SystemNexusSignalWithStartCallerWorkflow workflow) => + workflow.RunAsync(targetWorkflowId, worker.Options.TaskQueue!), + new(id: $"workflow-{Guid.NewGuid()}", taskQueue: worker.Options.TaskQueue!)); + Assert.Equal(targetWorkflowId, resultWorkflowId); + + var targetHandle = codecClient.GetWorkflowHandle< + SystemNexusSignalWithStartTargetWorkflow, + IReadOnlyCollection>(targetWorkflowId); + var events = await targetHandle.GetResultAsync(); + Assert.Equal(3, events.Count); + Assert.Contains("Started: start-value", events); + Assert.Contains("Signal: signal-one", events); + Assert.Contains("Signal: signal-two", events); + }, + workerOptions, + codecClient); + } + + [Workflow] + public class SystemNexusSignalWithStartTargetWorkflow + { + private readonly List events = new(); + + [WorkflowRun] + public async Task> RunAsync(string value) + { + events.Add($"Started: {value}"); + await Workflow.WaitConditionAsync(() => events.Count >= 3); + return events; + } + + [WorkflowSignal] + public Task SignalAsync(string value) + { + events.Add($"Signal: {value}"); + return Task.CompletedTask; + } + } + + [Workflow] + public class SystemNexusSignalWithStartCallerWorkflow + { + [WorkflowRun] + public async Task RunAsync(string workflowId, string taskQueue) + { + var handle = await Workflow.SignalWithStartWorkflowAsync( + (SystemNexusSignalWithStartTargetWorkflow workflow) => + workflow.RunAsync("start-value"), + workflow => workflow.SignalAsync("signal-one"), + new(workflowId, taskQueue) + { + IdConflictPolicy = WorkflowIdConflictPolicy.UseExisting, + }); + + await Workflow.SignalWithStartWorkflowAsync( + (SystemNexusSignalWithStartTargetWorkflow workflow) => + workflow.RunAsync("unused-start-value"), + workflow => workflow.SignalAsync("signal-two"), + new(workflowId, taskQueue) + { + IdConflictPolicy = WorkflowIdConflictPolicy.UseExisting, + }); + + return handle.Id; + } + } + + private static async Task ExecuteWorkerAsync( + Func action, + TemporalWorkerOptions options, + IWorkerClient client) + { + options = (TemporalWorkerOptions)options.Clone(); + options.AddWorkflow(); + options.Interceptors ??= new[] { new XunitExceptionInterceptor() }; + using var worker = new TemporalWorker(client, options); + await worker.ExecuteAsync(() => action(worker)); + } +} diff --git a/tests/Temporalio.Tests/Worker/WorkflowWorkerTests.cs b/tests/Temporalio.Tests/Worker/WorkflowWorkerTests.cs index 1a600e55..59ff31a4 100644 --- a/tests/Temporalio.Tests/Worker/WorkflowWorkerTests.cs +++ b/tests/Temporalio.Tests/Worker/WorkflowWorkerTests.cs @@ -991,55 +991,6 @@ async Task HandleSignalAsync(string arg) => public IList Events() => events; } - [Workflow] - public class SystemNexusSignalWithStartTargetWorkflow - { - private readonly List events = new(); - - [WorkflowRun] - public async Task> RunAsync(string value) - { - events.Add($"Started: {value}"); - await Workflow.WaitConditionAsync(() => events.Count >= 3); - return events; - } - - [WorkflowSignal] - public Task SignalAsync(string value) - { - events.Add($"Signal: {value}"); - return Task.CompletedTask; - } - } - - [Workflow] - public class SystemNexusSignalWithStartCallerWorkflow - { - [WorkflowRun] - public async Task RunAsync(string workflowId, string taskQueue) - { - var handle = await Workflow.SignalWithStartWorkflowAsync( - (SystemNexusSignalWithStartTargetWorkflow workflow) => - workflow.RunAsync("start-value"), - workflow => workflow.SignalAsync("signal-one"), - new(workflowId, taskQueue) - { - IdConflictPolicy = WorkflowIdConflictPolicy.UseExisting, - }); - - await Workflow.SignalWithStartWorkflowAsync( - (SystemNexusSignalWithStartTargetWorkflow workflow) => - workflow.RunAsync("unused-start-value"), - workflow => workflow.SignalAsync("signal-two"), - new(workflowId, taskQueue) - { - IdConflictPolicy = WorkflowIdConflictPolicy.UseExisting, - }); - - return handle.Id; - } - } - [Fact] public async Task ExecuteWorkflowAsync_Signals_ProperlyHandled() { @@ -1121,40 +1072,6 @@ await ExecuteWorkerAsync(async worker => }); } - [Fact] - public async Task ExecuteWorkflowAsync_SignalWithStartFromWorkflow_Succeeds() - { - var newOptions = (TemporalClientOptions)Client.Options.Clone(); - newOptions.DataConverter = DataConverter.Default with - { - PayloadCodec = new Base64PayloadCodec(), - }; - var codecClient = new TemporalClient(Client.Connection, newOptions); - var workerOptions = new TemporalWorkerOptions($"tq-{Guid.NewGuid()}"). - AddWorkflow(); - await ExecuteWorkerAsync( - async worker => - { - var targetWorkflowId = $"workflow-{Guid.NewGuid()}"; - var resultWorkflowId = await codecClient.ExecuteWorkflowAsync( - (SystemNexusSignalWithStartCallerWorkflow workflow) => - workflow.RunAsync(targetWorkflowId, worker.Options.TaskQueue!), - new(id: $"workflow-{Guid.NewGuid()}", taskQueue: worker.Options.TaskQueue!)); - Assert.Equal(targetWorkflowId, resultWorkflowId); - - var targetHandle = codecClient.GetWorkflowHandle< - SystemNexusSignalWithStartTargetWorkflow, - IReadOnlyCollection>(targetWorkflowId); - var events = await targetHandle.GetResultAsync(); - Assert.Equal(3, events.Count); - Assert.Contains("Started: start-value", events); - Assert.Contains("Signal: signal-one", events); - Assert.Contains("Signal: signal-two", events); - }, - workerOptions, - codecClient); - } - [Workflow] public class BadSignalArgsDroppedWorkflow { From 0cff561154b0e952a195836cf3e6fc85cd028896 Mon Sep 17 00:00:00 2001 From: Tim Conley Date: Mon, 17 Aug 2026 15:12:18 -0700 Subject: [PATCH 2/2] Add System Nexus transfer converter context --- CHANGELOG.md | 4 - .../Nexus/SystemNexusConverterContext.cs | 73 ----------------- .../Nexus/SystemNexusPayloadConverter.cs | 37 +++------ src/Temporalio/Worker/WorkflowInstance.cs | 23 +++--- .../Nexus/SystemNexusPayloadConverterTests.cs | 80 +++---------------- .../Worker/SystemNexusTests.cs | 14 +++- 6 files changed, 45 insertions(+), 186 deletions(-) delete mode 100644 src/Temporalio/Nexus/SystemNexusConverterContext.cs diff --git a/CHANGELOG.md b/CHANGELOG.md index 0cc4e84c..140262a5 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -21,10 +21,6 @@ to docs, or any other relevant information. ### Added -- Added `SystemNexusConverterContext`, which exposes the application's original payload and failure - converters while a System Nexus transfer type converter is executing. This lets generated System - Nexus transfer types serialize nested values without recursively applying their own transfer hooks. - ### Changed - A non-retryable `ApplicationFailureException` with error type `PayloadValidationError` thrown by a diff --git a/src/Temporalio/Nexus/SystemNexusConverterContext.cs b/src/Temporalio/Nexus/SystemNexusConverterContext.cs deleted file mode 100644 index 105c4a66..00000000 --- a/src/Temporalio/Nexus/SystemNexusConverterContext.cs +++ /dev/null @@ -1,73 +0,0 @@ -using System; -using System.Threading; -using Temporalio.Converters; - -namespace Temporalio.Nexus -{ - /// - /// Provides the user configured converters while System Nexus transfer conversion runs. - /// - /// - /// This context is only available while a System Nexus workflow operation converts its input - /// or result through a Temporal transfer type converter. - /// - public static class SystemNexusConverterContext - { - private static readonly AsyncLocal CurrentLocal = new(); - - /// - /// Gets the application's payload converter for the current System Nexus conversion. - /// - /// - /// Thrown when called outside a System Nexus transfer conversion. - /// - public static IPayloadConverter PayloadConverter => Current.PayloadConverter; - - /// - /// Gets the application's failure converter for the current System Nexus conversion. - /// - /// - /// Thrown when called outside a System Nexus transfer conversion. - /// - public static IFailureConverter FailureConverter => Current.FailureConverter; - - private static ConverterContext Current => CurrentLocal.Value ?? throw new InvalidOperationException( - "The System Nexus converter context is only available while a System Nexus transfer type converter is executing."); - - /// - /// Sets the converters for the duration of the returned scope. - /// - /// The application's payload converter. - /// The application's failure converter. - /// A scope that restores the preceding converter context. - internal static IDisposable Push( - IPayloadConverter payloadConverter, - IFailureConverter failureConverter) - { - var previous = CurrentLocal.Value; - CurrentLocal.Value = new(payloadConverter, failureConverter); - return new PopOnDispose(previous); - } - - private sealed record ConverterContext( - IPayloadConverter PayloadConverter, - IFailureConverter FailureConverter); - - private sealed class PopOnDispose : IDisposable - { - private readonly ConverterContext? previous; - private bool disposed; - - internal PopOnDispose(ConverterContext? previous) => this.previous = previous; - - public void Dispose() - { - if (!disposed) - { - CurrentLocal.Value = previous; - disposed = true; - } - } - } - } -} diff --git a/src/Temporalio/Nexus/SystemNexusPayloadConverter.cs b/src/Temporalio/Nexus/SystemNexusPayloadConverter.cs index 8bf2f4c6..bd7dab3a 100644 --- a/src/Temporalio/Nexus/SystemNexusPayloadConverter.cs +++ b/src/Temporalio/Nexus/SystemNexusPayloadConverter.cs @@ -1,4 +1,5 @@ using System; +using Temporalio.Api.Common.V1; using Temporalio.Converters; namespace Temporalio.Nexus @@ -7,44 +8,26 @@ namespace Temporalio.Nexus /// Payload converter for System Nexus outer protobuf envelopes. /// /// - /// This converter applies transfer type conversion to the outer System Nexus envelope while - /// making the application's converters available to generated transfer types. + /// This converter applies transfer type conversion to the outer System Nexus envelope. /// internal sealed class SystemNexusPayloadConverter : IPayloadConverter { - private readonly IPayloadConverter userPayloadConverter; - private readonly IFailureConverter userFailureConverter; - private readonly IPayloadConverter outerPayloadConverter; - - /// - /// Initializes a new instance of the class. - /// - /// The application's payload converter. - /// The application's failure converter. - internal SystemNexusPayloadConverter( - IPayloadConverter userPayloadConverter, - IFailureConverter userFailureConverter) - { - this.userPayloadConverter = userPayloadConverter; - this.userFailureConverter = userFailureConverter; - outerPayloadConverter = TemporalTransferTypePayloadConverter.Wrap( + private static readonly IPayloadConverter OuterPayloadConverter = + TemporalTransferTypePayloadConverter.Wrap( new DefaultPayloadConverter(new BinaryProtoConverter())); - } /// - public Temporalio.Api.Common.V1.Payload ToPayload(object? value) + public Payload ToPayload(object? value) { - using var context = SystemNexusConverterContext.Push( - userPayloadConverter, userFailureConverter); - return outerPayloadConverter.ToPayload(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); } /// - public object? ToValue(Temporalio.Api.Common.V1.Payload payload, Type type) + public object? ToValue(Payload payload, Type type) { - using var context = SystemNexusConverterContext.Push( - userPayloadConverter, userFailureConverter); - return outerPayloadConverter.ToValue(payload, type); + return OuterPayloadConverter.ToValue(payload, type); } } } diff --git a/src/Temporalio/Worker/WorkflowInstance.cs b/src/Temporalio/Worker/WorkflowInstance.cs index d8d547b8..b40d4ca5 100644 --- a/src/Temporalio/Worker/WorkflowInstance.cs +++ b/src/Temporalio/Worker/WorkflowInstance.cs @@ -2688,16 +2688,16 @@ public override Task> ScheduleNexusOperati new CanceledFailureException("Nexus operation cancelled before scheduled")); } - // TODO(cretz): Support Nexus serialization context - var payloadConverter = SystemNexusPayloadVisitor.IsSystemNexusEndpoint( + // TODO: Scope the generated System Nexus support converter context around this + // operation converter once the generated support file is ingested into the SDK. + var systemNexusPayloadConverter = SystemNexusPayloadVisitor.IsSystemNexusEndpoint( input.ClientOptions.Endpoint) ? - new SystemNexusPayloadConverter( - instance.payloadConverterNoContext, - instance.failureConverterNoContext) : - instance.payloadConverterNoContext; + new SystemNexusPayloadConverter() : null; + var operationPayloadConverter = + systemNexusPayloadConverter ?? instance.payloadConverterNoContext; var seq = ++instance.nexusOperationCounter; - var inputPayload = payloadConverter.ToPayload(input.Arg); + var inputPayload = operationPayloadConverter.ToPayload(input.Arg); var cmd = new ScheduleNexusOperation() { Seq = seq, @@ -2726,7 +2726,10 @@ public override Task> ScheduleNexusOperati var workflowCommand = new WorkflowCommand() { ScheduleNexusOperation = cmd }; if (input.Options.Summary is { } summary) { - workflowCommand.UserMetadata = new() { Summary = payloadConverter.ToPayload(summary) }; + workflowCommand.UserMetadata = new() + { + Summary = instance.payloadConverterNoContext.ToPayload(summary), + }; } instance.AddCommand(workflowCommand); @@ -2762,7 +2765,7 @@ public override Task> ScheduleNexusOperati // If there is a start sync fail, we have to fail the handle task and // there's nothing more we can do here var handle = new NexusWorkflowOperationHandleImpl( - payloadConverter, + operationPayloadConverter, // TODO(cretz): Support Nexus serialization context, ideally not // creating failure converter with context until actually needed instance.failureConverterNoContext, @@ -2772,7 +2775,7 @@ public override Task> ScheduleNexusOperati // TODO(cretz): Support Nexus serialization context handleSource.SetException( instance.failureConverterNoContext.ToException( - syncStartFail, payloadConverter)); + syncStartFail, operationPayloadConverter)); return; } diff --git a/tests/Temporalio.Tests/Nexus/SystemNexusPayloadConverterTests.cs b/tests/Temporalio.Tests/Nexus/SystemNexusPayloadConverterTests.cs index 11923432..91914472 100644 --- a/tests/Temporalio.Tests/Nexus/SystemNexusPayloadConverterTests.cs +++ b/tests/Temporalio.Tests/Nexus/SystemNexusPayloadConverterTests.cs @@ -1,94 +1,36 @@ namespace Temporalio.Tests.Nexus; -using Google.Protobuf; -using Temporalio.Api.Common.V1; using Temporalio.Converters; using Temporalio.Nexus; using Xunit; -using ApiWorkflowService = Temporalio.Api.WorkflowService.V1; +using ApiCommon = Temporalio.Api.Common.V1; public class SystemNexusPayloadConverterTests { [Fact] - public void TransferTypeWithEmbeddedPayload_RoundTrips() + public void TransferType_RoundTrips() { - var userPayloadConverter = TemporalTransferTypePayloadConverter.Wrap( - new StringPayloadConverter()); - var converter = new SystemNexusPayloadConverter( - userPayloadConverter, new DefaultFailureConverter()); - var value = new SystemNexusRequest("embedded-value"); + var converter = new SystemNexusPayloadConverter(); + var value = new SystemNexusRequest("value"); var payload = converter.ToPayload(value); Assert.Equal("binary/protobuf", payload.Metadata["encoding"].ToStringUtf8()); - var transferType = ApiWorkflowService.SignalWithStartWorkflowExecutionRequest.Parser.ParseFrom( - payload.Data); - Assert.Equal("test/string", transferType.Input.Payloads_[0].Metadata["encoding"].ToStringUtf8()); + Assert.Equal("value", ApiCommon.WorkflowType.Parser.ParseFrom(payload.Data).Name); Assert.Equal(value, converter.ToValue(payload, typeof(SystemNexusRequest))); - Assert.Throws(() => - _ = SystemNexusConverterContext.PayloadConverter); } [TemporalTransferTypeConverter(typeof(SystemNexusRequestConverter))] - public sealed record SystemNexusRequest(string EmbeddedValue); + public sealed record SystemNexusRequest(string Value); public sealed class SystemNexusRequestConverter : ITemporalTransferTypeConverter { - public Type TransferType => typeof(ApiWorkflowService.SignalWithStartWorkflowExecutionRequest); + public Type TransferType => typeof(ApiCommon.WorkflowType); - public object ToTransferType(object? value) - { - Assert.IsType( - SystemNexusConverterContext.PayloadConverter); - Assert.IsType(SystemNexusConverterContext.FailureConverter); - var request = (SystemNexusRequest)value!; - var input = new Payloads(); - input.Payloads_.Add(SystemNexusConverterContext.PayloadConverter.ToPayload( - new EmbeddedValue(request.EmbeddedValue))); - return new ApiWorkflowService.SignalWithStartWorkflowExecutionRequest { Input = input }; - } + public object ToTransferType(object? value) => + new ApiCommon.WorkflowType { Name = ((SystemNexusRequest)value!).Value }; - public object FromTransferType(object? transferType) - { - Assert.IsType( - SystemNexusConverterContext.PayloadConverter); - Assert.IsType(SystemNexusConverterContext.FailureConverter); - var request = (ApiWorkflowService.SignalWithStartWorkflowExecutionRequest)transferType!; - var embedded = SystemNexusConverterContext.PayloadConverter.ToValue( - request.Input.Payloads_[0]); - return new SystemNexusRequest(embedded.Value); - } - } - - [TemporalTransferTypeConverter(typeof(EmbeddedValueConverter))] - public sealed record EmbeddedValue(string Value); - - public sealed class EmbeddedValueConverter : ITemporalTransferTypeConverter - { - public Type TransferType => typeof(string); - - public object ToTransferType(object? value) => ((EmbeddedValue)value!).Value; - - public object FromTransferType(object? transferType) => new EmbeddedValue((string)transferType!); - } - - private sealed class StringPayloadConverter : IPayloadConverter - { - public Payload ToPayload(object? value) - { - var stringValue = Assert.IsType(value); - return new() - { - Metadata = { ["encoding"] = ByteString.CopyFromUtf8("test/string") }, - Data = ByteString.CopyFromUtf8(stringValue), - }; - } - - public object? ToValue(Payload payload, Type type) - { - Assert.Equal(typeof(string), type); - Assert.Equal("test/string", payload.Metadata["encoding"].ToStringUtf8()); - return payload.Data.ToStringUtf8(); - } + public object FromTransferType(object? transferType) => + new SystemNexusRequest(((ApiCommon.WorkflowType)transferType!).Name); } } diff --git a/tests/Temporalio.Tests/Worker/SystemNexusTests.cs b/tests/Temporalio.Tests/Worker/SystemNexusTests.cs index f8775b0c..4c26b644 100644 --- a/tests/Temporalio.Tests/Worker/SystemNexusTests.cs +++ b/tests/Temporalio.Tests/Worker/SystemNexusTests.cs @@ -5,7 +5,6 @@ namespace Temporalio.Tests.Worker; using Temporalio.Converters; using Temporalio.Tests.Converters; using Temporalio.Worker; -using Temporalio.Worker.Interceptors; using Temporalio.Workflows; using Xunit; using Xunit.Abstractions; @@ -18,7 +17,7 @@ public SystemNexusTests(ITestOutputHelper output, WorkflowEnvironment env) } [Fact] - public async Task ExecuteWorkflowAsync_SignalWithStartFromWorkflow_Succeeds() + public async Task ExecuteWorkflowAsync_SignalWithStartFromWorkflow_SucceedsAndReplays() { var newOptions = (TemporalClientOptions)Client.Options.Clone(); newOptions.DataConverter = DataConverter.Default with @@ -32,10 +31,11 @@ await ExecuteWorkerAsync( async worker => { var targetWorkflowId = $"workflow-{Guid.NewGuid()}"; - var resultWorkflowId = await codecClient.ExecuteWorkflowAsync( + var callerHandle = await codecClient.StartWorkflowAsync( (SystemNexusSignalWithStartCallerWorkflow workflow) => workflow.RunAsync(targetWorkflowId, worker.Options.TaskQueue!), new(id: $"workflow-{Guid.NewGuid()}", taskQueue: worker.Options.TaskQueue!)); + var resultWorkflowId = await callerHandle.GetResultAsync(); Assert.Equal(targetWorkflowId, resultWorkflowId); var targetHandle = codecClient.GetWorkflowHandle< @@ -46,6 +46,14 @@ await ExecuteWorkerAsync( Assert.Contains("Started: start-value", events); Assert.Contains("Signal: signal-one", events); Assert.Contains("Signal: signal-two", events); + + var replayer = new WorkflowReplayer( + new WorkflowReplayerOptions + { + DataConverter = newOptions.DataConverter, + }.AddWorkflow()); + var replay = await replayer.ReplayWorkflowAsync(await callerHandle.FetchHistoryAsync()); + Assert.Null(replay.ReplayFailure); }, workerOptions, codecClient);