feat(connectors): add RabbitMQ sink - #3973
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 #3973 +/- ##
============================================
+ Coverage 84.06% 84.89% +0.82%
- Complexity 1358 1405 +47
============================================
Files 1217 1225 +8
Lines 171902 179743 +7841
Branches 139658 146056 +6398
============================================
+ Hits 144508 152585 +8077
+ Misses 23465 23105 -360
- Partials 3929 4053 +124
🚀 New features to boost your workflow:
|
|
Thanks for the comments @numinnex ! Addressed, pls take a look |
|
/ready |
| // Confirms resolve in publish order over already-overlapped RTTs; returns are | ||
| // matched by message_id (lapin's Return-to-confirm FIFO has no delivery tag), so a | ||
| // post-publish failure re-publishes the already-delivered tail (at-least-once in batch). |
There was a problem hiding this comment.
The comment claims returns are "matched by message_id", but nothing matches on it. returned_id at L253 is only {:?}-formatted into the error string, and confirmed is never corrected.
lapin staples a return to whichever confirm resolves next: complete_pending pops returned_messages.get_waiting_message() from a FIFO with no delivery tag (lapin-2.5.5/src/acknowledgement.rs:115-121), and for Basic.Ack{multiple:true} — RabbitMQ's normal mode under load — complete_pending_before collects tags into a HashSet<DeliveryTag> before iterating (acknowledgement.rs:155-168, reached via channel.rs:1144-1147). So the offset reported at L254 is nondeterministic and usually not the message that was actually returned; the genuinely unroutable one resolves Ack(None) and is counted into confirmed.
This is the misattribution hazard flagged in the earlier confirm-pipelining thread, and that thread was resolved on the premise that message_id matching had been implemented. It has not been. Either parse returned.delivery.properties.message_id() back to an offset and fail only that message, or drop the claim from the comment and document the offset as unattributable.
| // post-publish failure re-publishes the already-delivered tail (at-least-once in batch). | ||
| for (confirm, offset) in in_flight { | ||
| match timeout(self.timeout, confirm).await { | ||
| Ok(Ok(Confirmation::Ack(None))) => confirmed += 1, |
There was a problem hiding this comment.
confirmed += 1 here can count a message the broker refused.
A break on any earlier arm drops the remaining PublisherConfirms, but dropping one does not remove its broadcaster from Acknowledgements::Inner::pending (acknowledgement.rs:99-111 inserts; only drop_pending/drop_all remove). When that orphaned tag's ack finally arrives, complete_pending (acknowledgement.rs:113-121) unconditionally pops get_waiting_message() and staples a later batch's Basic.Return onto the dead confirm, where register_dropped_confirm parks it unseen (returned_messages.rs:116-130). The live confirm for that later batch then resolves Ack(None) and lands on this line.
Result: an unroutable message is counted in confirmed and messages_published, consume() returns Ok(()), and the runtime has already committed the offset (AutoCommitWhen::PollingMessages, core/connectors/runtime/src/sink.rs:522) while discarding the FFI return code. Silent loss reported as a successful delivery, with no log and no metric.
The window is not narrow. delivery_mode = 2 (the default) defers the ack until RabbitMQ's batched msg-store fsync (~100-200 ms), while the shipped poll_interval = "5ms" re-enters consume() two orders of magnitude sooner, so the next batch is publishing on the same channel while the previous batch's orphaned tags are still unacked. The two preconditions are also correlated: the Ack(Some) break that creates the orphans is the same unroutable path that produces the returns.
Fixed by never reusing a channel that has unresolved confirms — see the comment on the confirm loop below.
| Ok(Ok(Confirmation::Ack(Some(returned)))) => { | ||
| let returned_id = returned.delivery.properties.message_id(); | ||
| last_error = Some(Error::InvalidRecordValue(format!( | ||
| "message offset {offset} (id {returned_id:?}) returned as unroutable by RabbitMQ" | ||
| ))); | ||
| last_retryable = false; | ||
| break; | ||
| } | ||
| Ok(Ok(Confirmation::Nack(_))) => { | ||
| last_error = | ||
| Some(Error::CannotStoreData("message nack'd by RabbitMQ".into())); | ||
| last_retryable = true; | ||
| break; | ||
| } | ||
| Ok(Ok(Confirmation::NotRequested)) => { | ||
| last_error = Some(Error::CannotStoreData( | ||
| "publisher confirms not enabled".into(), | ||
| )); | ||
| last_retryable = false; |
There was a problem hiding this comment.
Every break in this loop abandons the unread in_flight tail, and the two permanent arms (Ack(Some) here, NotRequested at L266) return at L296-302 without clear_state(), so the same channel is reused on the next poll.
PublisherConfirm::drop hands each abandoned confirm to register_dropped_confirm (lapin-2.5.5/src/publisher_confirm.rs:81-88 -> returned_messages.rs:116-130), which parks the promise in Inner::dropped_confirms or, if a return already resolved it, the whole BasicReturnMessage — including the full Delivery.data payload — in Inner::messages. Both are cleared only by Inner::drain(), whose sole caller is Channel::wait_for_confirms (channel.rs:334-341). This sink never calls it.
With a misconfigured routing key and mandatory: true, a batch_length = 100 batch retains ~99 entries with their payloads, every poll, for the process lifetime. Unbounded RSS growth with no diagnostic. The same orphans are what enable the counted-as-delivered case flagged at L251.
Suggest clear_state().await on every non-Ok exit from this loop (safer than wait_for_confirms(), which can itself block on a stalled broker), so a channel with unresolved confirms is never reused.
| .await | ||
| .as_ref() | ||
| .map(|s| s.channel.clone()) | ||
| .ok_or_else(|| Error::Connection("RabbitMQ not connected".into()))? |
There was a problem hiding this comment.
This can drop a whole batch without consuming a single retry attempt.
reconnect() is a try-lock: a caller that loses the CAS at L372-378 sleeps retry_delay and returns Ok(()) having connected nothing. Control then arrives here, state is still None, and the ? propagates Error::Connection("RabbitMQ not connected") straight out of publish_batch_with_retry with attempts == 0 and publish_errors never incremented — the retry loop, the backoff and the error accounting are all bypassed.
The runtime discards the FFI return code and has already committed the offset at poll time (core/connectors/runtime/src/sink.rs:522 and :745-753), so the batch is unrecoverable and invisible.
Concurrency is real whenever a connector has 2+ (stream, topic) pairs: setup_sink_consumers builds one consumer per pair (runtime/src/sink.rs:519-546) and spawn_consume_tasks spawns one task each against the same plugin instance (:266-299). The window is widest exactly when it matters — the CAS winner can hold the flag for up to timeout_secs (30s default, and unbounded given the untimed channel RPCs noted at L395) while the loser gives up after 1s.
The loser should wait on the winner's outcome and re-read state, or return a retryable error. It should not report success for work it did not do.
| props = props.with_headers(headers); | ||
| } | ||
|
|
||
| let confirm = match timeout( |
There was a problem hiding this comment.
Wrapping basic_publish in timeout() does not make it abortable, so a publish that "times out" here has usually already been delivered.
before_basic_publish registers the pending confirm and burns a delivery tag before anything is sent (lapin-2.5.5/src/channel.rs:503-509), and send_method_frame_with_body pushes every frame into Frames and wakes the io_loop synchronously on first poll, only then awaiting the write promise (channel.rs:393-408 -> frames.rs:149-170). tokio's Timeout polls the inner future before checking the deadline, so that prologue always runs.
The timeout arm at L237-241 treats this as not-sent: it breaks without clear_state() and the retry republishes messages[confirmed..], so the tail is delivered twice and each cancelled future leaves another orphaned confirm on the live channel (see L252). A publish timeout is a sent-unknown outcome, not a not-sent one.
Related: timeout_secs is accepted unvalidated at L151, so timeout_secs = 0 makes every publish time out while every message still lands.
| let result = async { | ||
| let conn = timeout( | ||
| self.timeout, | ||
| Connection::connect( |
There was a problem hiding this comment.
This timeout does not bound the connect syscall, which is the case raised in the earlier timeouts thread ("a blackholed SYN here waits out the kernel TCP timeout").
lapin runs the connect under executor.spawn_blocking (lapin-2.5.5/src/connection.rs:300-337), and amq-protocol-tcp-7.2.3/src/lib.rs:43-48 calls a bare TcpStream::connect unless the URI carries ?connection_timeout=<ms>. Since ConnectionProperties::default() leaves the executor unset, that lands on async-global-executor's blocking pool. When the outer timeout fires it cancels the await only; the blocking closure keeps running to the kernel SYN timeout (~127s), leaking one wedged pool thread per attempt, in a pool the connectors runtime cannot see.
Minimal fix is to append ?connection_timeout={timeout_ms} to the AMQP URI (or supply a custom connector). Same applies to the open() path at L433.
| .create_channel() | ||
| .await | ||
| .map_err(|e| Error::Connection(e.to_string()))?; | ||
| channel | ||
| .confirm_select(ConfirmSelectOptions::default()) | ||
| .await | ||
| .map_err(|e| Error::Connection(e.to_string()))?; | ||
| let exchange_kind = self.exchange_kind()?; | ||
| channel | ||
| .exchange_declare( |
There was a problem hiding this comment.
create_channel, confirm_select and exchange_declare are awaited with no timeout, in both reconnect() and open() (L442-455). Only Connection::connect is wrapped.
A broker that completes the TCP handshake and answers heartbeats while stalling channel RPCs — the usual shape of a disk or memory alarm, or a half-open connection behind a load balancer — hangs here, bounded only by the negotiated heartbeat (~2x60s on RabbitMQ defaults) and unbounded if the server negotiates heartbeat=0.
This compounds the reconnect() try-lock: the CAS winner is parked inside this block with reconnecting == true and no deadline, so every concurrent consume task takes the loser path and drops its batch (see the comment at L196).
Wrapping the whole open/reconnect body in a single timeout(self.timeout, async { .. }) covers these three and the await half of the connect.
|
/author |
github-actions
Bot
added
S-waiting-on-author
3 participants
Replaces #3811 (could not be reopened after rebasing onto master). Rebased to current master and addressed all review comments.
Which issue does this PR address?
Relates to #3747
Summary
Adds the RabbitMQ sink connector via the lapin client. Source will be a separate PR.
Review feedback addressed
amqp_urlstored assecrecy::SecretString, redacted viaiggy_common::serde_secret(never logged or serialized verbatim)durable_exchangeconfig (defaulttrue)Ack(None)counts as success;Ack(Some(_))/Nack(_)(unroutable mandatory publish) fail the batch permanentlydelivery_modeconfig (defaultpersistent)LongStringfor strings,ByteArrayfor binary) so headers exchanges can route on themiggy_offsetencoded as fulli64instead of narrowing tou32basic_publisherrors routed through the same retry flow; the retry loop resumes at the first unconfirmed message instead of republishing confirmed onesPayload::try_to_bytes()(no deep clone)AI usage
Which tools? opencode
Scope of usage? investigation, code suggestions, implementation of review feedback
How did you verify the generated code works correctly? Read through, compiled, ran unit tests, ran clippy/fmt/sort
Can you explain every line of the code if asked? Yes