Repository navigation
Conversation
5eecaa4 to
e6d27f4
Compare
cf82e1f to
77e275a
Compare
| pub(crate) fn role(&self) -> ComputeRuntimeRole { | ||
| self.role | ||
| } | ||
|
|
There was a problem hiding this comment.
Function moved for no particular reason.
There was a problem hiding this comment.
Agreed, no reason. Moved back to where #38389 introduced it, right after new(). This was churn imported by taking compute_state.rs wholesale from the original branch. (Posted by Claude Code.)
e6d27f4 to
a985417
Compare
6984b18 to
f359f28
Compare
3cc5402 to
15cc33c
Compare
f359f28 to
f45dc4c
Compare
15cc33c to
9311614
Compare
f45dc4c to
40e6744
Compare
9311614 to
eb80f6a
Compare
40e6744 to
0ba0d59
Compare
eb80f6a to
3767df8
Compare
0ba0d59 to
2db1bb4
Compare
3767df8 to
9fcfab0
Compare
2db1bb4 to
76b96ab
Compare
9fcfab0 to
c929c6d
Compare
76b96ab to
b288248
Compare
c929c6d to
8f441a1
Compare
b288248 to
f29b660
Compare
8f441a1 to
67eb09f
Compare
f29b660 to
bb7c5fc
Compare
| || (dataflow.is_transient() | ||
| && !dataflow.until.is_empty() | ||
| && dataflow.subscribe_ids().next().is_none() | ||
| && dataflow.copy_to_ids().next().is_none()), |
There was a problem hiding this comment.
I wonder if this should be a function on the dataflow description. We're repeating the condition from the multiplexer.
There was a problem hiding this comment.
Agreed, and it is now DataflowDescription::is_peek_dataflow() in mz-compute-types, next to is_single_time(). #38388's router and this tripwire ask the same method of the same description, so the assert can finally check what it claims to: previously a drift between the two spellings was exactly what it was meant to catch, and it was written as a second copy of the thing that might drift.
The predicate also got stricter on the way: is_single_time() in place of !until.is_empty(), per the discussion on the router. Details in the reply on #38388.
Posted by Claude Code.
| /// | ||
| /// `fuel` bounds how far the walk may go on this worker, exactly as it bounds a walk over a | ||
| /// maintained trace. | ||
| fn seek_fulfillment( |
There was a problem hiding this comment.
This looks like duplication with the regular index peek logic, is that right? The peek offload work aimed at reducing duplication, and canonicalizing all peek structure (which it not has done completely, but that's a different story.) So my ask here would be to check if we can combine the walk, maybe by making them generic over the type of index they're walking.
There was a problem hiding this comment.
Right that it was duplicated, though not in the walk. The walk was already shared: both build a PeekScan and hand it to the generic walk_scan, which is what the peek-offload canonicalisation bought. What was copied is the ~25 lines of frontier gating in front of it, and the only differences were cosmetic: where (oks, errs) come from, and TraceBundle::compaction_frontier() versus an explicit antichain_join of the same two frontiers, which is literally what that method computes.
Extracted as gate_peek, generic over both traces via the existing PeekOksTrace/PeekErrsTrace markers, returning a PeekGate of Open/NotReady/Compacted. That required pinning Time = Timestamp on the two marker traits, which they should have carried anyway: both only ever describe traces over mz_repr::Timestamp, and only the cursor GATs were constrained before.
IndexPeek::collect_finished_data is deliberately left alone. It is main's method with main's tests in index_peek_tests.rs, which call it to exercise the walk while bypassing the gate, so folding it in would have churned code this stack has no business touching.
Posted by Claude Code.
| // One is per collection and carries no dataflow identity, so it cannot go stale across a | ||
| // reconnection, and it only ever rises. Clearing it would drop the hold back to the | ||
| // minimum time until the replayed compactions raised it again. See | ||
| // `doc/developer/design/20260720_two_runtime_compute/design.md`. |
There was a problem hiding this comment.
Is the design still part of the stack?
There was a problem hiding this comment.
It never was in this stack. design.md lives in #38239, which is open and design-only, and whose description says it lands first so the implementation PRs can cite it rather than restate the argument. So the reference resolves iff #38239 merges ahead of this stack, and today it points at a file on no landed branch.
That ordering was tracked nowhere, which is why it could go unnoticed. Filed as CPU-239 on the Interactive read isolation project, naming this call site as the reason: the standing-hold argument is written down only there.
Left the reference as it is. Inlining the argument here would duplicate the design rather than cite it, which is the thing the design PR exists to avoid.
Posted by Claude Code.
2ed3a19 to
cfc275a
Compare
b7520b8 to
0262fa1
Compare
|
Agreed. Fixed in 0262fa1: Posted by Claude Code. |
cfc275a to
4fb99ee
Compare
QA LLM Review1. MEDIUM -- every shared-peek attempt wakes the publishing maintenance worker twice
DetailsEach woken arrange operator runs The cost scales with attempts, not with peeks. A peek parked in The peek does not need the holds. The cursor captures the batches it walks, and the published
|
|
Confirmed. I took the first option, a hold-free capture.
I didn't take the second option, skipping the wake when the accumulated frontiers don't move. A minted reader registers its physical hold at the chain's
Posted by Claude Code. |
| /// | ||
| /// Generic over both traces so a peek served from this runtime's `TraceBundle` and one served from | ||
| /// the sharing registry's published handles are gated by the same code rather than by two copies | ||
| /// that can drift. |
There was a problem hiding this comment.
This is obvious from the signature; it's even generic over more than two (only that two monomorphizations will exist.)
There was a problem hiding this comment.
Removed.
Posted by Claude Code
| ) -> Option<(SharedOksHandle, SharedErrsHandle)> { | ||
| ) -> Option<( | ||
| crate::shared_trace::SharedOksHandle, | ||
| crate::shared_trace::SharedErrsHandle, | ||
| )> { |
There was a problem hiding this comment.
Done. Imported under #[cfg(test)].
Posted by Claude Code
| pub pending_peeks: VecDeque<PendingPeek>, | ||
| /// Shared-index peeks waiting on a publication or a seal, keyed by the index they wait on. | ||
| /// | ||
| /// Empty on the maintenance runtime, which reads the traces it maintains. A peek here waits on | ||
| /// an event rather than on a turn, so no sweep polls it: it is re-examined only when its key is | ||
| /// marked dirty in the sharing registry and the worker is woken. Only peeks wait this way. An | ||
| /// interactive dataflow builds immediately and binds its imports through registry placeholders. | ||
| pub pending_work: BTreeMap<GlobalId, Vec<IndexPeek>>, |
There was a problem hiding this comment.
Is it possible to combine pending_peeks and pending_work into the same field? pending_peeks might benefit from the same optimization that pending_work has.
There was a problem hiding this comment.
pending_work is gone. A peek waiting on a shared index now waits in queued_peeks like a local one and is re-read every sweep. We dropped the separate pending_peeks re-key, because pending_peeks holds only persist and offloaded peeks, which already wake through their own channels. Making wakeups scale with what changed is deferred.
Posted by Claude Code
| // Tripwire for a Multiplexer routing bug. The Multiplexer routes on `is_peek_dataflow`, and | ||
| // this runtime's read path assumes nothing else arrives. Both sides now ask the same | ||
| // question of the same description, so a violation means the two runtimes disagree about a | ||
| // single dataflow, and it would be rendered against a runtime that cannot serve it. | ||
| // | ||
| // `soft_assert_or_log!` rather than `debug_assert!`, which compiles out under | ||
| // `[profile.optimized]` and `[profile.release]`. Those are the profiles mzcompose and | ||
| // `bin/environmentd` build, so a debug assertion here would be absent from the suites that | ||
| // exercise two-runtime most broadly. This panics where soft assertions are on and logs an | ||
| // error everywhere else, so a routing bug is visible in production without taking the | ||
| // replica down for a dataflow that may still render correctly. | ||
| mz_ore::soft_assert_or_log!( | ||
| self.compute_state.role() != ComputeRuntimeRole::Interactive | ||
| || dataflow.is_peek_dataflow(), | ||
| "interactive runtime received a dataflow that is not a peek dataflow: \ | ||
| export_ids={:?} transient={} single_time={} has_subscribes={} has_copy_tos={}", | ||
| dataflow.export_ids().collect::<Vec<_>>(), | ||
| dataflow.is_transient(), | ||
| dataflow.is_single_time(), | ||
| dataflow.subscribe_ids().next().is_some(), | ||
| dataflow.copy_to_ids().next().is_some(), | ||
| ); | ||
|
|
||
| // Every dataflow builds immediately, in command arrival order. Timely allocates per-worker | ||
| // channel ids in construction order, so deferring a build until its dependencies publish | ||
| // would diverge that order across workers, latently unsound under a multi-worker interactive | ||
| // runtime. On the interactive runtime a query dataflow imports its maintenance-index inputs | ||
| // from the sharing registry, binding each through a registry placeholder that a maintenance | ||
| // publisher adopts later (see `ArrangementSharingRegistry::import`). A not-yet-published | ||
| // dependency therefore yields an empty import held at the minimum frontier, so the build is | ||
| // always possible without waiting. |
There was a problem hiding this comment.
I think this one should be part of the PR where we introduce the multiplexer.
There was a problem hiding this comment.
The tripwire is gone rather than moved. Placement is now a DataflowClass on the description, which the adapter sets. The controller soft-asserts in Instance::create_dataflow (#38388) that the class fits the shape, and the runtime trusts the class.
Posted by Claude Code
| // | ||
| // Every dataflow builds immediately on `CreateDataflow`, inserting its | ||
| // `suspended_collections` entry before the `Schedule` that follows in arrival order can | ||
| // reach us. A `Schedule` with no entry is therefore a stray or duplicate command, a silent | ||
| // no-op. |
There was a problem hiding this comment.
It'd still be surprising to receive Schedule commands, because it'd indicate a confusion in the controller. I think it'd make sense to have a soft_assert here, too.
There was a problem hiding this comment.
Done. A Schedule for an id that is neither a collection nor a peer soft-asserts on every runtime (#38391, where peers arrive).
Posted by Claude Code
There was a problem hiding this comment.
Would it be possible that the shared traces also end up in compute_state.traces?
There was a problem hiding this comment.
Yes, that is the design now. TraceBundle holds an OksTrace/ErrsTrace enum, either local or a SharedReader over the peer's publication. The interactive runtime puts a peer bundle into traces for every index a maintenance-placed CreateDataflow exports. Import, peek and compaction then go through traces on both runtimes.
Posted by Claude Code
There was a problem hiding this comment.
The current design deliberately does not change the verbs of the compute command protocol, but in order to do so it silently changes the semantics of the existing verbs. Where a AllowCompaction previously required a CreateDataflow, it does not anymore for the interactive runtime, so depending on who we speak with, the protocol has different invariants, and different meanings.
This is bad from a design point, because interpreting the protocol is not context-free anymore, but requires whoever reads it to have additional information. A cleaner solution would be to introduce a new verb that indicates a different runtime eventually exports an object.
This seems to outweigh the benefits of making the multiplexer transparent.
There was a problem hiding this comment.
Agreed, and done with the alternative we discussed: placement is a DataflowClass on DataflowDescription, and the multiplexer forwards every command to both runtimes (peeks to interactive). The runtime that renders a class builds the dataflow. The other records every export as a peer, and an index export also gets a peer trace whose logical compaction is the hold. So AllowCompaction always names a collection the runtime knows, as either a collection or a peer, and it means "compact" on both. The ordering invariant follows from the broadcast: the peer hold exists from the index's own create on the interactive stream. The standing hold is gone.
Posted by Claude Code
There was a problem hiding this comment.
This would be simplified if we merge the pending peeks.
There was a problem hiding this comment.
Gone with the merge into queued_peeks. Cancel no longer scans a second map.
Posted by Claude Code
| // The interactive runtime keeps no introspection state of its own. Its peeks on the logging | ||
| // indexes read the maintenance runtime's publications from the registry (see | ||
| // `logging::publish_logging_index`), so installing the logging dataflow here would only | ||
| // create collections under ids this runtime does not own, which every command handler | ||
| // would then have to tell apart from its own. Without a `compute_logger` this runtime's | ||
| // own events are not logged. TODO(CPU-222): log them through the maintenance runtime. | ||
| if self.compute_state.role() == ComputeRuntimeRole::Interactive { | ||
| return; | ||
| } |
There was a problem hiding this comment.
We shouldn't try to call initialize_logging on the interactive runtime, instead of adding an early-return here.
There was a problem hiding this comment.
Done. CreateInstance initializes logging only where Placement renders maintained work. The interactive runtime instead holds maintenance's logging indexes as peers. The early return is gone.
Posted by Claude Code
| /// Parks `peek` in `pending_work` until the arrangement it reads is published or seals. | ||
| fn enqueue_shared_peek(&mut self, peek: IndexPeek, published: GlobalId) { | ||
| self.pending_work.entry(published).or_default().push(peek); | ||
| } | ||
|
|
There was a problem hiding this comment.
Consider inlining this function .. or if we merge the different pending variants, this'll go away.
There was a problem hiding this comment.
Gone. enqueue_shared_peek no longer exists.
Posted by Claude Code
There was a problem hiding this comment.
On the pending peek being a vec-deq: Pull out a separate PR (independent of this stack) to key it by global ID. Then the transition will be easier.
There was a problem hiding this comment.
We dropped this after looking closer. pending_peeks never holds index peeks, so keying it by GlobalId would not have eased the merge. The shared peeks merged into queued_peeks instead.
Posted by Claude Code
| if let IndexTraces::Local(trace_bundle) = &mut traces { | ||
| let empty_frontier = Antichain::new(); | ||
| let timestamp_frontier = Antichain::from_elem(peek.timestamp); | ||
| trace_bundle | ||
| .oks_mut() | ||
| .set_logical_compaction(timestamp_frontier.borrow()); | ||
| trace_bundle | ||
| .errs_mut() | ||
| .set_logical_compaction(timestamp_frontier.borrow()); | ||
| trace_bundle | ||
| .oks_mut() | ||
| .set_physical_compaction(empty_frontier.borrow()); | ||
| trace_bundle | ||
| .errs_mut() | ||
| .set_physical_compaction(empty_frontier.borrow()); | ||
| } |
There was a problem hiding this comment.
Confused here; why doesn't the shared trace need to set physical/logical compaction? Also, make this a match.
There was a problem hiding this comment.
It no longer differs. A peek pins both bundle variants the same way at its timestamp. For a shared trace that is a clone of the peer's logical hold, and the writer now wakes only when the readers' meet moves (#38386), so pinning costs the publisher nothing. The if let is gone with the variant split.
Posted by Claude Code
| pub(super) enum IndexTraces { | ||
| /// Traces this runtime maintains, pinned for the peek's life. | ||
| Local(TraceBundle), | ||
| /// An arrangement the sharing registry publishes, resolved on every attempt. A parked peek | ||
| /// holds nothing of the arrangement, so an unpublished slot registers no hold at the minimum. | ||
| Shared { | ||
| registry: ArrangementSharingRegistry, | ||
| /// The id the arrangement is published under. Differs from the peek's target when the | ||
| /// target re-exports an imported shared arrangement, see `ComputeState::shared_reexports`. | ||
| published: GlobalId, | ||
| }, | ||
| } |
There was a problem hiding this comment.
This is not great; if we have the two variants, they should only differ in type, not behavior.
There was a problem hiding this comment.
Done. IndexTraces is deleted. The two kinds of trace now differ only in type, inside OksTrace/ErrsTrace, and behave the same.
Posted by Claude Code
The interactive runtime holds no local traces, so an index peek there resolves against the sharing registry instead. `PeekScan` and the error walk beneath it become generic over the traces they read, and an interactive peek opens its scan over the registry's `SharedOksHandle` and `SharedErrsHandle`. Both flavours of index peek then spend one budget, report one set of metrics, and reach the peek stash through the one offload driver. A shared peek whose arrangement is not yet published, or whose upper has not sealed the peek timestamp, waits in `pending_work` keyed by a `WorkId` and indexed by the id it waits on. A publication or seal marks that id dirty and wakes the worker, which gives a turn to exactly the items indexed under the ids that changed, so wakeups scale with what changed rather than with total pending work. Past that gate a shared peek is an ordinary index peek: it queues for a turn like any other, and a walk that outruns the activation's fuel or fills a batch bound for the stash leaves for a driver. The sweep still runs on every step, because a persist read and an offloaded walk each wake the worker through a channel of their own rather than through the dirty set, but nothing waiting on a publication or a seal is ever swept. Interactive dataflows build immediately in command arrival order rather than deferring until their dependency is published. An import over an unadopted placeholder produces no data and holds its output frontier at the minimum until a publisher adopts the same slot, so late binding replaces the deferral. The interactive runtime reports only its transient collections' frontiers. It shares the identity of every non-transient collection with maintenance, which owns and reports the real frontiers, and the controller keeps one frontier stream per collection, so reporting the shared ones would race the owner and regress it. For the same reason its logging is forced off: it serves introspection peeks from maintenance's published copies, and its own empty copies would clobber them. The consequence is that nothing the interactive runtime does appears in introspection, tracked as CPU-222. Reconciliation drops `pending_work` and `dep_index`, whose peeks belong to the reconciled-away connection. The standing holds in the registry deliberately survive: one is per collection, carries no dataflow identity, and only rises, so clearing it would drop the arrangement's bound to the minimum until replayed compactions raised it again. Reachable only on a runtime holding the `Interactive` role, which requires the dyncfg that is still off everywhere. Tests are out of line in `compute_state/tests.rs`, per the convention in `src/compute/AGENTS.md`. Co-Authored-By: Claude Opus 5 (1M context) <[email protected]>
`handle_allow_compaction` no longer notes the controller's frontier on the registry: the trace manager's compaction of the index is what the published `since` reports. Tests keep a trace agent alive for as long as they read, since the point closes with the trace. Co-Authored-By: Claude Fable 5.1 <[email protected]> Claude-Session: https://claude.ai/code/session_01VDm7opomJLxbNUEP3r9BLk
Reconciliation pads every logging index's trace so the controller can read those collections from the minimum time after a reconnect. It takes the ids from the `CreateInstance` config, which the multiplexer delivers to both runtimes, and requires each to name a trace this runtime hosts. The interactive runtime installs no logging dataflow, so it hosts none of them and serves peeks on those ids from the sharing registry instead. The first reconciliation therefore panicked with "logging trace exists" on every interactive worker, taking the process with it. It needs a controller reconnect to reach, which is why the upgrade, restart and zippy suites saw it and the others did not. Skip the padding on the interactive runtime rather than tolerating a missing trace, because a logging trace absent on the maintenance runtime is still a bug and should stay loud. Co-Authored-By: Claude Opus 5 (1M context) <[email protected]> Claude-Session: https://claude.ai/code/session_015tLhSbZdXrTSK2KwSocT59
Co-Authored-By: Claude Opus 5.5 (1M context) <[email protected]> Claude-Session: https://claude.ai/code/session_015tLhSbZdXrTSK2KwSocT59
A shared peek attempt captured its arrangement through `registry.handles`, and dropping those readers at the end of the attempt woke the publishing maintenance worker every time, scheduling both arrange operators for nothing. The attempt now captures `ArrangementSharingRegistry::snapshots`, which registers no hold and so wakes nobody. The capture is sound without a hold because its batches are immutable and it reads only at or beyond its own `since`, which the gate checks. Co-Authored-By: Claude Opus 5.5 (1M context) <[email protected]>
Co-Authored-By: Claude Opus 5.5 (1M context) <[email protected]>
Co-Authored-By: Claude Opus 5.5 (1M context) <[email protected]>
Co-Authored-By: Claude Opus 5.5 (1M context) <[email protected]>
Co-Authored-By: Claude Opus 5.5 (1M context) <[email protected]>
|
Folded into #38391. After the rework this layer held only tests and test helpers, so they now ship with the runtime they exercise. Posted by Claude Code |
Seventh of eight PRs splitting #37770. Stacks on #38391. Tracked by CPU-216.
An index peek on the interactive runtime reads
compute_state.traceslike on any runtime. Its maintained indexes are peer bundles there, so oneIndexPeek, one scan, and one sweep serve both runtimes, and a peek pins a shared trace at its timestamp the same way it pins a local one. Pinning at or above the readers' meet does not wake the publisher.A peek whose index has not sealed its timestamp waits in
queued_peeksand is re-read by every sweep. The interactive worker registers itself with its registry when it is built, so every publication and seal unparks it. The index peek tests drive interactiveComputeStates over peer bundles through the ordinary sweep.The interactive runtime reports only the collections it renders and installs no logging, so nothing it does appears in introspection, tracked as CPU-222. That wants resolving before the flag is turned on anywhere it would hide real work.
Reachable only on a runtime holding the
Interactiverole, which requires the dyncfg that is still off everywhere.🤖 Generated with Claude Code