compute: publish maintained indexes into the sharing registry

Both export paths now publish their `oks`/`errs` arrangements into the per-process
registry when the runtime's role publishes. A re-export arm has no streams of its
own, so it registers its id as an alias of the arrangement's existing publication
point, which leaves the re-export's dataflow without operators, as it is on a
runtime that does not publish. A reader that bound the re-export's id before the
render holds its own unbacked point, which only a publisher into it can back, so
that case re-imports the shared traces and publishes them under the new id, and
logs the imported errors because `mz_compute_error_counts` forwards a dependency's
counts only to a re-export whose dataflow has no operators. Logging indexes
publish the same way, gated strictly on `Maintenance`: an interactive runtime
reads maintenance's slot, and its own copy would clobber it, while `Solo` has no
registry peer at all.

An alias shares its target's frontiers while the target lives: the alias dataflow
imports the target, so the controller never advances the target's `since` past an
alias's, and the target's frontier bounds every reader of the shared point. Once
the target drops, the meet of the remaining aliases' frontiers governs the point,
since the shared trace then compacts to exactly that meet. Seal notifications fan
out from the target to its aliases, since a reader waits under the id it imported.

`ComputeRuntimeRole::Interactive` stops being test-only. Nothing constructs it
yet, but `publishes()` has to name it, and `pub mod server` keeps the variant
reachable so dead-code analysis is satisfied without an attribute. The stale
`owns_process_globals` note claiming every constructible role owns the globals
goes with it.

Carrying the role and the registry to the render path is what the rest of this
change is: `Config` and `Worker` gain both, `ComputeState` stores them and
exposes `role()`, and clusterd builds one registry per process. Per process, not
per runtime, because a reader on one runtime looks up the slot a publisher on
another filled.

No behavior change. `Solo` is the only role anything constructs and it does not
publish, so every added block is skipped and no dataflow gains an operator.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01VDm7opomJLxbNUEP3r9BLk

Author: antiguru

src/clusterd/src/lib.rs

@@ -22,6 +22,7 @@ use mz_build_info::{BuildInfo, build_info};
 use mz_cloud_resources::AwsExternalIdPrefix;
 use mz_cluster_client::client::TimelyConfig;
 use mz_compute::server::{ComputeInstanceContext, ComputeRuntimeRole};
+use mz_compute::sharing::ArrangementSharingRegistry;
 use mz_http_util::DynamicFilterTarget;
 use mz_orchestrator_tracing::{StaticTracingConfig, TracingCliArgs};
 use mz_ore::cli::{self, CliConfig};
@@ -473,11 +474,16 @@ async fn run(args: Args) -> Result<(), anyhow::Error> {
     );
 
     // Start compute server.
+    //
+    // The sharing registry is per process rather than per runtime: a reader on one runtime looks up
+    // the slot a publisher on another runtime filled, so both must hold the same registry.
+    let sharing_registry = ArrangementSharingRegistry::new();
     let compute_client_builder = mz_compute::server::serve(
         compute_timely_config,
         ComputeRuntimeRole::Solo,
         &metrics_registry,
         persist_clients,
+        sharing_registry,
         txns_ctx,
         tracing_handle,
         ComputeInstanceContext {

src/compute/src/arrangement/manager.rs

@@ -228,6 +228,20 @@ where
     }
 }
 
+impl<Tr> PaddedTrace<TraceAgent<Tr>>
+where
+    Tr: TraceReader + 'static,
+{
+    /// Imports the trace into `scope` as a live arrangement named `name`.
+    pub fn import_named<'scope>(
+        &self,
+        scope: Scope<'scope, Tr::Time>,
+        name: &str,
+    ) -> Arranged<'scope, TraceAgent<Tr>> {
+        self.trace.clone().import_named(scope, name)
+    }
+}
+
 /// Bundles together traces for the successful computations (`oks`), the
 /// failed computations (`errs`), additional tokens that should share
 /// the lifetime of the bundled traces (`to_drop`).
@@ -287,6 +301,21 @@ impl TraceBundle {
         (&mut self.oks, &mut self.errs)
     }
 
