Add optional retry policy for PayloadCodec failures

[whitemanthedj]

Is your feature request related to a problem? Please describe.

A PayloadCodec may call an external service during Encode/Decode — examples include the in-tree NewRemotePayloadCodec (HTTP to a sidecar), KMS-backed encryption codecs that call a cloud KMS for data-key operations, and codecs that fetch schemas from a remote registry. Any of these can fail transiently (network blip, throttling, 5xx). Today the SDK has no way to retry a codec call: the first error from any codec immediately aborts the encode/decode and surfaces to user code. Each downstream user has to either fork the codec or wrap retry logic inside their own codec implementation, which duplicates effort and doesn't help the stdlib RemotePayloadCodec, which has no retries today (see converter/codec.go — RemotePayloadCodec.encodeOrDecode does a single pc.options.Client.Do(req) and returns the error directly).

Describe the solution you'd like

A sibling constructor NewCodecDataConverterWithOptions(parent, codecs, CodecDataConverterOptions) mirroring the *WithOptions pattern used in PR #2225 (Serialization Context). The options struct exposes an optional retry policy applied per codec call:

type CodecDataConverterOptions struct {
    RetryPolicy *CodecRetryPolicy // nil => no retry (default)
    IsRetryable func(error) bool  // nil => retry all errors
    Context     context.Context   // nil => context.Background()
}

type CodecRetryPolicy struct {
    InitialInterval    time.Duration
    BackoffCoefficient float64
    MaximumInterval    time.Duration
    MaximumAttempts    int
    ExpirationInterval time.Duration
}

The existing NewCodecDataConverter(parent, codecs...) stays unchanged and delegates to the new one with a zero-value options struct, preserving backward compatibility (no retries by default).

Retry boundary: each codec's Encode/Decode is retried independently, so a chain of [compress, encrypt] re-encrypts on transient KMS failure without re-compressing (relevant for non-idempotent codecs like authenticated encryption).

Example usage

dc := converter.NewCodecDataConverterWithOptions(
    parent,
    []converter.PayloadCodec{kmsCodec},
    converter.CodecDataConverterOptions{
        RetryPolicy: &converter.CodecRetryPolicy{
            InitialInterval:    100 * time.Millisecond,
            BackoffCoefficient: 2.0,
            MaximumAttempts:    4,
        },
        IsRetryable: func(err error) bool {
            // codec-specific predicate, e.g. detect KMS throttling
            return isTransientKMSError(err)
        },
        Context: ctx,
    },
)

Describe alternatives you've considered

• Each codec author implements retry internally. Works, but duplicates effort and doesn't fix RemotePayloadCodec.
• Exposing internal/common/backoff publicly. Larger surface; out of proportion to the need.
• Adding context.Context to PayloadCodec.Encode/Decode. Interface-breaking; separate discussion. The options-struct Context only governs retry sleeps, not the codec call itself.

Additional context

Related prior issues: #2206 (closed by #2228) hit exactly this scenario — a remote codec returned DeadlineExceeded during encodeArgs and, combined with a session activity cancellation, permanently bricked a workflow. PR #2228 fixed the catastrophic side-effect (no more panic; the workflow task fails gracefully and the server retries), but the underlying transient codec failure still kills the in-flight task and forces a full task retry. This proposal is the missing root-cause complement: codecs that opt in recover from a transient blip without burning a workflow task at all.

I'm happy to implement this and send up a PR, but ideally I want to confirm the API shape with maintainers before doing so. Primary open questions remaining are:

1. The local CodecRetryPolicy struct as proposed, or should it thread something else (e.g. temporal.RetryPolicy)? A local struct avoids an import cycle from converter/ back into temporal/.
2. Any objection to the per-codec (vs. whole-chain) retry boundary?
3. Default IsRetryable semantics — retry-all (proposed) vs. retry-none (force explicit classification)?

[whitemanthedj]

Opened #2373 with a proposed implementation, out of draft and ready for review! Happy to iterate on the API as we see fit

[Sushisource]

