mlevkov commented on PR #3795:
URL: https://github.com/apache/iggy/pull/3795#issuecomment-5310808521

   Thanks - the shutdown-drop finding is a real bug and I had it characterised 
wrong in the PR description. Pushed `8c4100e62` on top of your branch update.
   
   **The silent supersession.** Traced it and you are right: the drop returns, 
the forwarding loop keeps draining so a slot frees, the next batch enqueues and 
ships, and the save that follows carries a cursor covering the batch that was 
dropped. Calling that tail truncation was wrong; it is mid-stream loss, and it 
fires on SIGTERM and on API restart, not on some exotic path.
   
   Implemented the per-instance latch as you described. First drop sets 
`dropped`, every later batch from that instance is dropped too, so the 
persisted cursor can never pass the gap. Queued batches still flush since the 
ring is FIFO and they all predate the drop. The `error!` latches with the flag; 
the counter keeps moving per batch.
   
   Mutation-checked: removing the latch check fails the new test with "a batch 
after the gap must be dropped even with capacity free, or the persisted cursor 
would advance past the batch that was lost".
   
   **Grace round** before the first drop, as suggested - one bounded 
`send_timeout`, skipped once latched. Its test fails if the round is removed.
   
   **Dedicated counter.** `iggy_connector_messages_dropped_total`, 
`inc_by(message_count)`, separate from `errors` so permanent loss is not mixed 
in with decode and send failures that get retried.
   
   **Capacity default 1024 -> 64**, landed together with the fix rather than 
before it, per your sequencing note. READMEs and the skill doc follow.
   
   **Tests.** Both backoff tests move to capacity 16, since capacity 1 routes 
to `OneMpsc` and a different backoff regime than the shipped default - the 
park/wake path being pinned was not the one that ships. The two shutdown-signal 
tests are merged into one hermetic test with two entries (target set, sibling 
not, then `signal_shutdown_all` sets both); you were right that the sibling's 
whole-map call could have covered for `signal_shutdown` being a no-op.
   
   **Also taken:** close return code checked in `stop_connector` the way `init` 
already does it; `DashMap<u32, Arc<SourceSenderEntry>>` with 
`&SourceSenderEntry` passed down (6 params -> 3, and the flags become plain 
`AtomicBool`); unsafe narrowed to `from_raw_parts`; 
`DEFAULT_CHANNEL_CAPACITY`/`BatchSender`/`BatchReceiver` dropped to private; 
`recv()` named in the drain comment; the steady-state worker-starvation point 
added to the code comment and the skill doc; README now states the guarantee 
(resume from the last delivered batch's state, duplicates for offset sources 
and permanent loss for `delete_after_read`/`processed_column`) and qualifies 
the env override as local-provider-only; `config_format` -> 
`plugin_config_format` in the sources README.
   
   **Left for follow-ups, as you framed them:** the elasticsearch_source cursor 
override, the `producer.send()` err path with the same supersession shape, the 
`spawn_source_handler` parameter count (needs the resolved-path/version split 
first), and the `iggy_source_handle` return code. Happy to file those as issues 
so they do not evaporate - say the word and I will, or I can take any of them 
here.
   
   The saturate-stop-restart-assert-no-gap regression test you suggested is the 
one I would most like to add, but it belongs in the postgres integration suite 
next to `state_persists_across_connector_restart` rather than in this diff. 
Also happy to do that as a follow-up PR.
   
   Note on the branch: you had updated it with a merge from master, so I 
rebased my commit on top of that rather than force-pushing, and this was a 
normal fast-forward push - your merge commit is intact.
   
   Gate: fmt, sort, clippy `-D warnings`, `cargo test -p iggy-connectors` 134 
passing, taplo, typos, license headers (via `hawkeye check`; 
`scripts/ci/license-headers.sh` now requires bash >= 4.2 after #3837 and will 
not run on macOS's 3.2).
   


-- 
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]

Reply via email to