+    /// Imports both traces into `scope` as live arrangements, for publishers to attach to.
+    pub fn import_named<'scope>(
+        &self,
+        scope: Scope<'scope, Timestamp>,
+        name: &str,
+    ) -> (
+        Arranged<'scope, RowRowAgent<Timestamp, Diff>>,
+        Arranged<'scope, ErrAgent<Timestamp, Diff>>,
+    ) {
+        (
+            self.oks.import_named(scope.clone(), &format!("{name} oks")),
+            self.errs.import_named(scope, &format!("{name} errs")),
+        )
+    }
+
     /// Returns a reference to the `to_drop` tokens.
     pub fn to_drop(&self) -> &Option<Rc<dyn Any>> {
         &self.to_drop

src/compute/src/compute_state.rs

@@ -77,7 +77,8 @@ use crate::logging::compute::{CollectionLogging, ComputeEvent, PeekEvent};
 use crate::logging::initialize::LoggingTraces;
 use crate::metrics::{CollectionMetrics, WorkerMetrics};
 use crate::render::{LinearJoinSpec, StartSignal};
-use crate::server::{ComputeInstanceContext, ResponseSender};
+use crate::server::{ComputeInstanceContext, ComputeRuntimeRole, ResponseSender};
+use crate::sharing::ArrangementSharingRegistry;
 
 mod error_scan;
 mod peek_budget;
@@ -214,6 +215,11 @@ pub struct ComputeState {
     /// A process-global cache of (blob_uri, consensus_uri) -> PersistClient.
     /// This is intentionally shared between workers.
     pub persist_clients: Arc<PersistClientCache>,
+    /// A per-process registry of published index arrangements.
+    ///
+    /// Intentionally shared between all workers of the process, each of which publishes into its own
+    /// worker-ordinal slot. `Clone` shares the same underlying map.
+    pub sharing_registry: ArrangementSharingRegistry,
     /// Context necessary for rendering txn-wal operators.
     pub txns_ctx: TxnsContext,
     /// History of commands received by this workers and all its peers.
@@ -295,6 +301,12 @@ pub struct ComputeState {
 
     /// The storage worker forwards its introspection logs to the compute worker.
     pub storage_log_reader: Option<crate::server::StorageTimelyLogReader>,
+
+    /// Which of the process's compute runtimes this state belongs to.
+    ///
+    /// Only the maintenance runtime runs the non-idempotent process-global initializers. The
+    /// interactive runtime shares the same process and inherits those globals.
+    role: ComputeRuntimeRole,
 }
 
 impl ComputeState {
@@ -308,7 +320,9 @@ impl ComputeState {
 
     /// Construct a new `ComputeState`.
     pub fn new(
+        role: ComputeRuntimeRole,
         persist_clients: Arc<PersistClientCache>,
+        sharing_registry: ArrangementSharingRegistry,
         txns_ctx: TxnsContext,
         metrics: WorkerMetrics,
         tracing_handle: Arc<TracingHandle>,
@@ -334,6 +348,7 @@ impl ComputeState {
             peek_stash_persist_location: None,
             compute_logger: None,
             persist_clients,
+            sharing_registry,
             txns_ctx,
             command_history,
             max_result_size: u64::MAX,
@@ -353,9 +368,15 @@ impl ComputeState {
             init_system_time: mz_ore::now::SYSTEM_TIME(),
             replica_expiration: Antichain::default(),
             storage_log_reader,
+            role,
         }
     }
 
+    /// Which of the process's compute runtimes this state serves.
+    pub(crate) fn role(&self) -> ComputeRuntimeRole {
+        self.role
+    }
+
     /// Return a mutable reference to the identified collection.
     ///
     /// Panics if the collection doesn't exist.
@@ -1033,6 +1054,8 @@ impl<'a> ActiveComputeState<'a> {
             Rc::clone(&self.compute_state.worker_config),
             self.compute_state.workers_per_process,
             storage_log_reader,
+            self.compute_state.role(),
+            self.compute_state.sharing_registry.clone(),
         );
 
         let dataflow_index = Rc::new(dataflow_index);

src/compute/src/compute_state/peek_sweep_tests.rs

@@ -23,6 +23,7 @@ use tokio::sync::mpsc;
 
 use crate::metrics::ComputeMetrics;
 use crate::server::ComputeRuntimeRole;
+use crate::sharing::ArrangementSharingRegistry;
 
 use super::index_peek_tests::{
     TARGET_ID, cancelling_errors, index_peek_with_uuid, rows_answer, trace_bundle, wide_ok_rows,
@@ -104,7 +105,9 @@ impl Harness {
         };
 
         let state = ComputeState::new(
+            ComputeRuntimeRole::Solo,
             Arc::new(PersistClientCache::new_no_metrics()),
+            ArrangementSharingRegistry::new(),
             TxnsContext::default(),
             metrics,
             Arc::new(TracingHandle::disabled()),

src/compute/src/logging/initialize.rs

@@ -16,7 +16,7 @@ use differential_dataflow::logging::{DifferentialEvent, DifferentialEventBuilder
 use mz_compute_client::logging::{LogVariant, LoggingConfig};
 use mz_dyncfg::ConfigSet;
 use mz_ore::metrics::MetricsRegistry;
-use mz_repr::{Diff, Timestamp};
+use mz_repr::{Diff, GlobalId, Timestamp};
 use mz_storage_operators::persist_source::Subtime;
 use mz_timely_util::columnar::Column;
 use mz_timely_util::columnar::builder::ColumnBuilder;
@@ -35,7 +35,9 @@ use crate::extensions::arrange::{KeyCollection, MzArrange};
 use crate::logging::compute::{ComputeEvent, ComputeEventBuilder};
 use crate::logging::{BatchLogger, EventQueue, SharedLoggingState};
 use crate::render::errors::DataflowErrorSer;
-use crate::typedefs::{ErrBatcher, ErrBuilder};
+use crate::server::ComputeRuntimeRole;
+use crate::sharing::ArrangementSharingRegistry;
+use crate::typedefs::{ErrAgent, ErrBatcher, ErrBuilder, RowRowAgent};
 
 /// Initialize logging dataflows.
 ///
@@ -48,6 +50,8 @@ pub fn initialize(
     worker_config: Rc<ConfigSet>,
     workers_per_process: usize,
     storage_log_reader: Option<crate::server::StorageTimelyLogReader>,
+    role: ComputeRuntimeRole,
+    sharing_registry: ArrangementSharingRegistry,
 ) -> LoggingTraces {
     let interval_ms = std::cmp::max(1, config.interval.as_millis());
 
@@ -74,6 +78,8 @@ pub fn initialize(
         worker_config,
         workers_per_process,
         storage_log_reader,
+        role,
+        sharing_registry,
     };
 
     // Depending on whether we should log the creation of the logging dataflows, we register the
@@ -114,6 +120,11 @@ struct LoggingContext<'a> {
     workers_per_process: usize,
     /// Optional reader for storage timely logging events.
     storage_log_reader: Option<crate::server::StorageTimelyLogReader>,
+    /// This runtime's role. Only `Maintenance` publishes its logging indexes into the sharing
+    /// registry.
+    role: ComputeRuntimeRole,
+    /// The per-process registry maintenance publishes its logging indexes into.
+    sharing_registry: ArrangementSharingRegistry,
 }
 
 pub(crate) struct LoggingTraces {
@@ -206,6 +217,20 @@ impl LoggingContext<'_> {
             let traces = collections
                 .into_iter()
                 .map(|(log, collection)| {
+                    // Publish maintenance's logging index into the sharing registry so the
+                    // interactive runtime serves introspection peeks from it. Gated on the
+                    // Maintenance role inside the helper, so this is a no-op (adds no operators) on
+                    // Interactive and Solo.
+                    if let Some(&id) = self.config.index_logs.get(&log) {
+                        publish_logging_index(
+                            self.role,
+                            &self.sharing_registry,
+                            &scope,
+                            id,
+                            &collection.trace,
+                            &errs,
+                        );
+                    }
                     let bundle = TraceBundle::new(collection.trace, errs.clone())
                         .with_drop(collection.token);
                     (log, bundle)
@@ -353,3 +378,34 @@ impl ExtractTimestamp for (Timestamp, Subtime) {
         self.0
     }
 }
+
+/// Publishes a maintenance logging index's `oks`/`errs` arrangements into the sharing registry so
+/// the interactive runtime serves introspection peeks from them.
+///
+/// Gated on the `Maintenance` role. Interactive reads maintenance's slot, and its own empty copy
+/// would clobber it. Solo has no registry peer.
+fn publish_logging_index(
+    role: ComputeRuntimeRole,
+    registry: &ArrangementSharingRegistry,
+    scope: &timely::dataflow::Scope<'_, Timestamp>,
+    id: GlobalId,
+    oks_trace: &RowRowAgent<Timestamp, Diff>,
+    errs_trace: &ErrAgent<Timestamp, Diff>,
+) {
+    if role != ComputeRuntimeRole::Maintenance {
+        return;
+    }
+
+    // The arrange streams are consumed inside the per-log construction regions, so only the trace
+    // handles survive here. Re-import them to give the publishers a live stream to attach to.
+    let oks = oks_trace
+        .clone()
+        .import_named(scope.clone(), &format!("PublishLog({id})"));
+    let errs = errs_trace
+        .clone()
+        .import_named(scope.clone(), &format!("PublishLogErr({id})"));
+    registry.publish(id, &oks, &errs);
+}
+
+#[cfg(test)]
+mod tests;

src/compute/src/logging/initialize/tests.rs

@@ -0,0 +1,82 @@
+// Copyright Materialize, Inc. and contributors. All rights reserved.
+//
+// Use of this software is governed by the Business Source License
+// included in the LICENSE file.
+//
+// As of the Change Date specified in that file, in accordance with
+// the Business Source License, use of this software will be governed
+// by the Apache License, Version 2.0.
+
+use differential_dataflow::input::Input;
+use mz_repr::{Diff, GlobalId, Row, Timestamp};
+use mz_row_spine::{RowRowBatcher, RowRowBuilder};
+use mz_timely_util::columnation::ColumnationChunker;
+
+use crate::extensions::arrange::{KeyCollection, MzArrange};
+use crate::render::errors::DataflowErrorSer;
+use crate::server::ComputeRuntimeRole;
+use crate::sharing::ArrangementSharingRegistry;
+use crate::typedefs::{ErrBatcher, ErrBuilder, ErrSpine, RowRowSpine};
+
+use super::publish_logging_index;
+
+/// A logging/introspection index is a `RowRow` `oks` arrangement plus an (empty) `errs`
+/// arrangement, published into the sharing registry only by the maintenance runtime. Interactive
+/// and Solo must not publish: interactive reads maintenance's slot rather than clobbering it with
+/// its own empty copy, and Solo has no registry peer.
+///
+/// Builds real `RowRow`/`Err` arrangements (the exact types the logging path produces) and drives
+/// [`publish_logging_index`] for each role, asserting only maintenance ends up published.
+#[mz_ore::test]
+fn maintenance_publishes_logging_index_others_do_not() {
+    for (role, expect_published) in [
+        (ComputeRuntimeRole::Maintenance, true),
+        (ComputeRuntimeRole::Interactive, false),
+        (ComputeRuntimeRole::Solo, false),
+    ] {
+        let id = GlobalId::System(1);
+        let registry = ArrangementSharingRegistry::new();
+        let registry_in = registry.clone();
+
+        timely::execute_directly(move |worker| {
+            worker.dataflow::<Timestamp, _, _>(|scope| {
+                let (mut oks_input, oks_collection) = scope.new_collection::<(Row, Row), Diff>();
+                let oks = oks_collection.mz_arrange::<
+                    ColumnationChunker<_>,
+                    RowRowBatcher<_, _>,
+                    RowRowBuilder<_, _>,
+                    RowRowSpine<_, _>,
+                >("test log oks");
+
+                let (mut errs_input, errs_collection) =
+                    scope.new_collection::<DataflowErrorSer, Diff>();
+                let errs = KeyCollection::from(errs_collection).mz_arrange::<
+                    ColumnationChunker<_>,
+                    ErrBatcher<_, _>,
+                    ErrBuilder<_, _>,
+                    ErrSpine<_, _>,
+                >("test log errs");
+
+                publish_logging_index(
+                    role,
+                    &registry_in,
+                    &scope.clone(),
+                    id,
+                    &oks.trace,
+                    &errs.trace,
+                );
+
+                oks_input.advance_to(Timestamp::from(1_u64));
+                oks_input.flush();
+                errs_input.advance_to(Timestamp::from(1_u64));
+                errs_input.flush();
+            });
+        });
+
+        assert_eq!(
+            registry.handles(&id, 0).is_some(),
+            expect_published,
+            "role {role:?} publication mismatch"
+        );
+    }
+}