From 41fec41a957b941e613804aae7e317b6e8dea107 Mon Sep 17 00:00:00 2001 From: adarsh0728 Date: Fri, 22 May 2026 18:57:07 +0530 Subject: [PATCH 1/3] doc for numa resilience to udf failures Signed-off-by: adarsh0728 --- docs/specifications/numa-resilience-doc.md | 157 +++++++++++++++++++++ 1 file changed, 157 insertions(+) create mode 100644 docs/specifications/numa-resilience-doc.md diff --git a/docs/specifications/numa-resilience-doc.md b/docs/specifications/numa-resilience-doc.md new file mode 100644 index 0000000000..115272198c --- /dev/null +++ b/docs/specifications/numa-resilience-doc.md @@ -0,0 +1,157 @@ +# Numa Container Resilience to UDF Failures + +| | | +| --- | --- | +| Status | Proposed | +| Tracking issues | [#3367](https://github.com/numaproj/numaflow/issues/3367) (umbrella), [#3368](https://github.com/numaproj/numaflow/issues/3368) (this proposal) | +| Related future work | [#3369](https://github.com/numaproj/numaflow/issues/3369), [#3370](https://github.com/numaproj/numaflow/issues/3370), [#3371](https://github.com/numaproj/numaflow/issues/3371) | +| Reduce vertices | Excluded from automatic reconnect in Phase 1 pending durable-state design. | +| Target Release | 1.9 + +## Context + +Today each vertex and monovertex pod has a `monitor` sidecar, a `numa` container, and UDF sidecars. The monitor sidecar serves runtime errors from the shared emptyDir because the `numa` process exits when it receives a gRPC error from a UDF. While `numa` is restarting, it cannot host the error endpoint. + +The umbrella goal in #3367 is to remove the monitor sidecar. Phase 1, tracked by #3368, makes that possible by keeping `numa` alive across non-reduce UDF failures. + +## Goals + +- For source, transformer, map, and sink UDFs, `numa` must not exit on UDF-side gRPC failures. +- `numa` must continue serving metrics and health endpoints during UDF outage and reconnect cycles. +- UDF errors remain observable through logs, `critical_error_total`, and persisted runtime-error files. +- Processing resumes automatically once the UDF sidecar is healthy again. +- Reduce vertices keep today's persist-and-exit behavior until durable reduce state is designed and approved. + +## Non-Goals + +- Moving `/runtime/errors` into `numa`; that is Phase 2, #3369. +- Changing daemon-server polling or UI-facing APIs; that is Phase 3, #3370. +- Adding e2e failure coverage; that is Phase 4, #3371. +- Changing source SDK at-least-once contracts. +- Changing Go controllers, CRDs, UI, or monitor-sidecar code in this phase. + +## Proposed Design + +All non-reduce UDF gRPC failures are handled uniformly. There is no transient vs non-transient classifier. On UDF failure, `numa`: + +1. Logs the error. +2. Increments `critical_error_total` with a UDF-specific reason. +3. Persists the runtime error. +4. Reconnects to the UDF sidecar. +5. Redrives the in-flight request without exiting. + +### Reconnect Helper + +Reconnect reuses the startup flow: + +- Re-read SDK server-info. +- Recreate a fresh `tonic::transport::Channel`. +- Rebuild the typed client stub. +- Run the typed readiness check. +- Reopen the BiDi stream and handshake. + +Retries use the existing infinite one-second retry pattern and are bounded only by the normal shutdown cancellation token. + +### Map Clients + +Map unary, batch, and stream use task-level retry. The receiver task does not cache requests; the per-message or per-batch task owns the original `MapRequest`. + +Map clients need shared mutable stream state because old `read_tx` clones are permanently dead after the BiDi receiver drops. Reconnect must: + +1. Create the new `(read_tx, read_rx)`. +2. Open the new BiDi stream and complete handshake. +3. Install the new sender in shared state. +4. Reset `SenderMapState.closed = false` and clear residual sender-map entries. +5. Fire `reconnect_ready`. + +This ordering prevents retried calls from observing either a stale sender or a still-closed sender map. + +Streaming map can emit partial output before a stream failure. Already-emitted outputs are not rolled back. On retry, the UDF may re-emit them, producing duplicates. `parent_info.current_index` continues monotonically so post-reconnect outputs get fresh `MessageID.index` values. + +### Source Transformer + +Transformer retry happens per message, not per batch. Each future inside `transform_batch` retries its own `Transformer::transform(...)` call on `Error::UdfRedrive`. + +Only after `transform()` returns `Ok` does the future update the tracker, create output handles, and call `mark_success!`. Sibling futures that already completed are not rerun. + +### Sink + +`UserDefinedSink::sink()` reconnects in place when the response stream fails, reopens the stream, and re-sends the current `Vec` plus EOT. `mark_success_batch!` remains gated on `sink()` returning `Ok`. + +External sinks may see duplicates if the old UDF process wrote data before crashing. That is consistent with at-least-once contract. + +### Source Read, Ack, and Nack + +Source `read()` reconnects in place and returns `Ok(vec![])` for the broken tick. The source SDK remains responsible for redelivering unacked offsets. + +Source `ack()` and `nack()` also reconnect in place. Transport failures from `ack_tx.send(...)`, `ack_resp_stream.message()`, or stream-closed `tonic::Status` are redrivable, not `NonRetryable`. Truly terminal internal actor-shutdown cases remain terminal. + +The user-defined source SDK contract remains unchanged: offsets must not be committed until an explicit ACK is received and persisted. In-flight NAKs must either be persisted or leave the offset uncommitted so it is redelivered. + +### Reduce Exception + +Reduce vertices retain current behavior: persist the error and exit. Reduce UDFs hold window state in-process, so reconnecting after a crash could drop or double-count aggregates. This carve-out requires issue-owner signoff. + +## Observability + +Reuse `critical_error_total` with new reason strings: + +- `source_runtime_error` +- `source_transformer_runtime_error` +- `udf_runtime_error` +- `sink_runtime_error` +- `fallback_sink_runtime_error` + +Runtime errors continue to be written under `/var/numaflow/runtime/application-errors//`. + +Two persistence changes are required: + +- Remove the `OnceLock` gate so more than one error per process can be recorded. +- Replace second-resolution filenames with collision-safe names: `--numa.json`, with matching unique temp filenames. + +The existing ten-file rotation cap remains. + +## At-Least-Once Safety + +At-least-once depends on `MessageHandle` and `AckHandle` invariants: + +- `mark_success!` only decrements the ref count. ACK happens later when the final handle drops and the ref count is zero. +- `mark_failed!` records a reason but does not decrement the ref count. Drop then NAKs. +- Dropping a handle without success NAKs. + +Redrive must therefore never call `mark_success!` or `mark_failed!` on `Error::UdfRedrive`, and must never leak a `MessageHandle`. + +Safety by path: + +| Path | Loss risk | Duplicate risk | +| --- | --- | --- | +| Source read reconnect | None, SDK redelivers unacked offsets | Possible | +| Source ack reconnect | None, ACK is idempotent | Possible extra ACK frame | +| Source nack reconnect | Conditional on SDK not committing implicit ACKs | Possible extra NAK frame | +| Map and transformer redrive | None, task holds handle until success or drop | Possible UDF reinvocation | +| Sink redrive | None, success is gated on sink response | Possible external writes | + +The only conditional case is source NAK reconnect, which depends on the existing user-defined-source SDK contract. Built-in Rust source connectors do not use the UDF-source gRPC path and are out of scope for this reconnect work. + +## Risks + +- Poison messages can loop forever. This is accepted for Phase 1. +- Reconnect storms are possible during sustained UDF outages. The retry interval stays at the current one-second startup retry cadence. +- Reduce still restarts. This is an explicit carve-out. +- SDK NAK behavior must be audited in numaflow-go, numaflow-java, numaflow-python, and numaflow-rs. + +## Testing + +Required tests: + +- Per-client reconnect tests for source read, source ack, source nack, transformer, map unary, map batch, map stream, and sink. +- Tests that `critical_error_total` increments and runtime errors persist for each UDF failure path. +- Tests that upstream cancellation tokens are not cancelled for redrivable UDF failures. +- Map stale-sender and reopened-state regression tests. +- Transformer partial-completion retry test. +- Map-stream partial-output retry test. +- `ack_tx.send` reconnect test. +- Runtime-error filename collision test. +- MessageHandle drop-NAK invariant test during outage. +- Top-level assertion that non-reduce UDF errors do not make `run()` return `Err`. +- User-defined-source SDK NAK-preservation test or audit. From ec0c659ca67244269b9b3488392b80a144c8bc01 Mon Sep 17 00:00:00 2001 From: adarsh0728 Date: Fri, 22 May 2026 19:07:24 +0530 Subject: [PATCH 2/3] small fixes Signed-off-by: adarsh0728 --- docs/specifications/numa-resilience-doc.md | 8 ++++---- 1 file changed, 4 insertions(+), 4 deletions(-) diff --git a/docs/specifications/numa-resilience-doc.md b/docs/specifications/numa-resilience-doc.md index 115272198c..e0cc84dd54 100644 --- a/docs/specifications/numa-resilience-doc.md +++ b/docs/specifications/numa-resilience-doc.md @@ -6,7 +6,7 @@ | Tracking issues | [#3367](https://github.com/numaproj/numaflow/issues/3367) (umbrella), [#3368](https://github.com/numaproj/numaflow/issues/3368) (this proposal) | | Related future work | [#3369](https://github.com/numaproj/numaflow/issues/3369), [#3370](https://github.com/numaproj/numaflow/issues/3370), [#3371](https://github.com/numaproj/numaflow/issues/3371) | | Reduce vertices | Excluded from automatic reconnect in Phase 1 pending durable-state design. | -| Target Release | 1.9 +| Target Release | 1.9 | ## Context @@ -50,7 +50,7 @@ Reconnect reuses the startup flow: - Run the typed readiness check. - Reopen the BiDi stream and handshake. -Retries use the existing infinite one-second retry pattern and are bounded only by the normal shutdown cancellation token. +Retries use the existing infinite one-second retry pattern and are bounded only by the normal shutdown cancellation token. A failure of `sdk_server_info` during reconnect is itself retried at the same one-second cadence; the reconnect helper does not surface it as a hard error. ### Map Clients @@ -107,7 +107,7 @@ Runtime errors continue to be written under `/var/numaflow/runtime/application-e Two persistence changes are required: - Remove the `OnceLock` gate so more than one error per process can be recorded. -- Replace second-resolution filenames with collision-safe names: `--numa.json`, with matching unique temp filenames. +- Replace second-resolution filenames with collision-safe names: `--numa.json`, and switch the per-call temp filename from the shared `current-numa.json` to a matching unique `--numa.tmp` so concurrent writers do not race. The existing ten-file rotation cap remains. @@ -138,7 +138,7 @@ The only conditional case is source NAK reconnect, which depends on the existing - Poison messages can loop forever. This is accepted for Phase 1. - Reconnect storms are possible during sustained UDF outages. The retry interval stays at the current one-second startup retry cadence. - Reduce still restarts. This is an explicit carve-out. -- SDK NAK behavior must be audited in numaflow-go, numaflow-java, numaflow-python, and numaflow-rs. +- SDK NAK behavior must be audited in numaflow-go, numaflow-java, numaflow-python, and numaflow-rs **as a pre-merge gate**. Any non-compliant SDK is either remediated first or carved out from the resilience guarantee. ## Testing From f1cc0b875a561d060d0f9d4bcc719de5a952a926 Mon Sep 17 00:00:00 2001 From: adarsh0728 Date: Tue, 9 Jun 2026 09:33:05 +0530 Subject: [PATCH 3/3] built in sources Signed-off-by: adarsh0728 --- docs/specifications/numa-resilience-doc.md | 34 +++++++++++++++++++--- 1 file changed, 30 insertions(+), 4 deletions(-) diff --git a/docs/specifications/numa-resilience-doc.md b/docs/specifications/numa-resilience-doc.md index e0cc84dd54..183c3e48bd 100644 --- a/docs/specifications/numa-resilience-doc.md +++ b/docs/specifications/numa-resilience-doc.md @@ -1,18 +1,19 @@ -# Numa Container Resilience to UDF Failures +# Numa Container Resilience | | | | --- | --- | | Status | Proposed | | Tracking issues | [#3367](https://github.com/numaproj/numaflow/issues/3367) (umbrella), [#3368](https://github.com/numaproj/numaflow/issues/3368) (this proposal) | | Related future work | [#3369](https://github.com/numaproj/numaflow/issues/3369), [#3370](https://github.com/numaproj/numaflow/issues/3370), [#3371](https://github.com/numaproj/numaflow/issues/3371) | +| Built-in sources | Separate resilience track; not covered by UDF reconnect. | | Reduce vertices | Excluded from automatic reconnect in Phase 1 pending durable-state design. | | Target Release | 1.9 | ## Context -Today each vertex and monovertex pod has a `monitor` sidecar, a `numa` container, and UDF sidecars. The monitor sidecar serves runtime errors from the shared emptyDir because the `numa` process exits when it receives a gRPC error from a UDF. While `numa` is restarting, it cannot host the error endpoint. +Today each vertex and monovertex pod has a `monitor` sidecar, a `numa` container, and UDF sidecars. The monitor sidecar serves runtime errors from the shared emptyDir because the `numa` process exits on runtime failures such as UDF gRPC errors. While `numa` is restarting, it cannot host the error endpoint. -The umbrella goal in #3367 is to remove the monitor sidecar. Phase 1, tracked by #3368, makes that possible by keeping `numa` alive across non-reduce UDF failures. +The umbrella goal in #3367 is to remove the monitor sidecar. Phase 1, tracked by #3368, makes that possible by keeping `numa` alive across non-reduce UDF failures. Built-in sources are also part of the broader resilience story, but they do not use UDF gRPC streams and need connector-specific follow-up work. ## Goals @@ -20,6 +21,7 @@ The umbrella goal in #3367 is to remove the monitor sidecar. Phase 1, tracked by - `numa` must continue serving metrics and health endpoints during UDF outage and reconnect cycles. - UDF errors remain observable through logs, `critical_error_total`, and persisted runtime-error files. - Processing resumes automatically once the UDF sidecar is healthy again. +- Built-in source connectors get a separate resilience track for transient broker or service failures. - Reduce vertices keep today's persist-and-exit behavior until durable reduce state is designed and approved. ## Non-Goals @@ -28,6 +30,7 @@ The umbrella goal in #3367 is to remove the monitor sidecar. Phase 1, tracked by - Changing daemon-server polling or UI-facing APIs; that is Phase 3, #3370. - Adding e2e failure coverage; that is Phase 4, #3371. - Changing source SDK at-least-once contracts. +- Implementing built-in source resilience in Phase 1; this document only records the direction and required follow-up. - Changing Go controllers, CRDs, UI, or monitor-sidecar code in this phase. ## Proposed Design @@ -40,6 +43,8 @@ All non-reduce UDF gRPC failures are handled uniformly. There is no transient vs 4. Reconnects to the UDF sidecar. 5. Redrives the in-flight request without exiting. +Built-in source failures are not handled by this UDF reconnect path. Built-in connectors such as Kafka, Pulsar, NATS, Jetstream, SQS, HTTP, and generator sources are Rust implementations that talk directly to external systems through connector-specific APIs. + ### Reconnect Helper Reconnect reuses the startup flow: @@ -92,6 +97,17 @@ The user-defined source SDK contract remains unchanged: offsets must not be comm Reduce vertices retain current behavior: persist the error and exit. Reduce UDFs hold window state in-process, so reconnecting after a crash could drop or double-count aggregates. This carve-out requires issue-owner signoff. +### Built-in Source Resilience Track + +Built-in sources need connector-owned retry and classification instead of the UDF `tonic::Status` reconnect path. The follow-up design should: + +- Retry transient startup, pending, partition-discovery, read, ack, commit, or delete errors in place when retrying does not advance source state incorrectly. +- Preserve terminal errors for misconfiguration, invalid credentials, missing permissions, unsupported broker responses, or required resources that cannot recover without operator action. +- Keep at-least-once behavior explicit per connector: source state must not be advanced until downstream processing completes, and duplicate ack/commit/delete attempts must be safe. +- Start with connectors that can restart-loop during broker coordination, rebalance, throttling, or timeout states, then audit the remaining built-in sources. + +This track complements UDF reconnect work. UDF reconnect does not automatically solve built-in source restart loops. + ## Observability Reuse `critical_error_total` with new reason strings: @@ -130,8 +146,9 @@ Safety by path: | Source nack reconnect | Conditional on SDK not committing implicit ACKs | Possible extra NAK frame | | Map and transformer redrive | None, task holds handle until success or drop | Possible UDF reinvocation | | Sink redrive | None, success is gated on sink response | Possible external writes | +| Built-in source retry | Connector-specific commit/delete/ack contract | Connector-specific; should be none for transient startup probes and idempotent ack retries | -The only conditional case is source NAK reconnect, which depends on the existing user-defined-source SDK contract. Built-in Rust source connectors do not use the UDF-source gRPC path and are out of scope for this reconnect work. +The only conditional UDF case is source NAK reconnect, which depends on the existing user-defined-source SDK contract. Built-in source safety is connector-specific and must be covered by the built-in source resilience track. ## Risks @@ -139,6 +156,8 @@ The only conditional case is source NAK reconnect, which depends on the existing - Reconnect storms are possible during sustained UDF outages. The retry interval stays at the current one-second startup retry cadence. - Reduce still restarts. This is an explicit carve-out. - SDK NAK behavior must be audited in numaflow-go, numaflow-java, numaflow-python, and numaflow-rs **as a pre-merge gate**. Any non-compliant SDK is either remediated first or carved out from the resilience guarantee. +- Built-in source classifiers can hide terminal errors if they are too broad. Each connector follow-up must document retryable and terminal errors. +- Built-in source startup probes can restart-loop if transient broker states fail fast. Follow-up work should retry those probes in place with cancellation-aware backoff. ## Testing @@ -155,3 +174,10 @@ Required tests: - MessageHandle drop-NAK invariant test during outage. - Top-level assertion that non-reduce UDF errors do not make `run()` return `Err`. - User-defined-source SDK NAK-preservation test or audit. + +Built-in source follow-up tests: + +- Startup probe retry tests for transient metadata, pending-message, offset, or partition-discovery failures. +- Ack/commit/delete retry tests that verify source state is not advanced incorrectly. +- Terminal error tests for credentials, authorization, unsupported configuration, and permanently missing required resources. +- At-least-once tests for redelivery before source-state advancement and duplicate-safe retry after source-state advancement.