connectors: fix producer.send() failure lets a later batch supersede the skipped state save

[mlevkov]

Split out of the #3795 review at hubcio's suggestion, so it is not lost when that PR merges.

The shape

source_forwarding_loop persists a source's ConnectorState only on the success branch of the Iggy send. When producer.send() fails, the loop logs, skips the save, and continues to the next batch.

Sources advance their cursor at poll time and snapshot it into every batch (tracking_offsets in postgres_source, same shape in the others). So the next batch that does land persists a cursor that already covers the batch whose send failed, and a restart resumes past it.

That is the same supersession #3795 fixes for the shutdown-drop path with a per-instance dropped latch. The latch does not close this one: it is armed only by a drop, not by a send failure.

Why it is filed separately rather than fixed there

Two differences from the shutdown case, both of which argue for treating it on its own:

• It predates fix(connectors): bound source forwarding channel with backpressure #3795 and exists on master today.
• It flips the connector to Error status, so unlike the shutdown drop it is at least visible, rather than only logged and counted.

Whether the fix is the same latch armed from this path, a retry, or something that surfaces the gap rather than hiding it, is a design question worth its own discussion.

Happy to take it once the direction is settled.

[mlevkov]

Confirming the adoption question and adding what I found while auditing the shipped sources, since it bears on which of the two scopes actually fixes this.

random_source is the only adopter, yes: sources/random_source/src/lib.rs:196 is the single on_batch_result in the tree.

The other three all advance a cursor inside poll() and rely on the default no-op, so a NACK leaves the plugin's in-memory position ahead of the persisted one and the next poll skips the batch:

• postgres_source: state.tracking_offsets.insert(table, offset) at lib.rs:578
• influxdb_source: apply_v2_cursor_advance(&mut state, ...) at lib.rs:521
• elasticsearch_source: last_document_id / last_scroll_id / last_offset in State (lib.rs:51-53), written during the poll

postgres_source is worse than a cursor skip, and it is the case that decides the scope. Inside poll(), before the runtime has been handed anything, poll_tables calls mark_or_delete_processed_rows (lib.rs:557), which issues either DELETE FROM <table> WHERE <pk> IN (...) (lib.rs:614) or UPDATE <table> SET <processed> = TRUE (lib.rs:632) against the source database. So on a NACK the rows are already gone from Postgres, or already marked consumed. That loss is not recoverable by a restart, unlike the cursor cases where the persisted state is still behind.

Which means the runtime switch on its own does not close this. Having the runtime stop saving after a NACK until the next Ack prevents a later batch from persisting a cursor past a dropped one, which is the original report here. It cannot undo the DELETE, and it cannot rewind the plugin's in-memory tracking_offsets, so the next poll skips those rows regardless. For the shipped sources the fix has to be on the plugin side.

On deleting the trait default, which I read as your preference: the value is that it converts a silent no-op into a compile error, so a source author cannot omit the hook without noticing. Worth being clear that it does not by itself make the sources correct, since each still needs a real stage-and-apply implementation, and postgres_source's is the substantive work: the deletes and offset writes have to move out of poll() and only run on Ack.

One semver note in its favour. Removing the default breaks source compatibility for out-of-tree plugins, but it does not touch any FFI signature, so pre-built plugin .so files keep loading. That makes it a cheaper break than a trait change usually implies.

Happy to take the postgres_source staging rewrite, or the whole sweep across the three sources, whichever is more useful. http_source on #3798 already implements the hook this way, holding each batch until Ack and replaying on Nack, so there is a worked example to follow.