mlevkov commented on PR #3795: URL: https://github.com/apache/iggy/pull/3795#issuecomment-5364604684
I reassessed this against `af9ce9548` (#3855, source batch acknowledgments) instead of rebasing it, because the ack contract changes the premise this PR rests on. Three things hold on master `c0c74931b`: 1. The forwarding channel is `flume::unbounded()` (`runtime/src/source.rs:625`) and the loop is `while let Ok(batch) = receiver.recv_async().await`. flume yields queued items before it reports disconnect, so every queued batch is drained once the sender goes away. There is no drop path on master today. 2. `handle_messages` (`sdk/src/source.rs:278`) keeps `pending_batch: Option<PendingBatch>` and awaits a oneshot for the batch result before it increments `batch_id` and polls again. One batch in flight, by construction. 3. The state save runs inside the send-success branch, and `batch_result = Ack` is set only when that save also succeeded. That retires the bug your review found. "Drop batch N, the loop keeps draining, N+1 enqueues and persists a cursor covering N" needs two batches in flight, and (2) forbids it. It also means the bound this PR adds can never fill, since the producer stops after one batch until the runtime acks. So `channel_capacity`, the `dropped` latch, the grace `send_timeout` round and `iggy_connector_messages_dropped_total` all guard a path that can no longer execute, and the PR would be introducing the drop path in order to defend it. Your two comments about the stale 1024 in the doc strings are moot along with the rest. One gap does survive, and it is the piece I would keep. `BATCH_RESULT_TIMEOUT` is 30s (`sdk/src/source.rs:54`). When the forwarding loop lags past it the SDK nacks, re-polls and enqueues a new batch while the timed-out one is still queued, so depth grows by about one batch per 30s per source. `MAX_CONSECUTIVE_NACKS = 5` (`:57`) looks like a ceiling, but any successful ack resets the counter, so an alternating ack/timeout pattern never trips it. Ordering stays safe (FIFO plus a sequential loop), so this is memory pressure rather than cursor corruption. Worth noting that a bounded channel only helps here if the callback blocks. Dropping would reintroduce exactly the loss this PR set out to prevent. One more thing I noticed while reading: a batch that times out and then sends successfully has its Ack refused by the plugin on the `take_pending_batch` id mismatch (`sdk/src/source.rs:466`), so the runtime logs an error and sets connector error state for a batch that actually shipped fine. Two questions, since the call is yours: - Close this and file the 30s timeout growth separately, or reshape this PR around it as a blocking handoff with no drop? - Does the same reasoning retire #3898? It needs a later batch to ship while an earlier one failed, which (2) also forbids. #3797 hangs off this decision: it wraps the send callback in `block_in_place` for backpressure that only exists if this bound lands. I have left both unrebased pending your answer. #3798 and #3804 are rebased onto master and green. -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
