doc: design decoupled coordination

doc/developer/design/20260903_decoupled_coordination_prompt.md

@@ -0,0 +1,103 @@
+# Decoupled coordination: implementer session prompt
+
+```text
+Work on doc/developer/design/20260903_decoupled_coordination.md.
+
+Working PR: https://github.com/MaterializeInc/materialize/pull/38696
+Bookmark: decoupled-coordination
+Remote: origin, pointing to the contributor fork, not upstream
+
+Read the design and its implementation log, then inspect the current code,
+worktree, and remote bookmark. Preserve existing work and account for progress
+from other sessions. Use the latest handoff and current code to identify the
+active milestone in the design's Implementation and verification section. Choose
+the next coherent piece toward its observable outcome, briefly explain that
+choice, then implement and verify it. Historical proposals and next steps are
+context, not a cumulative task list.
+
+Current steering
+
+Check which of these review findings remain unresolved, then choose one coherent
+change. Remove resolved steering from this prompt. These are implementation
+priorities, not new design requirements or a reason to reopen agreed decisions.
+
+- Maintained protection: milestone 1 remains active. The protected-MV path is an
+  implementation checkpoint, not completion across maintained object types. Make
+  source/sink recovery requirements constrain catalog authorization even when
+  local controller accounting is absent, and bring maintained compute compaction
+  under committed permission. Follow the milestone's requirement-derivation and
+  recovery-semantics boundaries rather than prescribing new records for every type.
+- Publication scaling: measure publication cost at representative collection
+  counts, including DDL latency and retained history. Batched protection staging
+  removes the per-record pending-update scans, but does not establish end-to-end
+  scalability. Do not infer scalability from the small recovery demonstration or
+  make a general catalog redesign a prerequisite.
+- Ownership transition: after milestone 1, prioritize a real catalog subscriber
+  over more standalone APIs. In that transition, address prepare_state's reliance
+  on locally installed collections and make the writer-side responsibility for
+  complete maintained requirements explicit. An MV's requirement is a separate
+  sequencer op today. These are transitional dependencies, not evidence that the
+  current single-owner path fails.
+
+Keep the draft PR description accurate about what is implemented and what remains,
+with validation status in the PR rather than the design log.
+
+Prefer connecting existing pieces through the active milestone's production path
+over adding further standalone APIs. Preparatory work is appropriate when it
+unblocks that path. Let integration evidence refine intermediate interfaces
+rather than adding machinery to preserve them.
+
+Treat the design as the agreed boundaries, not a prescribed mechanism. Prefer
+the smallest coherent solution that preserves the full capability. Incremental
+progress is fine, but do not mistake an intermediate step for completion.
+Implementation choices within the agreed boundaries do not require renewed
+design approval.
+
+Bring discoveries, consequential tradeoffs, and scope growth to me, Aljoscha.
+Pause affected work when guidance is needed rather than silently narrowing
+scope, adding machinery, or changing an agreed boundary.
+
+Follow repository instructions and skills. Verify at the changed boundaries,
+and seek independent review when the risk warrants it.
+
+Use the draft PR's CI as the default test loop instead of running test suites
+locally. You may push changes and iterate on CI failures. Keep cheap local
+formatting and checks, and run targeted local tests when useful, not as a
+prerequisite for pushing. Follow the mz-debug-ci skill when investigating CI.
+Report pending or failed checks explicitly rather than treating a push as
+successful validation.
+
+Locally, `bin/fmt`, `cargo check`, and the Rust parts of `bin/lint`
+(check-cargo, check-formatting, check-python-docs) work. `bin/lint` also runs
+checks whose tools are not installed here (npm, helm-docs, trufflehog, zizmor);
+CI is the authority for those. Do not record local tooling gaps anywhere.
+
+Focus on regular PR CI for now. Nightly intentionally does not run on this
+draft PR, so do not treat its absence as a blocker or try to enable it. We will
+start nightly validation once we have a working implementation.
+
+Normally finish the session with one coherent change. Give it a clear commit
+message and change description explaining the outcome, rationale, and validation
+status, not the chronology of attempts. If blocked, report the blocker rather
+than claiming completion.
+
+Append only a minimal dated handoff to the design's log: consequential findings
+or decisions, unresolved questions, and the next useful step. Do not record
+validation status there at all: CI results, pending checks, formatting or
+compile checks, tool availability, and review outcomes are reconstructible from
+the PR and are noise in the log. Distinguish proposals from decisions we
+reviewed together. Do not rewrite earlier entries. Keep the main design focused
+on design, and change its agreed boundaries only after discussing them with me.
+
+You may commit and push progress to this bookmark without asking again.
+Prefer jj. In-progress commits and pushes are allowed while iterating. Before
+finishing, squash your session's intermediate commits, including fixes and log
+updates, into one coherent commit.
+You may rewrite and repush your own session's work-in-progress commits for
+this purpose. Check for remote changes before pushing. Do not overwrite
+others' work, and ask before rewriting other sessions' commits or commits
+that others have built on. Report validation failures honestly.
+
+Keep the existing PR as a draft. Do not merge, mark it ready, or push to
+upstream or other branches without asking.
+```

