Add System Nexus transfer payload converter

CHANGELOG.md

@@ -19,6 +19,8 @@ to docs, or any other relevant information.
 
 ## [Unreleased]
 
+### Added
+
 ### Changed
 
 - A non-retryable `ApplicationFailureException` with error type `PayloadValidationError` thrown by a

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)

src/Temporalio/Nexus/SystemNexusPayloadConverter.cs

@@ -0,0 +1,33 @@
+using System;
+using Temporalio.Api.Common.V1;
+using Temporalio.Converters;
+
+namespace Temporalio.Nexus
+{
+    /// <summary>
+    /// Payload converter for System Nexus outer protobuf envelopes.
+    /// </summary>
+    /// <remarks>
+    /// This converter applies transfer type conversion to the outer System Nexus envelope.
+    /// </remarks>
+    internal sealed class SystemNexusPayloadConverter : IPayloadConverter
+    {
+        private static readonly IPayloadConverter OuterPayloadConverter =
+            TemporalTransferTypePayloadConverter.Wrap(
+                new DefaultPayloadConverter(new BinaryProtoConverter()));
+
+        /// <inheritdoc />
+        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);
+        }
+
+        /// <inheritdoc />
+        public object? ToValue(Payload payload, Type type)
+        {
+            return OuterPayloadConverter.ToValue(payload, type);
+        }
+    }
+}

src/Temporalio/Worker/Generated/SystemNexusPayloadVisitor.cs

@@ -56,7 +56,7 @@ private static async Task<bool> 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,

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<Payload> 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<bool> TryVisitInputAsync(
             string? endpoint,
             Payload payload,

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;
@@ -2687,16 +2688,16 @@ public override Task<NexusWorkflowOperationHandle<TResult>> ScheduleNexusOperati
                         new CanceledFailureException("Nexus operation cancelled before scheduled"));
                 }
 
-                // TODO(cretz): Support Nexus serialization context
-                var payloadConverter = instance.payloadConverterNoContext;
+                // 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() : null;
+                var operationPayloadConverter =
+                    systemNexusPayloadConverter ?? 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 = operationPayloadConverter.ToPayload(input.Arg);
                 var cmd = new ScheduleNexusOperation()
                 {
                     Seq = seq,
@@ -2725,7 +2726,10 @@ public override Task<NexusWorkflowOperationHandle<TResult>> 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);
 
@@ -2761,7 +2765,7 @@ public override Task<NexusWorkflowOperationHandle<TResult>> 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<TResult>(
-                                payloadConverter,
+                                operationPayloadConverter,
                                 // TODO(cretz): Support Nexus serialization context, ideally not
                                 // creating failure converter with context until actually needed
                                 instance.failureConverterNoContext,
@@ -2771,7 +2775,7 @@ public override Task<NexusWorkflowOperationHandle<TResult>> ScheduleNexusOperati
                                 // TODO(cretz): Support Nexus serialization context
                                 handleSource.SetException(
                                     instance.failureConverterNoContext.ToException(
-                                        syncStartFail, payloadConverter));
+                                        syncStartFail, operationPayloadConverter));
                                 return;
                             }
 

tests/Temporalio.Tests/Nexus/SystemNexusPayloadConverterTests.cs

@@ -0,0 +1,36 @@
+namespace Temporalio.Tests.Nexus;
+
+using Temporalio.Converters;
+using Temporalio.Nexus;
+using Xunit;
+using ApiCommon = Temporalio.Api.Common.V1;
+
+public class SystemNexusPayloadConverterTests
+{
+    [Fact]
+    public void TransferType_RoundTrips()
+    {
+        var converter = new SystemNexusPayloadConverter();
+        var value = new SystemNexusRequest("value");
+
+        var payload = converter.ToPayload(value);
+
+        Assert.Equal("binary/protobuf", payload.Metadata["encoding"].ToStringUtf8());
+        Assert.Equal("value", ApiCommon.WorkflowType.Parser.ParseFrom(payload.Data).Name);
+        Assert.Equal(value, converter.ToValue(payload, typeof(SystemNexusRequest)));
+    }
+
+    [TemporalTransferTypeConverter(typeof(SystemNexusRequestConverter))]
+    public sealed record SystemNexusRequest(string Value);
+
+    public sealed class SystemNexusRequestConverter : ITemporalTransferTypeConverter
+    {
+        public Type TransferType => typeof(ApiCommon.WorkflowType);
+
+        public object ToTransferType(object? value) =>
+            new ApiCommon.WorkflowType { Name = ((SystemNexusRequest)value!).Value };
+
+        public object FromTransferType(object? transferType) =>
+            new SystemNexusRequest(((ApiCommon.WorkflowType)transferType!).Name);
+    }
+}

tests/Temporalio.Tests/Worker/NexusWorkerTests.cs

@@ -3115,4 +3115,4 @@ private async Task<WorkflowHandle<CustomFuncWorkflow, TResult>> RunInWorkflowAsy
             return handle;
         });
     }
-}
+}

tests/Temporalio.Tests/Worker/SystemNexusTests.cs

@@ -0,0 +1,122 @@
+namespace Temporalio.Tests.Worker;
+
+using Temporalio.Api.Enums.V1;
+using Temporalio.Client;
+using Temporalio.Converters;
+using Temporalio.Tests.Converters;
+using Temporalio.Worker;
+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_SucceedsAndReplays()
+    {
+        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<SystemNexusSignalWithStartTargetWorkflow>();
+        await ExecuteWorkerAsync<SystemNexusSignalWithStartCallerWorkflow>(
+            async worker =>
+            {
+                var targetWorkflowId = $"workflow-{Guid.NewGuid()}";
+                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<
+                    SystemNexusSignalWithStartTargetWorkflow,
+                    IReadOnlyCollection<string>>(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);
+
+                var replayer = new WorkflowReplayer(
+                    new WorkflowReplayerOptions
+                    {
+                        DataConverter = newOptions.DataConverter,
+                    }.AddWorkflow<SystemNexusSignalWithStartCallerWorkflow>());
+                var replay = await replayer.ReplayWorkflowAsync(await callerHandle.FetchHistoryAsync());
+                Assert.Null(replay.ReplayFailure);
+            },
+            workerOptions,
+            codecClient);
+    }
+
+    [Workflow]
+    public class SystemNexusSignalWithStartTargetWorkflow
+    {
+        private readonly List<string> events = new();
+
+        [WorkflowRun]
+        public async Task<IReadOnlyCollection<string>> 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<string> 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<TWorkflow>(
+        Func<TemporalWorker, Task> action,
+        TemporalWorkerOptions options,
+        IWorkerClient client)
+    {
+        options = (TemporalWorkerOptions)options.Clone();
+        options.AddWorkflow<TWorkflow>();
+        options.Interceptors ??= new[] { new XunitExceptionInterceptor() };
+        using var worker = new TemporalWorker(client, options);
+        await worker.ExecuteAsync(() => action(worker));
+    }
+}