[sentry-junior]

Fixes #583.

RunTaskInThreads::poll could silently drop messages when the next step applied backpressure. There were two ways this happened:

1. Several finished tasks rejected in one poll(). The loop over finished handles kept going after a MessageRejected, so each rejection overwrote message_carried_over. This has been there since the strategy was first added (feat(rust-consumer): Add RunTaskInThreads strategy snuba#4537), and it's the case the Rust RunTaskInThreads loses pending messages when downstream applies backpressure #583 repro hits.
2. Retry of the carried-over message fails, then the loop forwards another task. The original code returned early after a failed retry. That early return was dropped when the if let became a match in feat(rust): Strategy can return InvalidMessage error in poll, join, submit snuba#4929, so since then the loop runs anyway and can overwrite the pending message.

The fix stops forwarding finished tasks while message_carried_over is set. Later results stay in order in the handles queue until the carried-over message is accepted, and submit() already applies backpressure upstream while a message is carried over.

Tests

I added test_backpressure_with_multiple_finished_tasks_does_not_drop_messages. It submits 3 messages, waits for all tasks to finish, polls 3 times while the next step rejects everything, then unblocks and joins. It asserts that [1, 2, 3] are delivered in order.

To check that the test covers both paths, I ran it against three broken variants. Each one failed, delivering [3]:

• no fix
• only path 1 fixed (break after an in-loop rejection)
• only path 2 fixed (the original early return after a failed retry)

Checks run locally:

• cargo fmt --check: passes.
• cargo test --lib -- --skip backends::kafka: 44 passed.
• cargo clippy --all-targets: same warnings as main, nothing new.
• Kafka-backed tests: not run, because there's no broker in my environment. CI covers them.
• The reproducer from Rust RunTaskInThreads loses pending messages when downstream applies backpressure #583, run against this branch: delivers [1, 2].

Not in scope

submit() checks self.handles.len() > self.concurrency, which lets one more task run than the configured concurrency allows. It predates both bugs. Changing it affects throughput, so I'd rather handle it separately.

via markus.

--

View Junior Session [Sentry]