compute: serve fast-path peeks on the interactive runtime

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) <noreply@anthropic.com>

Author: antiguru

src/compute/src/compute_state.rs

@@ -8,27 +8,58 @@
 
 use std::time::{Duration, Instant};
 
-use differential_dataflow::trace::{Cursor, TraceReader};
+use differential_dataflow::trace::{Cursor, Navigable, TraceReader};
 use mz_compute_client::protocol::response::PeekError;
 use mz_repr::{Diff, GlobalId, Timestamp};
 use timely::order::PartialOrder;
 use tracing::error;
 
 use crate::arrangement::manager::PaddedTrace;
 use crate::compute_state::{PeekRowIterationTracker, peek_result_iterator};
+use crate::render::errors::DataflowErrorSer;
 use crate::typedefs::ErrAgent;
 
 /// The error trace of an index, as
 /// [`TraceBundle::errs_mut`](crate::arrangement::manager::TraceBundle::errs_mut) hands it out.
 pub(super) type ErrsHandle = PaddedTrace<ErrAgent<Timestamp, Diff>>;
 
+/// A trace an index peek's error walk can read.
+///
+/// The bound is spelled once here, so the walk and everything that carries it name the shape
+/// rather than restate it.
+pub(super) trait PeekErrsTrace:
+    TraceReader<
+    Batch: Navigable<
+        Cursor: for<'a> Cursor<
+            Key<'a> = &'a DataflowErrorSer,
+            TimeGat<'a>: PartialOrder<Timestamp>,
+            DiffGat<'a> = &'a Diff,
+        >,
+    >,
+>
+{
+}
+
+impl<Tr> PeekErrsTrace for Tr where
+    Tr: TraceReader<
+        Batch: Navigable<
+            Cursor: for<'a> Cursor<
+                Key<'a> = &'a DataflowErrorSer,
+                TimeGat<'a>: PartialOrder<Timestamp>,
+                DiffGat<'a> = &'a Diff,
+            >,
+        >,
+    >
+{
+}
+
 /// A walk over an index peek's error trace, suspendable between cursor positions.
 ///
 /// Holds nothing of the ok trace or of the rows a peek returns. A peek reaches those only once
 /// this walk reports [`ErrorScanStep::Finished`] with an `Ok`.