misc/python/materialize/mzcompose/__init__.py

@@ -181,6 +181,9 @@ def get_minimal_system_parameters(
     if version < MzVersion.parse_mz("v26.41.0-dev"):
         config["enable_zero_downtime_cluster_reconfiguration"] = "true"
 
+    if version >= MzVersion.parse_mz("v26.41.0-dev"):
+        config["enable_catalog_read_protection"] = "true"
+
     return config
 
 
@@ -683,6 +686,8 @@ def get_default_system_parameters(
     "memory_limiter_usage_bias",
     "memory_limiter_burst_factor",
     "catalog_info_metrics_reconcile_interval",
+    # Zero pauses publication, so vary this only in targeted consistency checks.
+    "catalog_read_protection_publish_interval",
     "compute_server_maintenance_interval",
     "compute_dataflow_max_inflight_bytes_cc",
     "compute_flat_map_fuel",

misc/python/materialize/parallel_workload/action.py

@@ -3233,6 +3233,8 @@ def __init__(
         # behavior, you should add it. Feature flags which turn on/off
         # externally visible features should not be flipped.
         self.uninteresting_flags: list[str] = [
+            # Latched at fresh catalog initialization, not changed by ALTER SYSTEM.
+            "enable_catalog_read_protection",
             # Read once at environmentd startup, so an ALTER SYSTEM SET only
             # takes effect after a restart. Flipping it here would be a no-op
             # for the running process.

src/adapter/benches/catalog.rs

@@ -7,11 +7,19 @@
 // the Business Source License, use of this software will be governed
 // by the Apache License, Version 2.0.
 
-use criterion::{Criterion, criterion_group, criterion_main};
+use std::collections::BTreeSet;
+use std::time::{Duration, Instant};
+
+use criterion::{BenchmarkId, Criterion, Throughput, criterion_group, criterion_main};
 use mz_adapter::catalog::{Catalog, Op};
-use mz_catalog::durable::test_bootstrap_args;
-use mz_persist_client::PersistClient;
+use mz_catalog::durable::objects::{CollectionCompactionBound, MaintainedReadRequirement};
+use mz_catalog::durable::{TestCatalogStateBuilder, test_bootstrap_args};
+use mz_ore::now::SYSTEM_TIME;
+use mz_persist_client::{PersistClient, ShardId};
+use mz_repr::{GlobalId, Timestamp};
 use mz_sql::session::user::MZ_SYSTEM_ROLE_ID;
+use mz_storage_client::controller::StorageTxn;
+use timely::progress::Antichain;
 use tokio::runtime::Runtime;
 use uuid::Uuid;
 
@@ -47,5 +55,137 @@ fn bench_transact(c: &mut Criterion) {
     });
 }
 
-criterion_group!(benches, bench_transact);
+fn bench_read_protection(c: &mut Criterion) {
+    let mut group = c.benchmark_group("catalog_read_protection");
+    for count in [100_u64, 1_000, 10_000] {
+        group.throughput(Throughput::Elements(count + count / 2));
+        for batched in [false, true] {
+            let mode = if batched { "batch" } else { "scalar" };
+            group.bench_with_input(BenchmarkId::new(mode, count), &count, |b, &count| {
+                let runtime = Runtime::new().unwrap();
+                // Count is total collections, split into disjoint input/output pairs.
+                let pairs: Vec<_> = (0..count / 2)
+                    .map(|i| (GlobalId::User(2 * i + 1), GlobalId::User(2 * i + 2)))
+                    .collect();
+                let records = |frontier| {
+                    let requirements = pairs
+                        .iter()
+                        .map(|&(input, output)| MaintainedReadRequirement {
+                            id: output,
+                            inputs: BTreeSet::from([input]),
+                            frontier: Some(frontier),
+                        })
+                        .collect();
+                    let bounds = pairs
+                        .iter()
+                        .flat_map(|&(input, output)| [input, output])
+                        .map(|id| CollectionCompactionBound {
+                            id,
+                            frontier: Some(frontier),
+                        })
+                        .collect();
+                    (requirements, bounds)
+                };
+                // Seed catalog metadata only. This isolates transaction processing
+                // from controller work, physical retention, and concurrent DDL.
+                let mut catalog = runtime.block_on(async {
+                    let persist_client = PersistClient::new_for_tests().await;
+                    let organization_id = Uuid::new_v4();
+                    let bootstrap_args = test_bootstrap_args();
+                    let mut storage = TestCatalogStateBuilder::new(persist_client.clone())
+                        .with_organization_id(organization_id)
+                        .with_default_deploy_generation()
+                        .build()
+                        .await
+                        .unwrap()
+                        .open(SYSTEM_TIME().into(), &bootstrap_args)
+                        .await
+                        .unwrap();
+                    // Acknowledge setup updates before transacting. The debug catalog
+                    // reconstructs in-memory state from the committed seed below.
+                    let _ = storage.sync_to_current_updates().await.unwrap();
+                    let mut tx = storage.transaction().await.unwrap();
+                    tx.insert_collection_metadata(
+                        pairs
+                            .iter()
+                            .flat_map(|&(input, output)| [input, output])
+                            .map(|id| (id, ShardId::new()))
+                            .collect(),
+                    )
+                    .unwrap();
+                    let (requirements, bounds) = records(Timestamp::new(0));
+                    tx.set_read_protection(requirements, bounds).unwrap();
+                    let _ = tx.get_and_commit_op_updates();
+                    let commit_ts = tx.upper();
+                    tx.commit(commit_ts).await.unwrap();
+                    storage.expire().await;
+                    Catalog::open_debug_catalog(persist_client, organization_id, &bootstrap_args)
+                        .await
+                        .unwrap()
+                });
+                let mut frontier = 0_u64;
+                // Only Catalog::transact is timed, not record construction or upper lookup.
+                // Updating the same keys keeps cardinality fixed across warmup and samples.
+                b.iter_custom(|iterations| {
+                    runtime.block_on(async {
+                        let mut elapsed = Duration::ZERO;
+                        for _ in 0..iterations {
+                            frontier = frontier.checked_add(1).unwrap();
+                            let (requirements, bounds) = records(Timestamp::new(frontier));
+                            let ops = if batched {
+                                vec![Op::SetReadProtection {
+                                    requirements,
+                                    bounds,
+                                }]
+                            } else {
+                                requirements
+                                    .into_iter()
+                                    .map(|requirement| Op::SetMaintainedReadRequirement {
+                                        id: requirement.id,
+                                        inputs: requirement.inputs,
+                                        frontier: requirement.frontier,
+                                    })
+                                    .chain(bounds.into_iter().map(|bound| {
+                                        Op::SetCollectionCompactionBound {
+                                            id: bound.id,
+                                            frontier: bound.frontier,
+                                        }
+                                    }))
+                                    .collect()
+                            };
+                            let commit_ts = catalog.current_upper().await;
+                            let start = Instant::now();
+                            let result = catalog.transact(None, commit_ts, None, ops).await;
+                            elapsed += start.elapsed();
+                            result.unwrap();
+                        }
+                        elapsed
+                    })
+                });
+                let state = catalog.state();
+                assert_eq!(state.maintained_read_requirements().len(), pairs.len());
+                assert_eq!(
+                    state.storage_metadata().compaction_bounds.len(),
+                    2 * pairs.len()
+                );
+                for &(input, output) in &pairs {
+                    assert_eq!(
+                        state.maintained_read_requirements()[&output].frontier,
+                        Some(Timestamp::new(frontier))
+                    );
+                    for id in [input, output] {
+                        assert_eq!(
+                            state.storage_metadata().compaction_bounds[&id],
+                            Antichain::from_elem(Timestamp::new(frontier))
+                        );
+                    }
+                }
+                runtime.block_on(catalog.expire());
+            });
+        }
+    }
+    group.finish();
+}
+
+criterion_group!(benches, bench_transact, bench_read_protection);
 criterion_main!(benches);