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 re-imports the shared traces under its own id and publishes those. That
import gives the re-export's dataflow operators, and `mz_compute_error_counts`
forwards a dependency's counts only to a re-export whose dataflow has none, so
the arm also logs the re-export's error counts from the imported errors. 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.

`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"
+        );
+    }
+}