Skip to content

fix: mark system Nexus envelope payloads #799

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Draft
tconley1428 wants to merge 2 commits into from
Draft

fix: mark system Nexus envelope payloads #799

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
3006723
fix: mark system Nexus envelope payloads
tconley1428 Jul 28, 2026
81ed67a
fix: handle system payload codec batches
tconley1428 Aug 10, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
12 changes: 3 additions & 9 deletions src/Temporalio.SystemNexus.Generator/Program.cs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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<string, Func<Payload, PayloadVisitor, PayloadsVisitor, EnvelopeVisitor?, Task>> EnvelopeVisitors =");
builder.AppendLine(" new Dictionary<string, Func<Payload, PayloadVisitor, PayloadsVisitor, EnvelopeVisitor?, Task>>");
builder.AppendLine(" {");
Expand All @@ -178,14 +176,13 @@ static void GeneratePayloadVisitor(

builder.AppendLine(" };");
builder.AppendLine();
builder.AppendLine(" private static async Task<bool> TryVisitAsync(");
builder.AppendLine(" string? endpoint,");
builder.AppendLine(" internal static async Task<bool> 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(" }");
Expand All @@ -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);
Expand Down
5 changes: 4 additions & 1 deletion src/Temporalio/Nexus/SystemNexusPayloadConverter.cs
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,6 +1,7 @@
using System;
using Temporalio.Api.Common.V1;
using Temporalio.Converters;
using Temporalio.Worker;

namespace Temporalio.Nexus
{
Expand All @@ -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;
}

/// <inheritdoc />
Expand Down
11 changes: 3 additions & 8 deletions src/Temporalio/Worker/Generated/SystemNexusPayloadVisitor.cs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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<string, Func<Payload, PayloadVisitor, PayloadsVisitor, EnvelopeVisitor?, Task>> EnvelopeVisitors =
new Dictionary<string, Func<Payload, PayloadVisitor, PayloadsVisitor, EnvelopeVisitor?, Task>>
{
Expand All @@ -34,14 +32,13 @@ internal static partial class SystemNexusPayloadVisitor
visitEnvelope),
};

private static async Task<bool> TryVisitAsync(
string? endpoint,
internal static async Task<bool> TryVisitAsync(
Payload payload,
PayloadVisitor visitPayload,
PayloadsVisitor visitPayloads,
EnvelopeVisitor? visitEnvelope)
EnvelopeVisitor? visitEnvelope = null)
{
if (!IsSystemNexusEndpoint(endpoint))
if (!IsSystemPayload(payload))
{
return false;
}
Expand All @@ -56,8 +53,6 @@ private static async Task<bool> 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,
Expand Down
13 changes: 0 additions & 13 deletions src/Temporalio/Worker/IWorkflowCodecHelperInstance.cs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -40,18 +40,5 @@ internal interface IWorkflowCodecHelperInstance
/// <param name="seq">Sequence.</param>
/// <returns>Context.</returns>
ISerializationContext.Workflow? GetPendingExternalSignalSerializationContext(uint seq);

/// <summary>
/// Gets the pending Nexus operation info for the given sequence.
/// </summary>
/// <param name="seq">Sequence.</param>
/// <returns>Info.</returns>
NexusOperationInfo? GetPendingNexusOperationInfo(uint seq);

/// <summary>
/// Pending Nexus operation info.
/// </summary>
/// <param name="Endpoint">Endpoint name.</param>
internal record NexusOperationInfo(string? Endpoint);
}
}
28 changes: 14 additions & 14 deletions src/Temporalio/Worker/SystemNexusPayloadVisitor.cs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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<Payload> payloads);

internal delegate Task EnvelopeVisitor(Payload payload);