Hi @whitemanthedj - we discussed this today and I think our take is that we're not sure this makes sense as a built-in in the SDK. At least, not for everything that isn't specifically RemotePayloadCodec. As you mentioned, authors can (and, IMO, should) decide what and when to retry things internally in their codec, since they'll have a better idea than we would with generic "retry any failure" logic. Plus, the workflow task tight timeline (< 10s) puts a bound on how much retrying makes sense.

I think we'd be OK with adjusting the PR to allow passing retry options specifically to NewRemotePayloadCodec. LMK what you think.

[whitemanthedj]

Hey @Sushisource thanks for responding and the framing! The point about codec authors being the right place for retry decisions held up well in practice, and that informed how we ended up handling this on our side without further SDK changes.

While working through it, we hit a related but distinct gap that may be worth a smaller follow-up discussion. Posting it here in case it is in scope for a separate change.

The gap

When a PayloadCodec returns an error from Encode or Decode during workflow input decode, activity result decode, signal decode, or any other workflow-goroutine codec invocation, the error path at internal/internal_worker.go:1084-1089 returns the error from workflowExecutor.Execute rather than panicking. A returned error is recorded as the workflow's final result and converted to a permanent workflow failure via the default failure converter.

In contrast, a panic from inside the workflow goroutine is captured by coroutineState.run at internal/internal_workflow.go:1217-1228, surfaces as a workflowPanicError, and under default BlockWorkflow becomes a retryable WorkflowTaskFailed (per the comment at internal/internal_task_handlers.go:1323-1327). The server then retries the workflow task.

So there are two paths into the codec error surface today:

Codec signal	SDK outcome
return payloads, err	Workflow fails permanently
panic(err)	Workflow task fails, server retries

For codec authors implementing external-service-calling codecs (HTTP sidecars, KMS clients, schema registries, anything that can fail transiently), the practical recourse today is to wrap the codec body in defer recover and panic(err) on transient classifications. That works because of the coroutine recover, but it is an undocumented technique that depends on internal SDK behavior, and panicking is an unusual idiom for "this call should be retried."

The ask

Would the team be open to a small typed-error addition in converter/ that codec authors could return to mean "this failure should be classified as a retryable workflow task failure rather than a workflow failure"? Rough shape:

// In converter/errors.go (new file) or converter/codec.go.

// RetryableTaskError, when returned from a PayloadCodec's Encode or Decode
// during workflow-goroutine execution, causes the SDK to convert the failure
// into a retryable WorkflowTaskFailed rather than a permanent workflow
// failure. The underlying error is preserved via Unwrap so failure converters
// and observability tooling see the original cause. When returned from
// activity-goroutine context or client-side use, the error behaves like any
// other returned error (no special routing).
type RetryableTaskError struct {
    Err error
}

func (e *RetryableTaskError) Error() string { return e.Err.Error() }
func (e *RetryableTaskError) Unwrap() error { return e.Err }

// NewRetryableTaskError wraps err in a RetryableTaskError. Returning the
// result of this function from PayloadCodec.Encode or Decode signals to the
// SDK that the failure is transient at the task-attempt level.
func NewRetryableTaskError(err error) error {
    return &RetryableTaskError{Err: err}
}

Wired in internal/internal_worker.go at the codec error site:

args, err = decodeArgsToRawValues(dataConverter, fnType, input)
if err != nil {
    var taskErr *converter.RetryableTaskError
    if errors.As(err, &taskErr) {
        panic(err)
    }
    return nil, fmt.Errorf(
        "unable to decode the workflow function input payload with error: %w, function name: %v",
        err, we.workflowType)
}

A symmetric check would go on the encode side and on any other workflow-goroutine codec invocation. Activity-goroutine codec calls would not change behavior (returning the wrapper there still surfaces as an activity failure, which is appropriate since the activity retry policy is the right escape hatch in that context).

The codec author then still owns the classification decision per error. The SDK does not impose a default retry policy or generic "retry any failure" behavior. A codec implementation might classify 503 and EOF as retryable but 400 and ErrConfigurationMissing as non-retryable, returning the wrapper only for the former. The SDK's role is just to honor the signal, not to make the decision.

The 10s task-budget concern that came up before does not bind here either, because the failure is being escalated out of the inline call to the server's workflow-task-retry mechanism. The retry happens at the task layer (potentially across many task attempts), not inside one task.

