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

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 , 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.

Start with apply_batch_result and handle in core/connectors/sdk/src/source.rs, then trace source_forwarding_loop, update_status, and set_error in core/connectors/runtime/src/source.rs and manager/source.rs. Agree on the policy and status-transition design first; done means transient failures do not cause unsafe loss and a stopped or recovered source reports accurate status and metrics.

Written by the indexing model from the issue text.