diff --git a/docs/specifications/numa-resilience-doc.md b/docs/specifications/numa-resilience-doc.md new file mode 100644 index 0000000000..183c3e48bd --- /dev/null +++ b/docs/specifications/numa-resilience-doc.md @@ -0,0 +1,183 @@ +# 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 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. 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 + +- 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. +- 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 + +- 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. +- 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 + +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. + +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: + +- 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. 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 + +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. + +### 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: + +- `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`, 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. + +## 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 | +| 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 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 + +- 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 **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 + +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. + +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.