compute: launch a second, interactive compute runtime

With `--interactive-compute-timely-config`, clusterd runs two compute runtimes in
one process: the first takes the `Maintenance` role, the second `Interactive`, and
both share the one sharing registry so a reader on worker `i` of either finds the
slot a publisher on worker `i` of the other filled. The two must span an equal
number of Timely peers, which the config preparation asserts, because the registry
pairs workers by ordinal and reads are sound only if both shard keys across the
same peer count. Without the flag the process runs a single `Solo` runtime and is
byte-unchanged from a deployment that has no second runtime.

One controller endpoint still fronts the replica. With two runtimes a
`Multiplexer` serves it, routing each command to the owning runtime and merging
responses; with one, the maintenance client builder serves it directly.

Shared fate is the read-hold mechanism. Both runtimes' worker and reader threads
are covered by the process-global panic hook installed at the top of `main`, so a
panic on either aborts the whole process. That bounds an interactive import's read
hold to the life of the replica without a lease, because there is no way for one
runtime to wedge while the other's holds continue. A subprocess test asserts the
abort, which cannot be observed from inside the panicking process.

`ClusterSpec::cluster_name` lets one process run two clusters of the same kind with
distinguishable tracing spans. The interactive runtime takes
`compute-interactive`; solo and maintenance keep the bare `compute` so
single-runtime logs are unchanged.

`enable_compute_interactive_runtime` is replica-scoped but resolved in
`environmentd`, because the controller decides `ServiceConfig::ports` before the
replica exists and so cannot read the value from the replica's `worker_config`.
`CatalogState::replica_scoped_bool` parses the override through the dyncfg rather
than `str::parse`, since a stored bool formats as `on`/`off`, which
`str::parse::<bool>()` rejects.

Off by default in production and in tests. Flipping it changes how a replica is
provisioned, so it is not a live toggle: a running replica keeps the layout it was
launched with.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>

Author: antiguru

Cargo.lock

