clusterd-test-driver: headless CTP harness over clusterd

New crate providing an alternate frontend to `clusterd` that stands in for
environmentd's controller in compute tests: it hosts persist PubSub, connects the
compute CTP channel, and drives dataflows directly, with no SQL layer, catalog,
or timestamp oracle.

This first piece is the harness library: connection-target resolution, the CTP
connect + `Hello` handshake, persist-PubSub hosting, synthetic-row and direct
persist writes, response demultiplexing (frontiers + peeks), the `Driver`
(submit/schedule/await-frontier/peek), and `DataflowBuilder`, which lowers
caller-supplied MIR to a shippable `RenderPlan` dataflow (optionally running the
MIR optimizer). Exercised end-to-end by `tests/index_smoke.rs` against a real
clusterd.

The text script runner that drives this harness follows in a stacked PR.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>

Author: antiguru

Cargo.lock

@@ -25,6 +25,7 @@ members = [
     "src/cluster",
     "src/cluster-client",
     "src/clusterd",
+    "src/clusterd-test-driver",
     "src/compute",
     "src/compute-client",
     "src/compute-types",
@@ -154,6 +155,7 @@ default-members = [
     "src/cluster",
     "src/cluster-client",
     "src/clusterd",
+    "src/clusterd-test-driver",
     "src/compute",
     "src/compute-client",
     "src/compute-types",

Cargo.toml

@@ -0,0 +1,50 @@
+[package]
+name = "mz-clusterd-test-driver"
+description = "Headless frontend to clusterd for scripted compute tests."
+version = "0.0.0"
+edition.workspace = true
+rust-version.workspace = true
+publish = false
+
+[lints]
+workspace = true
+
+[dependencies]
+anyhow.workspace = true
+futures.workspace = true
+mz-compute-client = { path = "../compute-client" }
+mz-compute-types = { path = "../compute-types" }
+mz-dyncfg = { path = "../dyncfg" }
+mz-expr = { path = "../expr" }
+# default-features = false to avoid pulling mz-ore's default `tokio-console`
+# feature (pulled transitively via mz-compute-client); the harness only needs the
+# async, tracing, and metrics helpers.
+mz-ore = { path = "../ore", default-features = false, features = [
+    "async",
+    "tracing",
+    "metrics",
+] }
+mz-persist-client = { path = "../persist-client" }
+mz-persist-types = { path = "../persist-types" }
+mz-repr = { path = "../repr" }
+mz-service = { path = "../service" }
+mz-storage-types = { path = "../storage-types" }
+# Optional MIR dataflow optimizer, run only by `DataflowBuilder::optimize`. Needed
+# for shapes that don't lower from raw MIR (notably joins, whose `implementation`
+# defaults to `Unimplemented`); the default path lowers faithfully without it.
+mz-transform = { path = "../transform" }
+timely.workspace = true
+tokio = { workspace = true, features = ["full"] }
+tokio-stream.workspace = true
+tracing.workspace = true
+uuid.workspace = true
+
+[dev-dependencies]
+# The `test` feature provides the `#[mz_ore::test]` macro; it is only needed
+# when building tests, so keep it out of the non-test feature set (see the
+# [dependencies] mz-ore comment).
+mz-ore = { path = "../ore", features = ["test"] }
+tempfile.workspace = true
+
+[package.metadata.cargo-udeps.ignore]
+normal = []

doc/developer/design/20260612_headless_clusterd_test_driver.md

@@ -0,0 +1,77 @@
+// 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.
+
+//! Compute CTP connection and the `Hello` step of the controller handshake.
+//! Generic: it sends any `ComputeCommand` and receives any `ComputeResponse`.
+//!
+//! Only `Hello` (the transport/version step) happens here; the controller
+//! handshake proper — `CreateInstance`, `UpdateConfiguration`,
+//! `InitializationComplete` — is driven by explicit caller commands, so the
+//! caller controls the instance config, the peek-stash setting, and exactly when
+//! the reconciliation window opens and closes.
+
+use std::time::Duration;
+
+use mz_compute_client::protocol::command::ComputeCommand;
+use mz_compute_client::protocol::response::ComputeResponse;
+use mz_service::client::GenericClient;
+use mz_service::transport::{Client, NoopMetrics};
+use uuid::Uuid;
+
+pub type ComputeCtpClient = Client<ComputeCommand, ComputeResponse>;
+
+/// Connects to a clusterd compute controller address and sends `Hello`, leaving
+/// the controller handshake (`CreateInstance` onward) to the caller.
+///
+/// A reconnect re-runs exactly this: a fresh transport connection plus `Hello`.
+/// The reconciliation window then opens when the script sends `CreateInstance`
+/// and closes when it sends `InitializationComplete`; in between, the replica
+/// reconciles the replayed dataflows against its live ones rather than
+/// rehydrating.
+pub async fn connect_and_hello(compute_addr: &str) -> anyhow::Result<ComputeCtpClient> {
+    // Use persist-client's BUILD_INFO: it is release-versioned (synced by
+    // bin/bump-version), so it matches the clusterd we connect to. Our own
+    // crate is `0.0.0`, which would fail the handshake's version check.
+    let version = mz_persist_client::BUILD_INFO.semver_version();
+    let mut client = Client::<ComputeCommand, ComputeResponse>::connect(
+        compute_addr,
+        version,
+        Duration::from_secs(30),
+        Duration::from_secs(60),
+        NoopMetrics,
+    )
+    .await?;
+
+    client
+        .send(ComputeCommand::Hello {
+            nonce: Uuid::new_v4(),
+        })
+        .await?;
+
+    Ok(client)
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use crate::target;
+
+    #[mz_ore::test(tokio::test)]
+    #[cfg_attr(miri, ignore)]
+    async fn hello_holds_connection() {
+        if !target::e2e_enabled() {
+            return;
+        }
+        let mut client = connect_and_hello(&target::compute_addr())
+            .await
+            .expect("hello");
+        let r = tokio::time::timeout(Duration::from_millis(500), client.recv()).await;
+        assert!(r.is_err() || r.unwrap().is_ok());
+    }
+}