fix(partitions): complete polls on the owning shard - #4119
fix(partitions): complete polls on the owning shard#4119diegomrsantos wants to merge 21 commits into
Conversation
|
#4092 will be merged first :) |
|
@diegomrsantos have you joined our discord? if yes, what's your username? |
Codecov Report❌ Patch coverage is Additional details and impacted files@@ Coverage Diff @@
## master #4119 +/- ##
=============================================
- Coverage 86.38% 76.25% -10.14%
+ Complexity 1455 1454 -1
=============================================
Files 1259 1261 +2
Lines 205992 192484 -13508
Branches 171193 157687 -13506
=============================================
- Hits 177955 146775 -31180
- Misses 23597 41269 +17672
Partials 4440 4440
🚀 New features to boost your workflow:
|
b0f7638 to
c4e45d4
Compare
A disk poll that completes after purge can restore an offset from the deleted history and cause Next to skip fresh messages. Exercise the production disk read, purge, and offset paths with a deterministic schedule. Keep the regression intentionally failing and include a passing control with automatic commit disabled.
An offset from purged history can also fit within the fresh history, so an upper bound check alone cannot establish that consumer progress is valid. Cover both fresh history sizes and verify the complete disk read before purge. Describe the completion output as an automatic commit candidate. Keep the regressions failing until the production fix is implemented.
Document why the reproduction delays completion, reuses message offsets, and covers an old offset within the fresh range. Clarify that the tests observe automatic commit candidates before server admission.
Explain the old poll and fresh messages before discussing reused offsets, so the comments can be understood without tracing the helper first.
Reservations stay on the owning shard and belong to one consumer. Express those constraints in the token API and remove unnecessary ownership requirements from capacity admission. Extract synchronous poll admission and share queue cleanup while preserving history validation and overload behavior. Validated with affected crate tests on macOS and Linux, plus the Linux cluster consumer offset quota regression.
Poll completion must reject a full persistence queue before recording consumer progress or assigning an operation. Apply the persistence admission guard on the owning shard and cover rejection followed by retry. Adapt recovery and consumer group test fixtures to the current polling and persistence APIs, including reads through the shard message pump. Co-authored-by: Codex <noreply@openai.com>
c4e45d4 to
d89b81f
Compare
|
@diegomrsantos which one should be reviewed/merged first, this one or #4122? |
|
I’d prioritize this one, since it fixes the correctness issue. #4122 contains the benchmark tooling and works against master without this fix. Neither PR depends on the other, so they can be reviewed independently, but I’d merge this one first. |
Poll completion crosses asynchronous reads, owner admission, and offset replication, so maintainers need the lifetime and ordering contracts at these boundaries. Document the flow, history identity, capacity guards, and reply semantics on the relevant types, fields, and functions. Co-authored-by: Codex <noreply@openai.com>
|
Added a short overview of poll completion and concise docs for the new types, fields, and functions. These explain history validation, reservation lifetimes, admission checks, and why poll replies can precede durable offset commits. This update changes documentation only. |
Reservation bookkeeping can use Rc and Cell because guards stay on the owner thread, outside the shard inbox's Send requirement. Maintainers also need to preserve capacity checking and acquisition as one operation. Document these constraints on the completion message, reservation types, and capacity admission method. Co-authored-by: Codex <noreply@openai.com>
|
Added concise docs explaining the reservation ownership boundary. Previously, |
A missing owner reply does not prove that a poll was rejected: its eventual completion can still advance consumer progress. Returning an empty success hid that uncertainty. Return ShardCommunicationError when the owner reply is unavailable, clarify acceptance and delivery contracts, and cover timeout followed by an undeliverable rejection in a regression test.
| ReadPolledMessagesError::Fallback(empty_poll_fallback(partition_id)) | ||
| }), | ||
| Some(PartitionReadReply::Rejected(error)) => Err(ReadPolledMessagesError::Rejected(error)), | ||
| None => { |
There was a problem hiding this comment.
Maps every None from partition_read to ShardCommunicationError (11001). Three of those cases prove the read never started: core/shard/src/lib.rs:1998 unroutable namespace, core/shard/src/lib.rs:2012 no sender, core/shard/src/lib.rs:2013 full inbox. Base answered an empty poll here.
There was a problem hiding this comment.
Good point. partition_read already combined submission failures and unavailable replies into None before this PR. The binary poll handler previously converted all those cases into a successful empty poll. I changed that to ShardCommunicationError because a timeout can leave the owner running and advancing progress, making empty success misleading.
However, that mapping is too broad for the cases you identified. If routing fails, the sender is missing, or the initial inbox rejects the request, we know this poll never reached the owner.
I propose returning TransientNotAccepted for those known submission failures, allowing the existing SDK retry logic to handle them. Timeouts and lost replies after submission would retain ShardCommunicationError, since acceptance remains uncertain.
Does that distinction make sense to you, or would you prefer a different error mapping or approach?
There was a problem hiding this comment.
Yes, split it that way. TransientNotAccepted for the three that prove the read
never started, ShardCommunicationError only once submission succeeded.
One addition. Do not just report the uncertainty, shrink it. Probe
crossfire::Sender::is_disconnected() on the reply sender in complete_poll
before admit_poll_auto_commit. A timed-out caller is already gone when the read
lands, so the completion rejects and nothing advances. ShardCommunicationError
then means a rare lost reply instead of the common case.
| ) { | ||
| let config = repair_config(); | ||
| let (_directory, mut partition) = Box::pin(disk_poll_partition(&config)).await; | ||
| let consumer = PollingConsumer::Consumer(7, 0); |
There was a problem hiding this comment.
No-auto-commit purge test uses PollingConsumer::Consumer(7, 0). On base an individual consumer with auto_commit=false updates neither consumer_offsets nor last_polled_offsets, so every assertion passes without the fix.
There was a problem hiding this comment.
Thanks, that makes sense. With an individual consumer and auto_commit=false, the old implementation did not update consumer progress, so the unchanged-offset and Next assertions do not demonstrate that the fix prevented a stale update.
The test still checks that the old result is rejected with TransientNotAccepted. There is also a separate group test using a constructed result, but we are missing a group test that exercises a real disk read across purge.
I’ll add that case with automatic commits disabled and ensure the delayed read produces a nonempty result. After purge and rejection of the old completion, it should verify that both the group’s last polled and committed offsets remain unset, before any fresh group read changes them. This would catch the old behavior where a delayed read restores a stale last polled mark after purge, potentially delaying partition handoff.
There was a problem hiding this comment.
Done in 70268ab7b. The new test covers a real group disk read across purge with automatic commits disabled. It checks that the delayed result is nonempty and that rejecting it leaves both last_polled and the committed offset unset before any fresh read.
| #[cfg(feature = "poll-diagnostics")] | ||
| queued_at: Some(std::time::Instant::now()), | ||
| }))); | ||
| if let Err(error) = inbox.try_send(frame) { |
There was a problem hiding this comment.
PollCompletionSender::complete calls try_send on the main lane. A full inbox discards a finished read and answers TransientNotAccepted.
There was a problem hiding this comment.
Agreed. Once the disk read finishes, the complete method on PollCompletionSender tries to enqueue the result in the owner’s main inbox. If that inbox is full, it drops the completed result and attempts a TransientNotAccepted reply. This is deliberate behavior, and the existing test explicitly verifies it.
The rejection preserves progress safety. The owner has not accepted this completion, so this poll has neither advanced the consumer cursor nor admitted an automatic commit. TransientNotAccepted is therefore appropriate even though the disk read already happened. The concern is wasted work. Under sustained pressure, retries can repeatedly read and discard the same messages, adding load.
A related failure path existed before this PR. Polls that needed to submit an automatic commit could already reject after reading their messages if the owner’s inbox refused that submission. However, polls without automatic commits, or whose commits needed no submission, could reply directly. This PR makes every disk poll depend on completion inbox capacity. Resident polls still complete inline.
We need to retain owner validation before allowing progress or a successful reply, since purge or partition replacement can happen while disk I/O is pending. I see two possible improvements.
- Wait asynchronously for inbox capacity from the detached worker. This retains the completed batch instead of immediately discarding it. The current sender’s ordinary
sendblocks, so using it could deadlock the shard thread that must drain the inbox. We would need an asynchronous sender, with waiting outside the pump. This also requires a bound on outstanding reads or retained bytes. A bounded inbox alone does not bound waiting workers. Completions would still compete with other main inbox traffic. - Reserve dedicated completion capacity before starting disk I/O. Each admitted read would then have space to return its result. If capacity is exhausted, we could defer or reject before doing the read. This addresses wasted I/O and competition with other traffic more directly, but introduces additional admission and scheduling machinery, including fair service for completions.
Either approach needs explicit timeout and shutdown handling, because the requester’s timeout currently does not cancel detached work. History validation must still happen when the owner accepts the result.
The current policy favors immediate rejection and releasing memory. Would you prefer retaining completed reads with bounded asynchronous waiting, reserving completion capacity before reading, or another approach?
There was a problem hiding this comment.
Do not wait on the inbox. A worker waiting for capacity keeps the batch and the read handle for as long as the pressure lasts, so memory and descriptors grow without bound.
Size the completion lane to the maximum in-flight disk polls instead, which the read budget already caps, or reserve the slot when the poll is dispatched. Then try_send cannot fail for capacity. Leave the unroutable and disconnected arms alone.
There was a problem hiding this comment.
I tested the option that reserves capacity when the disk poll is dispatched.
Each pending read owns one slot until its result is dequeued. Completed reads
use a dedicated bounded lane, and the owner processes one completion after
ordinary work so both lanes keep making progress. A read is rejected before
disk I/O if no slot is available. Finished reads do not wait for inbox space.
The missing and disconnected route checks remain, as does owner history
validation.
The new regression fills the ordinary inbox, then returns two nonempty reads.
It verifies delivery and observes an ordinary request between their acceptances.
It fails if completion service is removed or if the entire completion queue
is drained at once. Additional tests cover reservation release and shutdown, plus the behavior
after caller cancellation at a699ff9. In total, 103 shard tests and 143
simulator tests passed on macOS and Linux.
This experiment used a699ff9, before 7cdf64c added the disconnected receiver
check. It preserves the earlier timeout characterization. Adopting the lane
would require rebasing it and updating that regression for the new policy.
The performance comparison did not show an improvement, and the point estimates
were worse with a concurrent producer. The implementation remains an experimental patch. The performance reply and the evidence bundle contain the measurements and
exact patch.
| reply, | ||
| self.metrics.clone(), | ||
| ); | ||
| self.bus.spawn(read_poll(namespace, plan, completion)); |
There was a problem hiding this comment.
Adds one main-lane frame, one pump turn and one heap allocation per disk poll. The completion queues behind arbitrary consensus and write frames. This is the measured disk-offset regression.
There was a problem hiding this comment.
Thanks, I agree this extra step deserves attention. In disk-offset, the client tracks its next offset itself and automatic commits are disabled. The old worker could therefore reply directly after reading. With this PR, it allocates a completion and waits for the owner to process it before replying, so there is additional overhead in this case.
The measured slowdown is consistent with that extra work. Since the results describe an earlier revision, I’d like to rerun the comparison on the current code and check how much time goes into allocation and waiting for the owner. This case has no concurrent producer, so testing with concurrent writes would also help us understand the queue contention you mentioned.
The completion queue options discussed in the other thread seem worth exploring. We can preserve the owner’s history validation while looking for a cheaper way to return completed reads, then use the measurements to guide the choice.
There was a problem hiding this comment.
Re-measure before merge, with a concurrent producer. Give writes per poll next to latency. The completion adds a frame and an allocation, no I/O, so if the regression survives it is queueing behind consensus and write frames, which the reserved lane in the other thread also fixes.
There was a problem hiding this comment.
I completed the remeasurement and a separate experiment with reserved completion
capacity. The reserved lane passed its scoped correctness checks on a699ff9, but it
did not show a performance benefit, so it remains an experimental patch.
The first comparison used a699ff9 against
dc2b382, the master revision already merged
into the PR. Throughput was 4.01% lower without concurrent writes, with a 95%
interval of 1.94% to 6.26% lower, and 5.41% lower with writes, with an interval
of 2.59% to 8.41% lower.
Both campaigns predate 7cdf64c, which adds the
disconnected receiver check. These results do not establish the latest PR
revision's performance.
The second comparison used a699ff9 as the baseline and added a reserved
completion lane. It keeps the existing allocation and owner history validation.
Throughput changed by +0.64% without writes, interval −0.64% to +2.40%, and
−3.42% with writes, interval −7.67% to approximately zero. With writes, p99
latency changed by +13.86%, interval +0.31% to +29.94%.
For the concurrent producer cases, medians across runs were:
| Experiment and version | Producer batches per poll | Written messages per poll | Median poll latency | p99 poll latency |
|---|---|---|---|---|
| First comparison, master | 0.586 | 58.60 | 268.5 µs | 575.5 µs |
| First comparison, PR | 0.613 | 61.28 | 278.5 µs | 603.5 µs |
| Lane experiment, PR | 0.564 | 56.37 | 271.0 µs | 507.5 µs |
| Lane experiment, reserved lane | 0.574 | 57.40 | 274.0 µs | 574.0 µs |
The producer target is 50 MB/s in batches of 100 messages. These are logical
producer operations, not disk syscalls. The two experiments ran at different
times, so their absolute values should not be compared across experiments.
Each case has ten pairs of 33,000 explicit offset polls, with five pairs in each
order, one shard and one sequential TCP consumer. Each poll returns one 256 byte
message; automatic commits are disabled. Percentage changes are geometric means
of paired ratios, and the intervals resample whole pairs.
Each experiment also has four baseline repeatability pairs per case. All
1,848,000 clean polls in each experiment passed the accounting audit.
This is a Linux VM on an M1 Pro with disk data that may be cached. One identical
binary pair in the lane experiment differed by 8.54% in throughput. The mixed
throughput interval ends very close to zero, so I would not claim a universal
slowdown from it. It does not support calling this implementation a performance
fix. Separate diagnostics also did not show a handoff improvement.
Reports and evidence bundles contain the source patch, revisions,
build and run commands, configuration, raw observations, diagnostic results,
validation logs, analysis scripts and SHA256 hashes. The evidence is hosted
on my fork; the lane change has not been added to this PR.
| } | ||
| let partitions = self.plane.partitions(); | ||
| let consumer_kind = result.consumer_kind(); | ||
| match partitions.complete_poll(&namespace, result) { |
There was a problem hiding this comment.
Commits progress at core/partitions/src/iggy_partition.rs:2901, then tries the reply at core/shard/src/poll.rs:195 and discards the result. Budget expiry at core/shard/src/lib.rs:2021 drops the receiver while the client still waits. Needs auto_commit=true with Next.
There was a problem hiding this comment.
Agreed. With auto_commit=true, the internal timeout can drop the receiver while the owner still accepts the result and advances the cursor. A subsequent Next can then skip the batch the client never received.
The previous implementation also advanced progress before attempting the reply. On timeout, however, it returned a successful empty poll, even though the detached read could still finish and advance progress. This PR returns ShardCommunicationError instead, so the caller is told that the read failed to produce a reply rather than being given an empty success. That improves the reported outcome, but it does not cancel the read or prevent later progress updates. The new completion queue also adds another delay window.
I’ll add a regression that delays a nonempty completion past the timeout, then checks the cursor and a subsequent Next. What contract should we guarantee here? Should timeout cancel any poll that has not yet been admitted, ensuring it cannot advance progress afterward? Or is eventual admission acceptable if the caller receives an error indicating an unknown outcome? Once admission has happened, a lost reply can still leave the outcome unknown.
There was a problem hiding this comment.
Added the regression in ad9fc761b. It holds a nonempty result past the requester timeout with auto_commit=true, verifies that the reply channel has closed, then delivers the result through the completion sender and owner inbox. The test records the current behavior: local progress advances through offsets 0–2, and a subsequent Next returns only offset 3, skipping the messages the caller never received. This documents the behavior while leaving the timeout and cancellation contract open.
There was a problem hiding this comment.
Cancel it. A poll that cannot deliver its reply must not advance progress.
Duplicate delivery is fine, a silent skip is not, and that is the same
invariant this PR restores for purge.
Eventual admission with an error is not acceptable. The client cannot act on
"unknown outcome" and the skip is silent from its side.
crossfire::Sender::is_disconnected() is on the reply sender. Probe it in
complete_poll before admit_poll_auto_commit and reject when the caller is
gone. That covers the real case, since a timed-out caller is already gone when
the read lands. The remaining race is a few instructions wide and costs a
duplicate, not a skip.
Then flip the new test to assert re-delivery. Pinned as a skip it reads as the
intended contract.
There was a problem hiding this comment.
Agreed. A completion arriving after the requester has timed out should leave progress unchanged. I’ll add the disconnect check before admission and change the regression to verify that the next poll returns the unseen messages.
There is one remaining ordering detail I’d like to account for. The task awaiting the reply and the partition owner can run on different shards. In partition_read, the requesting shard sends the read to the shard selected by the namespace, then waits on its own reply receiver with a timeout. One thread per shard therefore still allows the requester’s timeout to run concurrently with the owner’s completion handler.
With the proposed check, this sequence remains possible:
- The owner checks
is_disconnected()and sees that the receiver is present. - The requesting shard times out and drops the receiver.
- The owner admits the automatic commit and advances progress through offset 2.
- The owner attempts to return messages 0–2, but the reply cannot be delivered.
- The next
Nextpoll starts at offset 3.
The check prevents the case where the requester is already gone when completion is processed. In the sequence above, however, the remaining race still skips unseen messages because progress advances before the reply attempt. The absence of an .await protects the owner’s state from interleaving on its own thread, but it does not prevent another shard from closing the receiver.
To make cancellation and admission mutually exclusive across shards, they need a shared decision: if cancellation wins, the owner must leave progress untouched; if acceptance wins, the requester must preserve the reply path instead of abandoning it when its timer fires. Checking whether the receiver is connected only observes its state at that moment.
I suggest keeping the disconnect check and the redelivery regression, and adding a deterministic test that places timeout between the check and admission. That would let us verify the additional coordination needed for the stated invariant. This concern is specifically about the internal timeout between the requesting shard and the owner.
There was a problem hiding this comment.
I looked into existing concurrency and messaging contracts to make the shared decision mentioned above more concrete.
If we want confirmed cancellation to guarantee that the poll cannot advance progress, the requester and owner need to compete for one explicit state transition:
Pending → Cancelled
Pending → Completing → Completed(result)
An atomic compare-and-exchange could decide which side wins:
- Cancellation wins: the owner discards the result without updating consumer progress, group
last_polled, or admitting an automatic commit. - Completion wins: cancellation can no longer succeed. The owner finishes validation and admission, applies the permitted effects, and publishes the outcome. That outcome can still be a rejection if validation or admission fails.
The partition state would remain exclusively owned by its shard. Only the request’s cancellation decision needs to be shared. This follows an established executor pattern: Python’s Future.set_running_or_notify_cancel() arbitrates between cancellation and execution, with cancellation no longer succeeding once execution has won.
The requester’s behavior is essential to that contract. If its timer fires after the owner has claimed completion, it must preserve the receiver and wait for the outcome to maintain the internal handoff guarantee. That may extend the wait beyond the original deadline. Returning an error with an unknown outcome instead would describe the uncertainty, but would not recover the unseen messages or satisfy the no-skip requirement.
I also found that Crossfire explicitly documents that a buffered send can succeed while the receiver concurrently disappears, leaving nobody to receive the message. Therefore, moving try_send() before the effects would not establish delivery either. It could additionally expose success before admission fails. Rolling back after a failed send would require undoing all associated progress and consensus admission, rather than simply restoring one offset.
This distinguishes two atomicity questions. We can make cancellation and permission to apply effects mutually exclusive within the process. Actual receipt by the external client requires a delivery and recovery protocol: a connection can fail after the owner completes but before the client observes the reply. gRPC documents that same ambiguity. Our poll reply also currently acknowledges admission rather than a durable offset commit, so admission, durability, and receipt are separate boundaries.
For reliable consumption across those failures, established messaging systems retain eligibility for redelivery until the consumer acknowledges the required work. When processing must be protected, acknowledgement follows processing; failure before acknowledgement can then produce duplicates. RabbitMQ documents this tradeoff, including the weaker safety of automatic acknowledgement.
Kafka offers a useful comparison, but its automatic commits operate through the consumer client. Its current position and committed recovery position are separate. Obtaining at-least-once behavior with automatic commits requires processing all records returned by poll() before subsequent polls or closing. That differs from the server admitting the current batch’s automatic commit before replying. Kafka consumer documentation
Another possible recovery contract is replayable polling: retrying the same request ID recovers the same batch without advancing again. Here, the effects and the recorded request outcome need to be committed together, following the atomicity requirement described in AWS’s idempotency guidance. For Iggy, this would also require defining batch retention, request expiry, and recovery across owner failure.
Any broader delivery contract must include consumer-group last_polled, which advances even without automatic commits. Disabling auto_commit alone therefore would not cover every path. The shared cancellation decision addresses the internal timeout; acknowledgements or replayable requests provide recovery when the external reply is lost.
There was a problem hiding this comment.
Added the disconnect guard in 7cdf64cfa. on_poll_completed now discards the result when the reply receiver is already disconnected, before complete_poll can admit an automatic commit or change progress. The same commit updates the timeout regression to require redelivery. The race between the connection check and admission remains as discussed above.
| /// `Send`, as required by the shard inbox. | ||
| /// `Default` creates a fresh identity. Only clones compare equal. | ||
| #[derive(Clone, Debug, Default)] | ||
| pub struct PollHistoryId(Arc<()>); |
There was a problem hiding this comment.
PollHistoryId is an Arc<()> compared by address where a u64 counter serves. Zero lines saved, one allocation removed per poll.
There was a problem hiding this comment.
Thanks, I think a numeric ID is worth trying here. The current allocation is created when a history is created or invalidated, and polls clone the existing Arc without allocating a new history object. Switching to a number would still remove the reference count updates from each poll.
We can keep the implementation small by using an AtomicU64 shared by the process to assign fresh history IDs, with a checked update that prevents wrapping and reuse. The counter would run only when the history changes. Each poll would carry a plain u64, so it would not access the atomic counter. This also preserves uniqueness when a partition is recreated with the same namespace.
I’ll retain the tests for partition replacement and stale results, and include the change in the planned performance comparison. It removes some polling overhead, although we should measure its effect separately from the completion allocation and queue processing.
There was a problem hiding this comment.
Do it here, not in a follow-up. The point is the invariant, not the allocation. Arc::ptr_eq only works because pending reads keep the old allocation alive so the address cannot be reused. A u64 counter cannot alias at all, and it is Send, so the "keeps frames Send" note does not argue for Arc either.
| "partition read reply unavailable; acceptance unknown" | ||
| ); | ||
| Err(ReadPolledMessagesError::Rejected( | ||
| IggyError::ShardCommunicationError, |
There was a problem hiding this comment.
Binary transports answer 11001 while HTTP answers 504 at core/server/src/http/handlers.rs:1322. The parity rule is stated at core/server/src/http/error.rs:474.
There was a problem hiding this comment.
Thanks for pointing this out. HTTP already returned 504 when the partition owner did not reply before this PR, while binary transports returned a successful empty poll. This PR changes the binary response to ShardCommunicationError with code 11001, so both now report a failure, but their error identities remain inconsistent.
The numbers represent different things. 11001 is an Iggy error code, whereas 504 is an HTTP status. However, the HTTP response also uses 504 as its JSON error ID, with partition_read_timeout as its code. We could preserve an appropriate HTTP status while exposing the same Iggy error identity in the body.
I suggest aligning this with the distinction discussed in the other thread. When we know the request was never submitted, both transports could return TransientNotAccepted, which already maps to HTTP 503. When submission succeeded but no reply arrived, we should preserve the unknown outcome. For an actual timeout, HTTP could retain 504 while including error ID 11001, without implying that the original poll had no effect.
This would need an explicit HTTP mapping. Passing ShardCommunicationError through the current generic handler would otherwise produce 400 Bad Request, which would incorrectly attribute the failure to the client.
Does that approach match the contract you have in mind, or would you prefer a different mapping?
There was a problem hiding this comment.
Yes. Keep 504 as the status and carry 11001 as the body error id. Add the
explicit mapping, the generic handler would give 400 and blame the client.
| .build_poll_snapshot(&namespace, consumer, &args) | ||
| .expect("poll snapshot"); | ||
| assert!( | ||
| !plan.needs_off_pump_io(), |
There was a problem hiding this comment.
The one new simulator test asserts !plan.needs_off_pump_io(), and no disk tier exists. The new inbox hop is never modeled.
There was a problem hiding this comment.
Thanks, you’re right. That test deliberately uses a resident read to verify that the client receives its reply while replication is stalled. It does not exercise the extra inbox step used by disk completions.
The existing disk purge test calls the completion handler directly, while the channel tests check enqueueing and rejection without processing the completion through the owner. We are missing coverage of those pieces working together.
I’ll add a test that holds back a nonempty result, purges or replaces the partition, and then delivers the old result through the actual completion sender and inbox. The owner should reject it without advancing progress. A fresh result should also succeed through the same route.
We can control when the result is released without building a complete disk model into the simulator, while keeping the existing disk test for actual I/O coverage.
There was a problem hiding this comment.
Added in a699ff91c. The new regression holds a group read result, replaces the partition with different messages at the same offsets, and delivers the old result through the actual completion sender, inbox, router, and owner pump. It checks TransientNotAccepted and both group offsets remaining unset before a fresh result succeeds through the same route. In an isolated copy, disabling the history check made the test fail because the stale result restored last_polled to Some(2), confirming the delayed result is nonempty and the regression catches stale progress. All 99 shard tests and strict Clippy passed on both macOS and Linux.
| if let Some(replication) = completion.replication { | ||
| // SimOutbox sends enqueue immediately, so replication cannot suspend. | ||
| futures::executor::block_on( | ||
| partitions.replicate_poll_completion(&namespace, replication), |
There was a problem hiding this comment.
Comment states replica sends never suspend, but core/simulator/src/bus.rs:234 awaits under cfg(test), which deadlocks the block_on at core/simulator/src/lib.rs:1415.
There was a problem hiding this comment.
Thanks for catching this. The comment is too strong. Replica sends normally finish immediately, but the new test mechanism can pause them, so the helper cannot safely assume that replication will never suspend.
Before this PR, the helper used block_on for the resident read but did not replicate automatic commits. This PR adds the replication wait alongside the ability to pause a replica send. If a test calls the helper while that pause is active, it can block before reaching the instruction that releases the send.
The existing test avoids this by using the asynchronous path, but the helper should handle that situation safely too. I’ll adjust it to use the simulator’s normal asynchronous execution path so the calling test can continue and release paused work, and add regression coverage for this combination.
There was a problem hiding this comment.
Fixed in 13afb497a. Simulator::poll_messages now returns an owned future and routes the poll through the partition owner, removing both block_on calls. The regression uses the helper with a replica send paused, checks that the nonempty reply is available before release, and then verifies that the automatic commit reaches every replica after replication resumes. The simulator demo also uses the asynchronous helper. All 143 simulator tests passed on both macOS and Linux, with five existing ignored tests on each platform.
Reads rejected before reaching the partition owner must remain safe for SDK retries. Return TransientNotAccepted for routing and inbox submission failures while preserving ShardCommunicationError for missing replies after submission. Merge current master, preserve both sets of regression tests, and cover refusal paths, lost replies, and consumer group ownership.
Groups record their last polled offset even when automatic commits are disabled. Keep a real disk read pending across purge and verify that its stale completion cannot restore group progress after new messages reuse the old offsets. Assert nonempty old results and both unset group offsets before a fresh read, then check that a fresh group completion succeeds without an automatic commit.
Explain the scenarios and state transitions so readers can follow the regression coverage without tracing low level helpers. Preserve scheduling, fixture data and assertions while naming the roles of requests, results and progress values.
Blocking on automatic commit replication could deadlock a test with a paused replica send. Return an owned poll future and let the owner pump accept the read and continue replication. Exercise the helper with replication paused, then verify its commit reaches every replica after release. Update the simulator demo to drive the future and register its client before sending requests.
Exercise history validation after a completed group read passes through the real completion sender, inbox and owner pump. Replace the partition with new data at the same offsets and verify stale rejection preserves both group offsets before a fresh completion succeeds through the same route. Share resident message setup with the timeout test. Disabling the history check in an isolated copy makes the regression fail on stale last-polled progress.
| assert_eq!( | ||
| message_offsets(&fragments), | ||
| vec![3], | ||
| "Next skips offsets 0 through 2, which the caller never received" |
There was a problem hiding this comment.
Contract: a poll that cannot deliver its reply must not advance progress. Duplicate beats skip, and that is the invariant this PR restores for purge.
crossfire::Sender::is_disconnected() is on the reply sender. Probe it in complete_poll before admit_poll_auto_commit. A timed-out caller is already gone when the read lands, so the completion rejects and the next poll re-reads. The race window drops from the whole disk read to a few instructions, and losing it costs a duplicate.
Then flip this assertion to re-delivery. Pinned as-is, the next reader takes the skip for the intended contract.
There was a problem hiding this comment.
Updated in 7cdf64cfa. The regression now requires an unchanged cursor, no automatic commit queued or assigned, and offsets [0, 1, 2, 3] on the next poll after the original requester times out. It fails without the disconnect guard and passes with it. This covers timeout before owner completion; the concurrent cancellation race is discussed in the completion thread.
| "partition read reply unavailable; acceptance unknown" | ||
| ); | ||
| Err(ReadPolledMessagesError::Rejected( | ||
| IggyError::ShardCommunicationError, |
There was a problem hiding this comment.
Yes. Keep 504 as the status and carry 11001 as the body error id. Add the
explicit mapping, the generic handler would give 400 and blame the client.
| tier = "disk", | ||
| "partition poll dispatch" | ||
| ); | ||
| let completion = completion::PollCompletionSender::new( |
There was a problem hiding this comment.
Needs the re-measurement before merge, with a concurrent producer. This hop is
on every disk poll and the posted numbers describe an earlier revision.
Report writes per poll next to latency. The completion adds a frame and an
allocation, no I/O, so a surviving regression is queueing behind consensus and
write frames, which the reserved lane also fixes.
| ReadPolledMessagesError::Fallback(empty_poll_fallback(partition_id)) | ||
| }), | ||
| Some(PartitionReadReply::Rejected(error)) => Err(ReadPolledMessagesError::Rejected(error)), | ||
| None => { |
There was a problem hiding this comment.
Yes, split it that way. TransientNotAccepted for the three that prove the read
never started, ShardCommunicationError only once submission succeeded.
One addition. Do not just report the uncertainty, shrink it. Probe
crossfire::Sender::is_disconnected() on the reply sender in complete_poll
before admit_poll_auto_commit. A timed-out caller is already gone when the read
lands, so the completion rejects and nothing advances. ShardCommunicationError
then means a rare lost reply instead of the common case.
| } | ||
| let partitions = self.plane.partitions(); | ||
| let consumer_kind = result.consumer_kind(); | ||
| match partitions.complete_poll(&namespace, result) { |
There was a problem hiding this comment.
Cancel it. A poll that cannot deliver its reply must not advance progress.
Duplicate delivery is fine, a silent skip is not, and that is the same
invariant this PR restores for purge.
Eventual admission with an error is not acceptable. The client cannot act on
"unknown outcome" and the skip is silent from its side.
crossfire::Sender::is_disconnected() is on the reply sender. Probe it in
complete_poll before admit_poll_auto_commit and reject when the caller is
gone. That covers the real case, since a timed-out caller is already gone when
the read lands. The remaining race is a few instructions wide and costs a
duplicate, not a skip.
Then flip the new test to assert re-delivery. Pinned as a skip it reads as the
intended contract.
| reply, | ||
| self.metrics.clone(), | ||
| ); | ||
| self.bus.spawn(read_poll(namespace, plan, completion)); |
There was a problem hiding this comment.
Re-measure before merge, with a concurrent producer. Give writes per poll next to latency. The completion adds a frame and an allocation, no I/O, so if the regression survives it is queueing behind consensus and write frames, which the reserved lane in the other thread also fixes.
| &mut self, | ||
| result: PollReadResult, | ||
| ) -> Result<PollCompletion, IggyError> { | ||
| self.resynchronize_consumer_offset_reservations(); |
There was a problem hiding this comment.
Move this below the history guard. Every rejected stale completion pays the
reconciliation before being thrown away.
| #[cfg(feature = "poll-diagnostics")] | ||
| queued_at: Some(std::time::Instant::now()), | ||
| }))); | ||
| if let Err(error) = inbox.try_send(frame) { |
There was a problem hiding this comment.
Do not wait on the inbox. A worker waiting for capacity keeps the batch and the read handle for as long as the pressure lasts, so memory and descriptors grow without bound.
Size the completion lane to the maximum in-flight disk polls instead, which the read budget already caps, or reserve the slot when the poll is dispatched. Then try_send cannot fail for capacity. Leave the unroutable and disconnected arms alone.
| /// `Send`, as required by the shard inbox. | ||
| /// `Default` creates a fresh identity. Only clones compare equal. | ||
| #[derive(Clone, Debug, Default)] | ||
| pub struct PollHistoryId(Arc<()>); |
There was a problem hiding this comment.
Do it here, not in a follow-up. The point is the invariant, not the allocation. Arc::ptr_eq only works because pending reads keep the old allocation alive so the address cannot be reused. A u64 counter cannot alias at all, and it is Send, so the "keeps frames Send" note does not argue for Arc either.
github-actions
Bot
added
S-waiting-on-author
A completion arriving after its caller times out could advance progress and make the next poll skip messages the caller never received. Check the reply connection before owner admission. Update the timeout regression to require unchanged progress, no queued or assigned automatic commit, and redelivery of the unseen messages. Document that cancellation racing with the check remains an unknown outcome. Validated with all 99 shard tests, strict shard Clippy, formatting checks, and the shard documentation build.
3 participants
Fixes #4117.
Problem
A disk poll can remain pending while purge replaces a partition's message history and resets its offsets to zero. The existing
PollPlancarries shared handles that allow its detached worker to update consumer offsets and grouplast_polledstate after the read finishes. A result created before purge can therefore update the new history, even when its numeric offset is valid again because new messages now occupy the same offsets.For example, an old poll for offsets 0 through 2 can complete after purge and record offset 2. If five new messages have since been appended at offsets 0 through 4, the consumer's next
Nextpoll starts at offset 3 and skips the first three new messages.Implementation
PollPlannow contains read resources and immutable request context. Its worker returns an ownedPollReadResultand cannot mutate consumer progress.PollHistoryIdchanges before purge, state installation, partition replacement, and other paths that replace message history. Results from an older history are rejected withTransientNotAcceptedbefore progress changes or a reply is authorized.last_polledupdates now occur behind the partition completion API. Replication remains asynchronous, so a successful poll reply does not wait for the durable commit.The benchmark tooling is reviewed separately in #4122. This PR contains the poll completion fix and its regression tests.
Correctness verification
After extracting the benchmark tooling, 1,239 tests across the affected crates pass on macOS. The final code tree is identical to the previous PR head with only the benchmark changes removed.
The baseline reproduces the delayed disk poll failure after purge, including a new history in which the stale offset is numerically valid.
The candidate passes the affected crate suites on Linux with 791 tests and ten targeted cluster, purge, and reconnection integration tests.
Native server, shard, and simulator verification passes 620 tests, including a successful reply becoming available before deliberately stalled replication completes.
Formatting, dependency sorting with
--no-format, changed TOML formatting, Markdown lint, whitespace checks, and scoped newline checks pass.Clippy with all features, all targets, and warnings denied passes for the affected crates on Linux. On macOS, the same check reaches existing
unused_selfandunnecessary_wrapsfindings in unchangedcore/server/src/shard_allocator.rs; allowing those two categories passes.Performance
These measurements were collected before the subsequent admission cleanup and benchmark extraction. They describe the earlier candidate and are retained as historical evidence. No new performance comparison was run for this split.
The recorded comparison used the same low level benchmark client for the baseline at
b692160e4and the earlier candidate. Each case has ten paired runs, with five pairs run baseline first and five run candidate first. Three A/A pairs per case provide a small noise diagnostic. All runs use TCP withTCP_NODELAYand 256 byte messages. No measured empty polls, errors, timeouts, or cancellations were recorded.The values below are geometric means of the paired candidate to baseline ratios, expressed as percentage changes. Positive throughput favors the candidate; positive p99 means higher latency. Parentheses contain the lower and upper one sided 95% bootstrap bounds, which together form a central 90% interval.
disk-offsetdisk-nextdisk-batchdisk-groupdisk-writesdisk-fsyncThe most repeatable cost is the
disk-offsetcase. Median throughput changes from 3,609.0 to 3,461.6 polls per second in the Linux VM and from 5,515.1 to 5,016.2 on macOS. Median p99 changes from 515 to 557 microseconds on Linux and from 293 to 315 microseconds on macOS. Estimated CPU cost per completed poll rises by 5.63% on Linux and 4.60% on macOS.The other cases have wider uncertainty. In particular, the resident, batched, group, and fsync measurements do not establish either a consistent gain or unchanged performance.
Environment and limits
The Linux measurements ran in a native ARM64 container inside Docker Desktop's Linux virtual machine. The container used CPUs 0 through 3 and an 8 GiB memory limit, with the server on CPU 0 and clients on CPUs 1 and 2. Data lived in a Docker volume. The server used its required
io_uringbackend through a seccomp profile that permits the threeio_uringsystem calls.The macOS measurements ran on an Apple M1 Pro with 32 GiB RAM and the native I/O backend, without CPU affinity. Both versions used the repository's Rust 1.98.0 toolchain and identical release settings within each platform.
These results describe one sequential consumer against one shard. Group cases use one group consumer. Reads through the disk path may be served from the operating system file cache. The Linux result describes the Docker Desktop ARM64 virtual machine rather than native Linux hardware. The three A/A pairs expose noise but are too few to characterize it precisely, especially in the cases with background writes and fsync.
The reusable commands and analysis method are documented in
scripts/benchmarks/README.mdin #4122.