-pub(super) struct ErrorScan {
-    cursor: peek_result_iterator::TraceCursor<ErrsHandle>,
-    storage: peek_result_iterator::TraceStorage<ErrsHandle>,
+pub(super) struct ErrorScan<Tr: PeekErrsTrace> {
+    cursor: peek_result_iterator::TraceCursor<Tr>,
+    storage: peek_result_iterator::TraceStorage<Tr>,
     /// The limit spans this walk and the ok scan after it, so the count accrued here is handed
     /// on with [`ErrorScanStep::Finished`].
     row_iteration_tracker: PeekRowIterationTracker,
@@ -48,12 +79,12 @@ pub(super) enum ErrorScanStep {
     OutOfFuel,
 }
 
-impl ErrorScan {
+impl<Tr: PeekErrsTrace> ErrorScan<Tr> {
     /// Opens a walk over `errs`.
     ///
     /// The walk starts without a row-iteration limit. The limit in effect is the caller's to
     /// supply through [`ErrorScan::set_row_iteration_limit`] before each step.
-    pub(super) fn new(errs: &mut ErrsHandle) -> Self {
+    pub(super) fn new(errs: &mut Tr) -> Self {
         let scan_start = Instant::now();
         let (cursor, storage) = errs.cursor();
         let mut scan = Self::from_cursor(cursor, storage);
@@ -63,8 +94,8 @@ impl ErrorScan {
 
     /// Opens a walk over an already-opened cursor.
     pub(super) fn from_cursor(
-        cursor: peek_result_iterator::TraceCursor<ErrsHandle>,
-        storage: peek_result_iterator::TraceStorage<ErrsHandle>,
+        cursor: peek_result_iterator::TraceCursor<Tr>,
+        storage: peek_result_iterator::TraceStorage<Tr>,
     ) -> Self {
         Self {
             cursor,

src/compute/src/compute_state/error_scan.rs

@@ -51,7 +51,10 @@ pub(crate) fn error_batch(
 
 /// Builds a walk over a single-batch error trace holding `updates`, bounded by
 /// `row_iteration_limit`.
-pub(crate) fn error_scan(updates: ErrorUpdates, row_iteration_limit: Option<usize>) -> ErrorScan {
+pub(crate) fn error_scan(
+    updates: ErrorUpdates,
+    row_iteration_limit: Option<usize>,
+) -> ErrorScan<ErrsHandle> {
     let storage = vec![error_batch(updates)];
     let cursor = CursorList::new(vec![storage[0].cursor()], &storage);
     let mut scan = ErrorScan::from_cursor(cursor, storage);
@@ -79,7 +82,10 @@ pub(crate) fn holding(error: &DataflowErrorSer) -> ErrorUpdates {
 
 /// Runs `scan` to an answer in slices of `fuel_per_step` units, and returns that answer, the
 /// fuel the walk spent, and the number of calls it took.
-fn run_sliced(scan: &mut ErrorScan, fuel_per_step: usize) -> (ErrorScanStep, usize, usize) {
+fn run_sliced(
+    scan: &mut ErrorScan<ErrsHandle>,
+    fuel_per_step: usize,
+) -> (ErrorScanStep, usize, usize) {
     let mut consumed = 0;
     // Bounded so that a walk which restarts from the first key on each resumption fails the
     // test instead of hanging it.

src/compute/src/compute_state/error_scan/tests.rs

@@ -282,8 +282,8 @@ enum Answer {
     Ready(PeekResponse),
 }
 
-impl From<PeekStatus> for Answer {
-    fn from(status: PeekStatus) -> Self {
+impl From<PeekStatus<IndexPeekScan>> for Answer {
+    fn from(status: PeekStatus<IndexPeekScan>) -> Self {
         match status {
             PeekStatus::NotReady => Answer::NotReady,
             // The scan an offload carries has no comparison of its own. What is comparable

src/compute/src/compute_state/index_peek_tests.rs

@@ -38,8 +38,11 @@ use tracing::{debug, warn};
 use uuid::Uuid;
 
 use crate::compute_state::PeekRowIterationConfig;
+use crate::compute_state::error_scan::PeekErrsTrace;
 use crate::compute_state::peek_metrics::PeekWalkMetrics;
-use crate::compute_state::peek_scan::{IndexPeekScan, RowBatch, ScanOutcome, rows_response};
+use crate::compute_state::peek_scan::{
+    PeekOksTrace, PeekScan, RowBatch, ScanOutcome, rows_response,
+};
 use crate::compute_state::peek_stash::{StashTarget, StashUpload};
 
 /// The bound on how many offloaded peek walks run at once.
@@ -221,15 +224,20 @@ impl OffloadedPeek {
     ///
     /// The scan may already hold a full batch. This driver takes it, so offloading is how a peek
     /// too large to answer inline reaches the stash.
-    pub(super) fn start(
+    pub(super) fn start<Tr, ETr>(
         peek: Peek,
-        scan: IndexPeekScan,
+        scan: PeekScan<Tr, ETr>,
         stash: Option<StashTarget>,
         permits: Arc<PeekPermits>,
         config: OffloadConfig,
         metrics: PeekWalkMetrics,
         worker: Thread,
-    ) -> Self {
+    ) -> Self
+    where
+        Tr: PeekOksTrace,
+        ETr: PeekErrsTrace,
+        PeekScan<Tr, ETr>: Send + 'static,
+    {
         let (mut result_tx, result_rx) = oneshot::channel();
         permits.resize(config.permit_fraction.get());
 
@@ -305,14 +313,19 @@ impl OffloadedPeek {
     /// Drives the scan in `state` to the peek's answer, writing what it may not answer with inline
     /// to `stash`. `None` means the peek was cancelled, which is the one way the walk ends without
     /// an answer.
-    async fn walk(
-        mut state: WalkState,
+    async fn walk<Tr, ETr>(
+        mut state: WalkState<Tr, ETr>,
         peek_uuid: Uuid,
         stash: Option<StashTarget>,
         config: &OffloadConfig,
         metrics: &PeekWalkMetrics,
         order_by: Arc<[ColumnOrder]>,
-    ) -> (WalkState, Option<PeekResponse>) {
+    ) -> (WalkState<Tr, ETr>, Option<PeekResponse>)
+    where
+        Tr: PeekOksTrace,
+        ETr: PeekErrsTrace,
+        PeekScan<Tr, ETr>: Send + 'static,
+    {
         // Opened by the first batch the scan hands over, so a walk that never crosses the stash
         // threshold neither opens a shard nor writes a byte. Whether it is open is also what
         // decides how the peek is answered: an upload answers with a handle, and no upload means
@@ -436,15 +449,23 @@ impl OffloadedPeek {
 /// on a blocking thread that an abort cannot interrupt, and the permit accounts for that thread
 /// until the scan leaves it. Fields drop in declaration order, so the scan and its batches go
 /// before the permit that accounts for them.
-struct WalkState {
-    scan: IndexPeekScan,
+struct WalkState<Tr, ETr>
+where
+    Tr: PeekOksTrace,
+    ETr: PeekErrsTrace,
+{
+    scan: PeekScan<Tr, ETr>,
     _permit: WalkPermit,
     /// The sending end of the peek's result channel. Its receiver is dropped by cancellation and
     /// by nothing else, so a closed channel is the cancellation signal.
     result_tx: oneshot::Sender<(PeekResponse, Duration)>,
 }
 
-impl WalkState {
+impl<Tr, ETr> WalkState<Tr, ETr>
+where
+    Tr: PeekOksTrace,
+    ETr: PeekErrsTrace,
+{
     /// Steps the scan until it ends, offers a batch, or the peek is cancelled, whichever comes
     /// first. `None` is the cancellation.
     ///

src/compute/src/compute_state/peek_offload.rs

@@ -27,7 +27,7 @@ use crate::compute_state::index_peek_tests::{
     cancelling_errors, index_peek, ok_row, rows_answer, trace_bundle, trivial_finishing,
     wide_ok_rows,
 };
-use crate::compute_state::peek_scan::{PeekScan, StashBounds};
+use crate::compute_state::peek_scan::{IndexPeekScan, PeekScan, StashBounds};
 use crate::compute_state::peek_stash::tests::{CountedBlob, stashed_rows};
 use crate::metrics::{ComputeMetrics, WorkerMetrics};
 use crate::server::ComputeRuntimeRole;

src/compute/src/compute_state/peek_offload/tests.rs

@@ -14,7 +14,6 @@ use std::mem;
 use std::num::{NonZeroI64, NonZeroUsize};
 use std::time::{Duration, Instant};
 
-use differential_dataflow::trace::cursor::BatchCursor;
 use differential_dataflow::trace::implementations::BatchContainer;
 use differential_dataflow::trace::{Cursor, Navigable, TraceReader};
 use mz_compute_client::protocol::command::Peek;
@@ -27,13 +26,51 @@ use mz_repr::fixed_length::ExtendDatums;
 use mz_repr::{Diff, GlobalId, Row, Timestamp};
 use timely::order::PartialOrder;
 
-use crate::compute_state::error_scan::{ErrorScan, ErrorScanStep, ErrsHandle};
+use crate::arrangement::manager::PaddedTrace;
+use crate::compute_state::error_scan::{ErrorScan, ErrorScanStep, ErrsHandle, PeekErrsTrace};
 use crate::compute_state::peek_result_iterator::{PeekResultIterator, Step};
+use crate::shared_trace::{SharedErrsHandle, SharedOksHandle};
+use crate::typedefs::RowRowAgent;
 
-/// The scan an index peek builds, over the ok trace of the arrangement that answers it.
-pub(super) type IndexPeekScan = PeekScan<
-    crate::arrangement::manager::PaddedTrace<crate::typedefs::RowRowAgent<Timestamp, Diff>>,
->;
+/// A trace an index peek's ok walk can read.
+///
+/// The bound is spelled once here, so the walk and everything that carries it name the shape
+/// rather than restate it.
+pub(super) trait PeekOksTrace:
+    TraceReader<
+    Batch: Navigable<
+        Cursor: for<'a> Cursor<
+            Key<'a>: ExtendDatums + Eq,
+            KeyContainer: BatchContainer<Owned = Row>,
+            Val<'a>: ExtendDatums,
+            TimeGat<'a>: PartialOrder<Timestamp>,
+            DiffGat<'a> = &'a Diff,
+        >,
+    >,
+>
+{
+}
+
+impl<Tr> PeekOksTrace for Tr where
+    Tr: TraceReader<
+        Batch: Navigable<
+            Cursor: for<'a> Cursor<
+                Key<'a>: ExtendDatums + Eq,
+                KeyContainer: BatchContainer<Owned = Row>,
+                Val<'a>: ExtendDatums,
+                TimeGat<'a>: PartialOrder<Timestamp>,
+                DiffGat<'a> = &'a Diff,
+            >,
+        >,
+    >
+{
+}
+
+/// The scan an index peek over a trace this runtime maintains builds.
+pub(super) type IndexPeekScan = PeekScan<PaddedTrace<RowRowAgent<Timestamp, Diff>>, ErrsHandle>;
+
+/// The scan an index peek over an arrangement the sharing registry publishes builds.
+pub(super) type SharedIndexPeekScan = PeekScan<SharedOksHandle, SharedErrsHandle>;
 
 /// Rows a scan hands to its driver, in the order the scan produced them.
 ///
@@ -128,9 +165,9 @@ pub(super) enum ScanOutcome {
 /// The state of a [`PeekScan`]'s walk over its error trace.
 ///
 /// Both ended states drop the walk, so a peek pins error batches only while it reads them.
-enum ErrorPhase {
+enum ErrorPhase<Tr: PeekErrsTrace> {
     /// The walk is under way, and resumes from the cursor position it stopped on.
-    Scanning(ErrorScan),
+    Scanning(ErrorScan<Tr>),
     /// The error trace holds no error at the peek's timestamp, which is the only way to the ok
     /// trace. The rows the walk examined have been handed to the ok walk.
     Clean,
@@ -146,15 +183,16 @@ enum ErrorPhase {
 /// A stash-eligible scan retains at most the threshold before its first batch and the batch size
 /// after, plus the row that crossed either. A scan that cannot use the stash fills no batch, and
 /// `max_result_size` alone bounds its prefix.
-pub(super) struct PeekScan<Tr>
+pub(super) struct PeekScan<Tr, ETr>
 where
-    Tr: TraceReader<Batch: Navigable>,
+    Tr: PeekOksTrace,
+    ETr: PeekErrsTrace,
 {
     /// The time at which the error trace is read.
     peek_timestamp: Timestamp,
     /// The collection the peek reads, for logging.
     target_id: GlobalId,
-    error_phase: ErrorPhase,
+    error_phase: ErrorPhase<ETr>,
     /// The walk over the ok trace, reached only once the error walk reports the error trace
     /// clean. Its cursor is opened with the scan, and nothing advances it before then.
     oks: PeekResultIterator<Tr>,
@@ -195,16 +233,10 @@ where
     pub(super) rows_thinned: usize,
 }
 
-impl<Tr> PeekScan<Tr>
+impl<Tr, ETr> PeekScan<Tr, ETr>
 where
-    Tr: TraceReader<Batch: Navigable>,
-    for<'a> BatchCursor<Tr>: Cursor<
-            Key<'a>: ExtendDatums + Eq,
-            KeyContainer: BatchContainer<Owned = Row>,
-            Val<'a>: ExtendDatums,
-            TimeGat<'a>: PartialOrder<Timestamp>,
-            DiffGat<'a> = &'a Diff,
-        >,
+    Tr: PeekOksTrace,
+    ETr: PeekErrsTrace,
 {
     /// Opens a scan of `peek` over the traces that answer it.
     ///
@@ -213,7 +245,7 @@ where
     /// the caller's to supply to each [`PeekScan::step`].
     pub(super) fn new(
         peek: &Peek,
-        errs_handle: &mut ErrsHandle,
+        errs_handle: &mut ETr,
         oks_handle: &mut Tr,
         max_result_size: u64,
         stash: StashBounds,

src/compute/src/compute_state/peek_scan.rs

@@ -116,7 +116,7 @@ fn ok_iterator_with_copies(keys: &[Row], copies: Diff) -> PeekResultIterator<Tes
 
 /// A walk over an error trace holding `keys` errors that each cancel to zero at
 /// [`PEEK_TIMESTAMP`], so the walk examines every one of them and finds no error.
-fn clean_error_scan(keys: usize) -> ErrorScan {
+fn clean_error_scan(keys: usize) -> ErrorScan<ErrsHandle> {
     crate::compute_state::error_scan::tests::error_scan(cancelling_errors(keys), None)
 }
 
@@ -125,7 +125,7 @@ fn clean_error_scan(keys: usize) -> ErrorScan {
 /// Mirrors what [`PeekScan::new`] builds. Tests use this rather than `new` to hold a second
 /// cursor layout under test, and to start from an [`ErrorPhase`] that a fresh scan cannot be
 /// in.
-fn scan(error_phase: ErrorPhase, keys: &[Row]) -> PeekScan<TestTrace> {
+fn scan(error_phase: ErrorPhase<ErrsHandle>, keys: &[Row]) -> PeekScan<TestTrace, ErrsHandle> {
     PeekScan {
         peek_timestamp: PEEK_TIMESTAMP,
         target_id: GlobalId::User(1),
@@ -613,7 +613,7 @@ fn open(
     max_result_size: u64,
     peek_stash_eligible: bool,
     peek_stash_threshold_bytes: usize,
-) -> PeekScan<OksHandle> {
+) -> PeekScan<OksHandle, ErrsHandle> {
     let (oks, errs) = bundle.oks_errs_mut();
     PeekScan::new(
         peek,
@@ -635,7 +635,7 @@ fn open(
 /// this reports is comparable across runs that cross the stash threshold and runs that do
 /// not.
 fn run_sliced(
-    subject: &mut PeekScan<OksHandle>,
+    subject: &mut PeekScan<OksHandle, ErrsHandle>,
     fuel_per_step: usize,
     row_iteration_limit: Option<usize>,
 ) -> (ScanOutcome, RowBatch, usize) {