What this would replace

Today the workaround for codec authors who want this behavior is panic(err) from inside the codec, relying on coroutineState.run's recover. That works but has two real downsides:

1. It is an internal-SDK mechanism that is not part of the public contract. Codec authors are reaching past the public API to get a behavior that should arguably be a first-class part of it.
2. Panic semantics are surprising. The codec implementation has to scope panics carefully (workflow goroutine yes, activity goroutine no, client side no) to avoid noisy Activity panic. logs, polluted ActivityTaskErrorCounter metrics, or crashing client processes.

A typed-error approach replaces both with an explicit, documented API.

Open questions for the team

1. Is this small enough in scope to take as a separate contribution, separate from the closed/narrowed retry-on-RemotePayloadCodec work?
2. Preference on placement: new converter/errors.go file, or addition to existing converter/codec.go?
3. Should RetryableTaskError route through panic internally (mirroring the existing coroutine recover path) or through a new explicit task-failure return path? The former is a smaller diff; the latter is a cleaner conceptual separation.
4. Should symmetric treatment apply to FailureConverter errors, or is that out of scope?
5. Are there other anticipated changes for something similar already on the roadmap?

Happy to draft a similar PR if this sounds more reasonable

[whitemanthedj]

Quick bump on this one. We're still hitting a related gap on our side and would love to revive the conversation, though we may want to pivot the framing slightly.

For context, we run a custom PayloadCodec that calls an external service for per-customer encryption and decryption of payloads. Transient failures from that service are a real concern - 5xx's, throttles, deadline exceeded, occasional connection resets. Today we retry inside the codec and apply an exponential backoff, but once we exhaust those in-process retries the error simply propagates out of the codec and the workflow fails in a non-retryable fashion. We have no way to tell the SDK "this final failure is transient, please leverage Temporal's native workflow task retries here."

One alternative we evaluated was panicking on exhausted retries from inside the workflow goroutine so the SDK's coroutine recover would translate the failure into a WorkflowTaskFailed under default BlockWorkflow, effectively piggybacking on the native workflow task retry behavior. We don't want to take that path since it's a hack. It would only work in workflow context (activity / client-side codec usage gets nothing), and it would pollute our workflow task failure metrics and alerting in ways that make it operationally painful.

Given that, the change we'd most love to see is less about a built-in RetryPolicy inside the codec wrapper (which is what #2373 proposes) and more about a way for the codec to surface "this error is retryable vs not" back to the SDK, so that codec failures can flow into Temporal's native workflow task retry path rather than terminating the workflow. Concretely something like a typed retryable codec error, or an interface the codec error can implement, that the SDK respects when translating codec failures into workflow task outcomes. We'd keep our own retry loop inside the codec, we just want a clean way to tell the SDK "this final failure is transient, don't terminate the workflow."

Open to a different shape if you think there's a better one. The in-codec RetryPolicy in #2373 also closes a chunk of the gap for us, especially if it ends up integrated with the workflow task retry path rather than being a self-contained sleep loop. Mostly want to see this move forward in some form so we can close the gap without resorting to the panic workaround.

Any thoughts on direction, or is the team open to either approach?

[yuandrew]

Can you provide a minimal example or identify the exact codec call site where a returned codec error permanently fails the workflow?

Looking at the workflow input decode path, a returned decode error from workflowExecutor.Execute appears to route through the default BlockWorkflow policy as a workflow task failure, not a workflow execution failure. But codec errors can surface differently depending on where decode happens, e.g. workflow start arguments vs activity result Future.Get vs signals/updates/queries.

If your case is activity result decode or another workflow-goroutine API where the codec error is returned into workflow code and then returned by the workflow, that would explain the permanent workflow failure behavior. A concrete reproducer would help pin down whether this needs a general typed retryable codec error, or a narrower fix for specific decode sites.

[whitemanthedj]

Thanks for digging into this! After going through it a bit more, I don't think the decode path actually retries the way you described though. From what I can tell a returned decode error lands on permanent workflow completion, not BlockWorkflow. Let me walk through what I'm seeing and have you double check it, and point me at anything I'm misreading, because it really comes down to one distinction: I think BlockWorkflow only gets reached via a panic, and the decode path returns instead of panicking.

