shuffle: gate the startup heap drain on a Barrier from the Session - #3456
Open
williamhbaker wants to merge 4 commits into
Open
shuffle: gate the startup heap drain on a Barrier from the Session#3456williamhbaker wants to merge 4 commits into
williamhbaker wants to merge 4 commits into
Conversation
williamhbaker
force-pushed
the
wb/shuffle-startup-drain-gate
branch
2 times, most recently
from
September 2, 2026 14:40
16cfa4c to
a790eb1
Compare
ListingSnapshotComplete reports that a binding's initial journal-listing snapshot is fully delivered. InitialReadsStarted is an empty SliceRequest whose stream position marks the end of the initial StartReads. Regenerate the checked-in Go and Rust bindings here so generated changes remain separate from the implementation.
Store SliceRequest values in the cross-shard FIFO instead of wrapping StartRead values at send time. This preserves FIFO behavior and lets InitialReadsStarted follow all previously queued StartReads.
A Slice can see all known reads reach their write heads before the Session has routed every journal found during initial listing. It could then emit a document before an undispatched read with an earlier clock or higher binding priority. Each binding's listing owner now sends ListingSnapshotComplete after every ListingAdded in its initial snapshot, including an empty snapshot. Once every binding reports, the Session queues one InitialReadsStarted per Slice behind the initial StartReads. A Slice opens its drain gate only after consuming it. StartReads queued after InitialReadsStarted still close the gate through pending_probes until their write-head probes finish.
Add initial_reads_started beside the other heap-drain inputs so traces distinguish a missing InitialReadsStarted from an idle Slice.
williamhbaker
force-pushed
the
wb/shuffle-startup-drain-gate
branch
from
September 2, 2026 15:54
a790eb1 to
6d555df
Compare
jgraettinger
approved these changes
Sep 3, 2026
| /// FIFO queue of per-shard requests, with InitialReadsStarted after StartReads. | ||
| pub slice_requests: std::collections::VecDeque<(usize, shuffle::SliceRequest)>, | ||
| /// Whether each binding's initial listing snapshot has completed. | ||
| pub listing_snapshot_complete: Vec<bool>, |
Member
There was a problem hiding this comment.
nit: I'm not sure this is worth the per-binding bool, as opposed to just keeping a counter of completed or remaining ones (it's just used for finer-grain error reporting of what can fundamentally only be an implementation error, right?). We can still detect if we see more than num_bindings.
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
3 participants
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Description:
At startup, a Slice could drain a document too early. The Session had not yet routed all of the journals from the initial listing. One of these journals could hold a document with an earlier adjusted clock or a higher binding priority. The
Slice must drain that document first.
Each binding has one listing owner. The listing owner now sends a
ListingSnapshotCompletemessage for its binding. The Session waits for one of these messages from each binding. Then the Session puts oneInitialReadsStartedmessage in the request queue for each Slice. This message comes after the initialStartReadmessages. A Slice keeps its drain gate closed until theInitialReadsStartedmessage arrives.See the commit messages for the details.
Closes #3457
Workflow steps:
(How does one use this feature, and how has it changed)
Documentation links affected:
(list any documentation links that you created, or existing ones that you've identified as needing updates, along with a brief description)
Notes for reviewers:
Tested against a local stack, on a derivation whose transforms have descending priorities, so a premature drain shows up as a priority inversion. With an 8s delay injected into the top-priority binding's listing, the drain order is
[alpha, beta, gamma]with the Barrier and[beta, gamma, alpha]without it. Also checked: bindings with no journals still sync (empty partition selector, and a collection never written to), broker outage during startup, crash recovery, and journals that appear after the Barrier.