@@ -649,6 +649,9 @@ def get_default_system_parameters(
 # all. Only add it in UNINTERESTING_SYSTEM_PARAMETERS if none of the above
 # apply.
 UNINTERESTING_SYSTEM_PARAMETERS = [
+    # Registered here rather than varied, because the interactive runtime cannot serve
+    # index peeks yet. Moves to get_variable_system_parameters once it can.
+    "enable_compute_interactive_runtime",
     "enable_compute_half_join2",
     "enable_mz_join_core",
     "linear_join_yielding",

misc/python/materialize/mzcompose/__init__.py

@@ -3094,6 +3094,8 @@ def __init__(
             BOOLEAN_FLAG_VALUES
         )
         self.flags_with_values["enable_upsert_v2"] = BOOLEAN_FLAG_VALUES
+        # Pinned off: the interactive runtime cannot serve index peeks yet.
+        self.flags_with_values["enable_compute_interactive_runtime"] = ["FALSE"]
         self.flags_with_values["enable_coalesce_case_transform"] = BOOLEAN_FLAG_VALUES
         self.flags_with_values["enable_any_all_null_array_semantics"] = (
             BOOLEAN_FLAG_VALUES

misc/python/materialize/parallel_workload/action.py

@@ -2527,6 +2527,45 @@ impl CatalogState {
         &self.scoped_system_parameters
     }
 
+    /// Resolves a boolean replica-local override for `replica_id`, falling back to `default` when
+    /// the replica has no override for `name`.
+    ///
+    /// For configs consumed when a replica is provisioned rather than on the replica itself. Those
+    /// cannot read the value from their own `worker_config`, because the decision is made in
+    /// `environmentd` before the replica exists.
+    ///
+    /// Parses through the dyncfg rather than `str::parse`, because a stored override is a var-format
+    /// string: `bool` values format as `on`/`off`, which `str::parse::<bool>()` rejects. Parsing it
+    /// the wrong way silently resolved `false` for an override every other surface reported as on.
+    pub fn replica_scoped_bool(&self, replica_id: ReplicaId, name: &str, default: bool) -> bool {
+        let Some(value) = self
+            .scoped_system_parameters
+            .replica
+            .get(&replica_id)
+            .and_then(|overrides| overrides.get(name))
+        else {
+            return default;
+        };
+        let dyncfgs = self.system_configuration.dyncfgs();
+        let parsed = dyncfgs
+            .entry(name)
+            .and_then(|entry| entry.parse_val(value).ok())
+            .and_then(|val| match val {
+                mz_dyncfg::ConfigVal::Bool(parsed) => Some(parsed),
+                _ => None,
+            });
+        match parsed {
+            Some(parsed) => parsed,
+            None => {
+                tracing::warn!(
+                    %name, %value, %replica_id,
+                    "cannot parse replica-scoped override, falling back to the environment value",
+                );
+                default
+            }
+        }
+    }
+
     /// Return a mutable reference to the current system configuration.
     pub fn system_config_mut(&mut self) -> &mut SystemVars {
         Arc::make_mut(&mut self.system_configuration)

src/adapter/src/catalog/state.rs

@@ -121,6 +121,7 @@ use mz_controller::clusters::{
     ClusterConfig, ClusterEvent, ClusterStatus, ProcessId, ReplicaLocation,
 };
 use mz_controller::{ControllerConfig, Readiness};
+use mz_controller_types::dyncfgs::ENABLE_COMPUTE_INTERACTIVE_RUNTIME;
 use mz_controller_types::{ClusterId, ReplicaId, WatchSetId};
 use mz_dyncfg::{ConfigUpdates, ParameterScope};
 use mz_expr::{MapFilterProject, MirRelationExpr, OptimizedMirRelationExpr, RowSetFinishing};
@@ -2581,6 +2582,12 @@ impl Coordinator {
             )?;
             for replica in instance.replicas() {
                 let role = instance.role();
+                let interactive_runtime = self.catalog().state().replica_scoped_bool(
+                    replica.replica_id,
+                    ENABLE_COMPUTE_INTERACTIVE_RUNTIME.name(),
+                    ENABLE_COMPUTE_INTERACTIVE_RUNTIME
+                        .get(self.catalog().system_config().dyncfgs()),
+                );
                 self.controller.create_replica(
                     instance.id,
                     replica.replica_id,
@@ -2590,6 +2597,7 @@ impl Coordinator {
                     replica.config.clone(),
                     enable_worker_core_affinity,
                     enable_storage_introspection_logs,
+                    interactive_runtime,
                 )?;
             }
         }

src/adapter/src/coord.rs

@@ -41,6 +41,7 @@ use mz_cloud_resources::VpcEndpointConfig;
 use mz_compute_client::logging::LogVariant;
 use mz_compute_client::protocol::response::{PeekError, PeekResponse};
 use mz_controller::clusters::{ClusterRole, ReplicaConfig};
+use mz_controller_types::dyncfgs::ENABLE_COMPUTE_INTERACTIVE_RUNTIME;
 use mz_controller_types::{ClusterId, ReplicaId};
 use mz_ore::collections::CollectionExt;
 use mz_ore::error::ErrorExt;
@@ -1675,6 +1676,12 @@ impl Coordinator {
         // configuration replays with them. Render-frozen flags make a later push
         // too late, which is why the push precedes `create_replica`.
 
+        let interactive_runtime = self.catalog().state().replica_scoped_bool(
+            replica_id,
+            ENABLE_COMPUTE_INTERACTIVE_RUNTIME.name(),
+            ENABLE_COMPUTE_INTERACTIVE_RUNTIME.get(self.catalog().system_config().dyncfgs()),
+        );
+
         self.controller
             .create_replica(
                 cluster_id,
@@ -1685,6 +1692,7 @@ impl Coordinator {
                 replica_config,
                 enable_worker_core_affinity,
                 enable_storage_introspection_logs,
+                interactive_runtime,
             )
             .expect("creating replicas must not fail");
 

src/adapter/src/coord/catalog_implications.rs

@@ -9,6 +9,7 @@
 
 //! An interactive cluster server.
 
+use std::borrow::Cow;
 use std::fmt;
 use std::sync::{Arc, Mutex};
 use std::thread::Thread;
@@ -167,6 +168,15 @@ pub trait ClusterSpec: Clone + Send + Sync + 'static {
     /// The name of this cluster ("compute" or "storage").
     const NAME: &str;
 
+    /// The name recorded on the per-worker Timely tracing span.
+    ///
+    /// Defaults to [`Self::NAME`]. A spec that runs more than one cluster of the same kind in a
+    /// process (for example the maintenance and interactive compute runtimes) overrides this to
+    /// keep their spans distinguishable in the logs.
+    fn cluster_name(&self) -> Cow<'static, str> {
+        Cow::Borrowed(Self::NAME)
+    }
+
     /// Run the given Timely worker.
     fn run_worker(
         &self,
@@ -257,11 +267,12 @@ pub trait ClusterSpec: Clone + Send + Sync + 'static {
         }
 
         let spec = self.clone();
+        let cluster_name = self.cluster_name();
         let worker_guards = execute_from(builders, other, worker_config, move |timely_worker| {
             let worker_idx = timely_worker.index();
 
             // Per worker tracing span, lets us identify Timely clusters and workers in the logs.
-            let span = info_span!("timely", name = Self::NAME, worker_id = worker_idx);
+            let span = info_span!("timely", name = %cluster_name, worker_id = worker_idx);
             let _span_guard = span.enter();
 
             // Every Timely instance in this process names its threads `timely:work-N`, restarting

src/cluster/src/client.rs

@@ -23,6 +23,7 @@ mz-build-info = { path = "../build-info" }
 mz-cloud-resources = { path = "../cloud-resources" }
 mz-cluster-client = { path = "../cluster-client" }
 mz-compute = { path = "../compute", default-features = false }
+mz-compute-client = { path = "../compute-client" }
 mz-dyncfgs = { path = "../dyncfgs" }
 mz-http-util = { path = "../http-util" }
 mz-metrics = { path = "../metrics" }