The encode side does line up with that, so I'm with you there: when the workflow goroutine encodes activity args to schedule an activity, a codec error becomes a panic (internal/workflow.go:1085-1088). That gets recovered in coroutineState.run, turns into a workflowPanicError, and under the default BlockWorkflow policy becomes a retryable WorkflowTaskFailed. That path retries forever, agreed. It's also the path #2228 hardened for session workflows, so I suspect it's the one behind the "returns route through BlockWorkflow" intuition.

It seems to diverge on the decode side where the codec error is returned, not panicked, and I can't find where it reaches BlockWorkflow. Tracing workflow input decode:

• workflowExecutor.Execute returns the decode error rather than panicking (internal/internal_worker.go:1084-1089). It becomes the workflow result error and flows to env.Complete(nil, err), which stores it as a plain error in w.err (internal/internal_task_handlers.go:642).
• applyWorkflowPanicPolicy only pulls w.err into the BlockWorkflow branch when it's a *workflowPanicError (internal/internal_task_handlers.go:1293-1300). A plain returned error skips that branch, so the workflow completes with a permanent failure rather than failing the task.

Activity result decode looks like the same shape: future.Get returns the codec error to workflow code (internal/internal_workflow.go:377-378, and 1749-1752 for the decode future). Normal workflow code propagates that (if err != nil { return err }), which returns from the workflow function and again lands on the permanent-completion path, not task retry.

The thing that stands out is the split is entirely on the SDK side of the call, not something a codec author can make a distinction on. Encode and Decode are symmetric from the codec's point of view, you implement both the same way. But the SDK panics at the encode-when-scheduling site and returns at the decode sites, and that choice alone decides whether a transient failure retries or goes terminal.

So my read is that a transient codec failure during decode (workflow input, or decoding an activity result back into the workflow) permanently fails the workflow, while the same transient failure during encode when scheduling an activity is retried via BlockWorkflow.

For a returned decode error to retry, it would have to become a *workflowPanicError in w.err, and the only producers of that I can find are the recover in coroutineState.run (internal/internal_workflow.go:1221-1224) and the deadlock detector, neither of which touches a returned error. If there's a reclassification step I'm missing, that's the spot I'd want to be pointed at, since it would change the conclusion.

Minimal shape for the decode case:

// PayloadCodec that calls an external service on Decode (KMS, sidecar,
// schema registry, etc) and hits a transient blip.
func (c *codec) Decode(ps []*commonpb.Payload) ([]*commonpb.Payload, error) {
    if transientBlip() {
        return ps, fmt.Errorf("service unavailable: 503") // returned, not panicked
    }
    // ... normal decode ...
}

With that codec, a transient blip while decoding the workflow's start args, or while decoding an activity result inside future.Get, surfaces as a returned error that the SDK routes to workflow completion, and the workflow fails permanently on what is really a recoverable network hiccup.

Couple of questions:

1. Does that follow what you're seeing on your end? Specifically, is there a path where a returned (non-panicked) decode error becomes a *workflowPanicError and hits the BlockWorkflow branch? That's the load-bearing assumption in the earlier comment, and I can't reproduce it in the source. If it doesn't hold, then returned decode errors are permanent, and that's the gap the issue is about.
2. If decode does fall through to permanent completion, that's the case the issue was really about. The only ways I see for a codec author to get task-retry behavior on decode today are to panic from inside the codec (undocumented, and awkward to scope correctly across workflow vs activity vs client contexts) or to retry inside Decode before returning. Is in-codec retry the intended answer here, or is there appetite for a documented signal (e.g. a typed error the codec returns that the SDK treats as a retryable task failure on the workflow-goroutine decode path)?

Separately, there's the activity-goroutine path (internal/internal_worker.go:1130 arg decode, :1159 result encode), where a codec error becomes an activity failure subject to the activity's RetryPolicy. Activities configured with MaximumAttempts: 1 turn a transient codec blip into a terminal activity failure there. Different domain from the above, just flagging it's the third distinct path.

And the earlier offer to take retry options on NewRemotePayloadCodec still stands as far as I'm concerned - happy to put that up as its own small PR independent of whatever we land on here.