Source NACK breaker stops non-replayable sources, and the stop is unobservable

[mlevkov]

Follows af9ce9548 (#3855). Everything below is read off master c0c74931b.

The batch acknowledgment contract ends with a circuit breaker: after five consecutive NACKs the SDK stops polling the source. That is the right call for a source that can re-read its position, and the wrong one for a source whose unsent work exists only in memory, because for the latter "stop" is not a safe state, it is data loss. There is also no way for a plugin to opt out, and no way for an operator to see that it happened.

The breaker

apply_batch_result (core/connectors/sdk/src/source.rs:511) counts NACKs and stops the poll loop once the count reaches MAX_CONSECUTIVE_NACKS = 5 (:57, checked at :530). BatchCompletion::Stop breaks out of handle_messages (:398), so the poll task ends. An Ack resets the counter, so the five have to be consecutive.

The window is short. nack_retry_delay (:544) passes consecutive_nacks - 1 to exponential_backoff (core/connectors/sdk/src/retry.rs:164) with a 100ms base (:55), so the sleeps before the fifth NACK are 100, 200, 400 and 800ms, about 1.5s in total. The runtime's own retry does not lengthen it much: send_with_failed_tail_retries (core/connectors/runtime/src/source.rs:785) retries at most three times (:55), only for IggyError::ProducerSendFailed with a non-empty failed tail, and with no sleep between attempts. Any other error returns on the first attempt.

So roughly two seconds of a broker being unavailable is enough to stop a source permanently. A rolling upgrade, a leader election or a short partition all clear that bar.

Why "stop" is not safe for every source

A source that stops is expected to be restarted and to re-read from its persisted cursor, losing nothing. The trait doc says as much (core/connectors/sdk/src/lib.rs:107): stage cursor changes in poll and apply them on Ack.

That reasoning does not hold for a source whose input is pushed to it. http_source is the concrete case. It answers HTTP 200 to a webhook sender, which is a transfer of ownership: the sender will never send that event again. The accepted events live in an in-memory crossfire bridge with a default capacity of 10,000. When the breaker fires, the poll task ends while the HTTP listener keeps accepting, so the bridge fills, handlers start answering 429, and nothing drains. A restart clears the bridge, so up to 10,000 events that were acknowledged with a 200 are gone.

The transient failure that triggered this is exactly the case where holding is correct and stopping is not. #3798 now holds a NACKed batch and replays it on the next poll rather than dropping it, so the connector recovers a failed send on its own. The breaker cuts that recovery off after about two seconds.

A plugin cannot work around it

on_batch_result returns Result<(), Error>, and an Err maps to BatchCompletion::Stop immediately (:520). There is no return value that means "keep polling, this failure is transient". BatchPolicy (:94) is constructed with BatchPolicy::default() inside handle (:249) and is not reachable from configuration.

There is one loophole and it should not be the answer. The counter resets on any Ack, and an empty batch always acks because IggyProducer::send returns Ok(no_confirmations()) early for an empty vector (core/sdk/src/clients/producer.rs:570). A plugin can therefore emit an empty batch purely to reset the breaker and survive an outage of any length. That defeats a deliberate safety mechanism from inside a plugin, invisibly to anyone reading the SDK, so I have deliberately not done it in #3798.

Second, separable defect: the stop is unobservable

This is what turns an aggressive breaker into a silent one. It may be worth splitting into its own issue.

The only signal that the breaker fired is one error! line at :531. Nothing else changes:

• The runtime's forwarding loop does not exit. It is parked on receiver.recv_async() and its sender is still registered in SOURCE_SENDERS, so update_status(..., ConnectorStatus::Stopped, ...) at core/connectors/runtime/src/source.rs:603 never runs.
• set_error (core/connectors/runtime/src/manager/source.rs:104) sets info.status = Error but takes no metrics handle, unlike update_status (:79). Only update_status moves the iggy_connectors_sources_running gauge (core/connectors/runtime/src/metrics.rs:197).
• Nothing restores Running after a successful send. In source_forwarding_loop, Running is set once at :391 and set_error is called at :505, :537 and :575, so Error is sticky: a connector that failed once and fully recovered still reports Error.

The result is that a source killed by the breaker reports Error with a stale last_error, indistinguishable from one that erred once and recovered, while iggy_connectors_sources_running still counts it as running, indefinitely.

End to end

1. The Iggy broker restarts. It is unavailable for about three seconds.
2. producer.send fails. The runtime sets Error and NACKs. The plugin holds its batch and replays it.
3. Five NACKs, about 1.5s of backoff. apply_batch_result returns Stop and the poll task ends.
4. The broker comes back. Nothing polls. The forwarding loop is parked, so no status transition happens.
5. iggy_connectors_sources_running still counts the source. The status API says Error.
6. Webhooks keep being accepted until the bridge hits 10,000, then senders get 429 and retry.
7. An operator eventually restarts the connector. The bridge is in memory, so up to 10,000 events that received a 200 are lost.

Proposal

1. Make BatchPolicy configurable per connector rather than always BatchPolicy::default(). result_timeout and max_consecutive_nacks are the two that matter, and a source that cannot replay needs to disable the breaker outright.
2. Let a source keep polling across a NACK it considers transient. A dedicated variant, or a bool from on_batch_result, is more honest than the empty-batch loophole above.
3. When the poll task does end on its own, make it observable: transition the connector out of Running, decrement the gauge, and give set_error the same metrics handle update_status has, so the running count cannot disagree with the reported status.
4. Clear Error on a subsequent successful send so the status means "failing now" rather than "failed once".

Happy to send a PR for any of these, but the shape of 1 and 2 is a design call, so I would rather agree on it first. Related: the BATCH_RESULT_TIMEOUT observation in #3795 (comment), which is the other place MAX_CONSECUTIVE_NACKS does not behave the way its name suggests, since an alternating ack/timeout pattern never trips it.