Conversation
Convert shard_source_descs and shard_source_fetch from AsyncOperatorBuilder to synchronous OperatorBuilderRc operators paired with tokio tasks that own the persist I/O (reader/snapshot/listen and batch fetching). This drops the persist source's dependence on the timely async bridge with no behavior change. * shard_source_descs runs a listen task on the chosen worker that sends parts (split into ExchangeableBatchPart + Lease) and progress over a channel; the operator downgrades capabilities and parks leases. The former shard_source_descs_return operator is merged in as a disconnected completed_fetches input, and the listen handle (the reader's SeqNo hold) is released via a oneshot once that frontier empties. * shard_source_fetch forwards descs to a fetch task and retains a per-flight capability pair; results are matched FIFO, emitted at the data capability, with the completed-fetches capability dropped to release the lease. On a missing blob it reports through the ErrorHandler and freezes, retaining capabilities (and crucially ceasing to drain results, so a later good result cannot advance the frontier past the missing part). * Both operators reproduce builder_async's two-phase shutdown via build_reschedule + the coordinated button, so a local-only press cannot advance the downstream frontier past times other workers still feed. Adds ErrorHandler::report_and_freeze, module documentation of the consumer contract and the operator/task architecture, and regression tests for end-to-end fetch, mid-stream shutdown, listing-path error freeze, and fetch-path error freeze. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
With the persist source's listen polling moved into a tokio task, a fully caught-up DataSubscribe with no listen retry timer produces no worker activations. The stress test's read workers parked indefinitely on step_or_park(None) while polling a oneshot, so they never observed the shutdown signal. Bound the parks so they re-check between steps. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
The lease-return input now lives on the shard_source_descs operator, so introspection names that operator instead. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
The fetch operator matched results to capabilities by arrival order: it pushed a capability pair per desc into a VecDeque and popped the front for each result. That was sound only because the fetch task is a strictly sequential loop, and it would silently misassign capabilities the moment the task fetched concurrently — releasing a part's completed-fetches capability (and its lease) for a part not actually fetched. Tag each desc with the time it was minted at, echo that time back with the result, and track outstanding fetches in a per-time BTreeMap of (data cap, completed cap, count). Emit each blob at its time's capability and drop both capabilities when a time's count reaches zero. This is independent of the order results return in, and it aligns with the descs-side LeaseManager, which is also keyed by time. As a bonus the freeze-on-error is now structural: the failed time's capability is never decremented, so the frontier holds there regardless of what else drains. Per review feedback on MaterializeInc#36910. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
While catching up to the shard upper observed at hydration time, the persist source withholds per-batch progress and forwards it in larger, byte-bounded steps. This turns a long historical replay from one progress round (and one downstream arrangement-maintenance pass) per persist batch into a handful of steps. Emitted parts keep their real timestamps, so only the frontier granularity changes, not the data. Gated by persist_source_hydration_frontier_coalesce_bytes (default 0 = disabled); tracking snaps back to per-batch once the source is live. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
Extract the forward/coalesce decision into `should_forward_progress` and cover it two ways. `test_frontier_coalesce_decision` checks the round reduction directly (disabled forwards once per batch; a budget larger than the replay forwards once), independent of timely's progress batching. `test_shard_source_hydration_frontier_coalesce` hydrates an index over 64 unmerged single-timestamp batches (compaction disabled, mirroring a held-back since) and asserts that coalescing on and off both emit parts through the final timestamp without stalling. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
antiguru
force-pushed
the
persist-source-deasync-design
branch
from
June 19, 2026 13:57
ffa9df5 to
b55c0d2
Compare
antiguru
force-pushed
the
persist-source-deasync-design
branch
from
July 20, 2026 14:51
b55c0d2 to
51e5ca4
Compare
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
1 participant
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.
Motivation
Hydrating an index with retained history replays the whole
since..upperwindow, and the persist source downgrades its output frontier once per persist batch.With production retention (days at ~1 write/s, batches kept unmerged by the held-back
since), that is hundreds of thousands of progress rounds, and each round drives a full per-arrangementread_upper/ maintenance pass on the replica.Stacked on MaterializeInc#36910.
Description
While catching up to the shard upper observed at hydration, the source withholds per-batch progress and forwards it in byte-bounded steps, collapsing the replay into a handful of frontier rounds.
Emitted parts keep their real timestamps, so only frontier granularity changes, not the data.
Gated by
persist_source_hydration_frontier_coalesce_bytes(default0= disabled); tracking snaps back to per-batch once the source reaches the hydration-time upper, so consumers that rely on tight frontier tracking such aspersist_sinkare unaffected in steady state.Verification
Unit tests for the round reduction (independent of timely progress batching) and for hydration correctness with coalescing on and off.