feat(connectors): add source batch acknowledgments - #3855
feat(connectors): add source batch acknowledgments#3855rohankumardubey wants to merge 4 commits into
Conversation
|
Thanks for the PR. It is labeled Slash commands (own line, regular comment) move it around the queue:
See CONTRIBUTING.md for details. |
Codecov Report❌ Patch coverage is
Additional details and impacted files@@ Coverage Diff @@
## master #3855 +/- ##
=============================================
- Coverage 82.72% 67.07% -15.66%
Complexity 1296 1296
=============================================
Files 1199 1199
Lines 160319 136350 -23969
Branches 129924 106059 -23865
=============================================
- Hits 132625 91451 -41174
- Misses 24188 41303 +17115
- Partials 3506 3596 +90
🚀 New features to boost your workflow:
|
|
/ready |
|
@hubcio This PR is ready for review. |
hubcio
left a comment
hubcio
left a comment
There was a problem hiding this comment.
i actually had comments ready yesterday, just as pending, forgot to post them. here a few things that don't sit on changed lines:
- no test exercises the failure path - nothing forces a send or state-save failure and checks that the nack arrives and the state file stays put. and no integration test ever waits for a second batch, so a handshake that wedges after batch 1 still passes the suite.
manager/source.rs:150comment "close FFI (stops callbacks)" is stale now - close doesn't stop the runtime->plugin result callback, it guarantees it fails.- the connector authoring docs and templates don't mention
on_batch_result, so the next source written from them lands with the no-op default too.
| } | ||
|
|
||
| if state_saved { | ||
| batch_result = SourceBatchResult::Ack; |
There was a problem hiding this comment.
ack fires even when messages were dropped before the send - decode failures above and transform/encode/build failures inside process_messages shrink the batch, but nothing compares sent_count to count. extreme case: every message drops, send(vec![]) returns ok without touching the server, still ack. under this contract ack tells the plugin to commit destructive work, so dropped messages are gone at the source.
process_messages already counts its errors - return that and nack when decode_errors + error_count > 0 (keep filtered_count out, those drops are intentional). one catch: a plain nack on a message that can never decode means infinite redelivery, so this also needs a drop-after-n-attempts or dlq policy.
| // Total histogram + emit (below) run regardless of send outcome. | ||
| let mut state_save_us: Option<u64> = None; | ||
| let mut batch_result = SourceBatchResult::Nack; | ||
| if let Err(error) = send_result { |
There was a problem hiding this comment.
the error here already says which chunks committed (ProducerSendFailed { committed, failed, .. }) but it gets formatted into a string and the whole batch is nacked - redelivery then duplicates the committed prefix. resending just failed before deciding ack/nack would avoid that; at minimum log committed.len().
| } | ||
|
|
||
| let result_code = batch_result_callback(plugin_id, batch_id, batch_result as u8); | ||
| if result_code != 0 { |
There was a problem hiding this comment.
every stop with a batch in flight ends up here: iggy_source_close removes the INSTANCES entry before cleanup_sender runs, so the callback returns -1 and this logs an error, bumps the counter and sets connector error for a benign shutdown race. handle_produced_messages below treats the mirror case as trace-only for exactly this reason. side effect: the loop exit then sees status Error, so update_status(Stopped) skips the sources_running decrement and the gauge leaks on every restart.
| ); | ||
| } | ||
|
|
||
| let mut state_saved = true; |
There was a problem hiding this comment.
state is saved (two fsyncs) every batch even when nothing moved. letting a source return state: None when nothing changed would skip it - the runtime already acks on None - though postgres would first have to stop re-stamping last_poll_time on every poll.
|
|
||
| #[doc(hidden)] | ||
| pub fn complete_batch(&self, batch_id: u64, result: u8) -> i32 { | ||
| let Ok(result) = SourceBatchResult::try_from(result) else { |
There was a problem hiding this comment.
an unrecognized code returns -1 without completing the pending batch - the plugin then waits on result_receiver forever and the runtime waits on recv_async, a permanent stall on both sides. safer to treat unknown as nack and complete the batch. unreachable today, but it arms the moment SourceBatchResult grows a third variant, so worth fixing before that ever happens.
| return -1; | ||
| } | ||
|
|
||
| let Some(current) = pending.take() else { |
There was a problem hiding this comment.
this else-arm is unreachable (the guard above already proved Some) and returns -1 with no log. Option::take_if collapses the check and the take - just keep the expected id in the error message, a mismatch is the only visible sign of counter desync.
| }); | ||
| } | ||
|
|
||
| if callback(plugin_id, batch_id, messages.as_ptr(), messages.len()) != 0 { |
There was a problem hiding this comment.
messages stays alive across the whole result wait, but the runtime copies the bytes out before the callback returns - dropping it right after the callback frees the duplicate buffer for the whole send window.
| /// A [`source::SourceBatchResult::Nack`] means the staged changes must be discarded so the | ||
| /// batch can be polled again. The SDK allows only one batch to be in flight at a time and | ||
| /// stops polling if this method returns an error. | ||
| async fn on_batch_result(&self, _result: source::SourceBatchResult) -> Result<(), Error> { |
There was a problem hiding this comment.
nothing in-tree implements this - every shipped source keeps the no-op default, and postgres still deletes/marks rows and advances its cursor inside poll(), so a nack rolls back nothing. worse, the loss becomes durable: after a failed send the next poll is empty (rows already consumed) but still carries the advanced cursor, send(vec![]) short-circuits ok, and the state file gets saved. the postgres migration needs to land with this, or the readme claims scoped down until an adopter exists.
| 4. The runtime reports `SourceBatchResult::Ack` to the plugin. A send or state-save failure reports `SourceBatchResult::Nack` instead. | ||
| 5. `Source::on_batch_result()` commits or discards the plugin's staged work before the next poll starts. | ||
|
|
||
| An empty batch follows the same handshake. This prevents a successful no-op send from persisting state left over from an earlier failed delivery. Producer errors, including request timeouts, report a NACK. A successful send from the legacy Iggy server is still an ACK even though that server returns an empty confirmation list. |
There was a problem hiding this comment.
this sentence only holds if the plugin rolled back on the earlier nack - no shipped plugin does, so today the empty poll is exactly what persists the stale cursor (see postgres). worth rewording until an adopter exists.
|
|
||
| The crash behavior is intentionally at-least-once: | ||
|
|
||
| | Crash point | Recovery behavior | |
There was a problem hiding this comment.
the table misses a server-side row: a batch is confirmed once committed in server memory, not fsynced, so ack -> destructive cleanup -> server crash before fsync loses the data on both ends. worth a row here plus a note on Ack that it means committed-in-memory (durability rides on the server's enforce_fsync, which ships off).
github-actions
Bot
added
S-waiting-on-author
|
ignore failing java examples, we'll check it |
|
Thanks for the detailed review. I agree with the issues you identified and will address the ABI mismatch, shutdown and timeout handling, callback error propagation, NACK backoff, docs/templates, and missing regression tests. For processing failures, I propose NACK with capped backoff, eventually stopping the connector without ACKing instead of losing data. For partial producer failures, I’ll report committed/failed counts and retry only the failed tail before NACKing. As agreed in #3635, I’ll keep PostgreSQL migration in the follow-up PR, use a small generic source to test this contract, and narrow the README claims until PostgreSQL is migrated. Does this look good? |
|
@rohankumardubey looks good. |
2 participants
Which issue does this PR address?
Relates to #3635
Rationale
Establishes the generic source acknowledgment contract before migrating PostgreSQL source behavior and adding the kill-server regression test in a follow-up PR.
What changed?
Source plugins could poll another batch without knowing whether the previous batch was delivered, allowing cursors and persisted state to advance after failed sends.
The source FFI now carries batch IDs and supports ACK/NACK results. The SDK permits one in-flight batch, while the runtime ACKs only after both the Iggy send and state persistence succeed. Send failures, timeouts, and state-save failures produce a NACK.
Local Execution
cargo fmt --alliggy_source_batch_resultsymbol is exportedcargo-sort,markdownlint, andtaplowere not available locally