internal static Task<bool> 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<bool> 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<T>(
Payload payload,
Expand All @@ -48,11 +43,16 @@ private static async Task VisitEnvelopeAsync<T>(
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);
}
}
97 changes: 67 additions & 30 deletions src/Temporalio/Worker/WorkflowCodecHelper.cs
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,3 +1,4 @@
using System.Collections.Generic;
using System.Linq;
using System.Threading.Tasks;
using Google.Protobuf;
Expand Down Expand Up @@ -125,20 +126,9 @@ await childCodec2.DecodeFailureAsync(
}
if (job.ResolveNexusOperation.Result.Completed != null)
{
var operationInfo = context.Instance?.GetPendingNexusOperationInfo(
Comment thread
tconley1428 marked this conversation as resolved.
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).

@jmaeagle99 jmaeagle99 Aug 7, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

This payload possibly does not have the system payload marker on it even if it is a system payload. Which means that it'll go through the normal user codec decode path and then fail deserialization causing the workflow to continuously fail the task and timeout the ExecuteWorkflowAsync_SignalWithStartFromWorkflow_Succeeds test. Might need to check for the marker or see if it's from the system nexus endpoint; or maybe somehow stamp the payload with the marker if it comes from the system nexus endpoint before we attempt the decode here.

@tconley1428 tconley1428 Aug 10, 2026

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

In what scenario would it be a system payload without the system payload marker? That should be tautologically equivalent as far as I know. (At least once server is updated sufficiently) Are you talking backwards compat or something else?

@jmaeagle99 jmaeagle99 Aug 10, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Back compat. Today, system payloads are identified by the endpoint, but this change completely drops that (I think). So in-flight workflows that do not have the new marker now going through the path of decoding the envelop. But if you don't care about those, then I think this is fine.

But I see that ExecuteWorkflowAsync_SignalWithStartFromWorkflow_Succeeds hung in every job. It's likely retrying a workflow task indefinitely for some reason. Probably due to the missing marker.

@tconley1428 tconley1428 Aug 10, 2026

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Yeah, this was in draft for a reason. All you say is true, but it's also not enabled at all yet, so we're good to do the breaking change. Nice catch though.

ConfigureAwait(false);
}
else if (job.ResolveNexusOperation.Result.Failed != null)
{
Expand Down Expand Up @@ -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:
Expand Down Expand Up @@ -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<Payload>();
var codecPayloads = new List<Payload>();
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,
Expand Down Expand Up @@ -538,15 +547,43 @@ private static async Task DecodeAsync(IPayloadCodec codec, RepeatedField<Payload
{
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.DecodeAsync(payloads).ConfigureAwait(false)).ToList();
var newPayloads = new List<Payload>();
var codecPayloads = new List<Payload>();
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);
Expand Down
9 changes: 0 additions & 9 deletions src/Temporalio/Worker/WorkflowInstance.cs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -898,13 +898,6 @@ public WorkflowActivationCompletion Activate(WorkflowActivation act)
return pending?.SerializationContext;
}

/// <inheritdoc/>
public IWorkflowCodecHelperInstance.NexusOperationInfo? GetPendingNexusOperationInfo(uint seq)
{
nexusOperationsPending.TryGetValue(seq, out var pending);
return pending == null ? null : new(pending.Endpoint);
}

/// <inheritdoc/>
protected override IEnumerable<Task>? GetScheduledTasks() => scheduledTasks;

Expand Down Expand Up @@ -2735,7 +2728,6 @@ public override Task<NexusWorkflowOperationHandle<TResult>> ScheduleNexusOperati

var handleSource = new TaskCompletionSource<NexusWorkflowOperationHandle<TResult>>();
var pending = new PendingNexusOperationInfo(
Endpoint: input.ClientOptions.Endpoint,
StartCompletionSource: new(),
ResultCompletionSource: new());
instance.nexusOperationsPending[seq] = pending;
Expand Down Expand Up @@ -3210,7 +3202,6 @@ private record PendingExternalCancel(
TaskCompletionSource<ResolveRequestCancelExternalWorkflow> CompletionSource);

private record PendingNexusOperationInfo(
string? Endpoint,
TaskCompletionSource<ResolveNexusOperationStart> StartCompletionSource,
TaskCompletionSource<NexusOperationResult> ResultCompletionSource);

Expand Down
Loading
Loading