adapter: address review comments on 0dt migrated-MV hydration

- Seed the caught-up gate's transitive-dependent walk from all of
  `new_builtin_collections`, not just the MVs, and require the write
  frontier to be within the allowed lag of `now` on the
  no-live-frontier path so a hydrated-then-stalled collection no longer
  passes the gate forever.
- Document the `enable_0dt_hydrate_migrated_builtin_mvs` break-glass
  flag: it is read once at startup, so flipping it means changing the
  setting on the leader and restarting, and it is not an exact revert.
- Correct the read-only write-enable comments to match the code: the
  `allow_writes_in_read_only` tripwire blocks promotion, exclusive
  shard ownership holds per (build version, deploy generation), a new
  builtin MV cannot be write-enabled, and cite PR #35402 for the
  v26.17 leader floor.

Author: mtabebe

doc/developer/design/20251015_builtin_schema_migration.md

@@ -81,6 +81,11 @@ Because this environment exclusively owns the replacement shard, the read-only p
 This lets a migrated builtin materialized view and its dependents hydrate before cut-over instead of all at once at cut-over.
 It is safe only for the self-owned replacement shard, never a shard the leader is still serving, which is why it applies to shard replacement and not schema evolution.
 
+The force-write is conditional on two things.
+First, the leader must be at v26.17 or later, because every builtin materialized view reads the catalog shard and only leaders from that version on keep its frontier advancing with the current time; against an older leader the dataflow would sit at a stale frontier.
+Second, the `enable_0dt_hydrate_migrated_builtin_mvs` feature flag must be on; it exists as a break-glass revert.
+When either condition does not hold, the migrated materialized views and their dependents are instead excluded from the 0dt caught-up check, which is the older behaviour: promotion proceeds without them and they hydrate at cut-over.
+
 A leader process performing shard replacement performs the same steps as in read-only mode.
 Additionally, it cleans up durable state written by earlier versions and/or deploy generations by:
   - arranging for the previous shards used by the migrated storage collections to be finalized

src/adapter-types/src/dyncfgs.rs

@@ -95,9 +95,11 @@ pub const ENABLE_0DT_HYDRATE_MIGRATED_BUILTIN_MVS: Config<bool> = Config::new(
     true,
     "Write-enable replacement-migrated builtin materialized views while read-only during a 0dt \
      deployment, so they hydrate before cut-over and keep gating promotion. Emergency break-glass \
-     flag: disabling reverts to excluding migrated MVs (and their dependents) from the caught-up \
-     check, which lets promotion proceed with them unhydrated and hydrate at cut-over instead. \
-     Only takes effect when the leader is new enough for the write to make progress.",
+     flag: disabling excludes migrated MVs (and their dependents) from the caught-up check again, \
+     so promotion proceeds with them unhydrated. Not an exact revert: a collection with no live \
+     leader frontier must be hydrated either way. Only takes effect when the leader is new enough \
+     for the write to make progress, and is read once at startup, so changing it means setting it \
+     on the leader and restarting the new deployment.",
 );
 
 /// Enable logging of statement lifecycle events in mz_internal.mz_statement_lifecycle_history.

src/adapter/src/coord.rs

@@ -250,10 +250,10 @@ mod validity;
 ///
 /// Every builtin materialized view reads `mz_internal.mz_catalog_raw`, so its dataflow only makes
 /// progress up to the catalog shard's frontier. Holding that frontier at the current time is the
-/// leader's job, and leaders only started doing it in v26.17. Write-enable such an MV against an
-/// older leader and it sits at a stale frontier and never reports caught up, which blocks
-/// promotion outright instead of merely leaving the collection cold at cut-over. We still support
-/// upgrading from before v26.17, so that leader is a real case, not a hypothetical.
+/// leader's job, and leaders only started doing it in v26.17 (PR #35402). Write-enable such an MV
+/// against an older leader and it sits at a stale frontier and never reports caught up, which
+/// blocks promotion outright instead of merely leaving the collection cold at cut-over. We still
+/// support upgrading from before v26.17, so that leader is a real case, not a hypothetical.
 const MIN_LEADER_VERSION_FOR_MIGRATED_MV_WRITES: Version = Version::new(26, 17, 0);
 
 /// A pool of pre-allocated user IDs to avoid per-DDL persist writes.
@@ -2799,8 +2799,9 @@ impl Coordinator {
                     self.ship_dataflow(df_desc, mview.cluster_id, mview.target_replica)
                         .await;
 
-                    // If this is a replacement MV, it must remain read-only until the replacement
-                    // gets applied.
+                    // A pending `REPLACEMENT FOR` MV must stay read-only until
+                    // `ALTER ... APPLY REPLACEMENT` swaps it in. Unrelated to the
+                    // builtin-migration `Replacement` mechanism below.
                     if mview.replacement_target.is_none() {
                         let gid = mview.global_id_writes();
                         if hydrate_migrated_mvs
@@ -2811,6 +2812,10 @@ impl Coordinator {
                             // writing it while read-only hydrates the MV and its dependents before
                             // cut-over. An `Evolution`-migrated MV reuses the leader's live shard
                             // and must never reach here.
+                            //
+                            // A *new* builtin MV gets no such treatment: its shard allocation
+                            // lives only in this read-only savepoint, so the promoted leader
+                            // allocates a different shard and discards whatever we wrote.
                             self.controller
                                 .compute
                                 .allow_writes_in_read_only(mview.cluster_id, gid)
@@ -5187,32 +5192,31 @@ pub fn serve(
 
                 // A collection that can't advance its write frontier in read-only mode
                 // stalls its transitive dependents too, so exclude those from the caught-up
-                // check as well. That's new builtin MVs, whose fresh shard has no writer until
-                // this deployment promotes, plus migrated MVs whenever the leader is too old for
-                // them to write. An excluded dependent may still be hydrating right after
-                // promotion, a brief blip we accept because these MVs are small and get a writer
-                // at cut-over.
+                // check as well. That's every *new* builtin collection, whose fresh shard has no
+                // writer until this deployment promotes, plus migrated MVs whenever the leader is
+                // too old for them to write. An excluded dependent may still be hydrating right
+                // after promotion, a brief blip we accept because these collections are small and
+                // get a writer at cut-over.
                 //
-                // A migrated builtin *table* needs no such treatment even though a builtin MV can
-                // read one (`mz_clusters` joins `mz_cluster_replica_size_internal`):
-                // `read_only_mode_table_worker` keeps advancing the uppers of migrated tables, so
-                // an MV over one still catches up.
-                let new_builtin_mvs = new_builtin_collections
-                    .iter()
-                    .map(|global_id| {
-                        catalog
-                            .state()
-                            .try_get_entry_by_global_id(global_id)
-                            .expect("new builtin collections have catalog entries")
-                    })
-                    .filter(|entry| entry.is_materialized_view())
-                    .map(|entry| entry.id());
+                // Seeded from all of `new_builtin_collections`, not just the MVs: a new builtin
+                // table or source has no read-only writer either (`register_table_collections`
+                // retains only *migrated* tables), so an MV reading one never advances past its
+                // empty frontier. A *migrated* table is the opposite case, even though a builtin
+                // MV can read one (`mz_clusters` joins `mz_cluster_replica_size_internal`):
+                // `read_only_mode_table_worker` keeps advancing migrated tables' uppers.
+                let new_builtin_items = new_builtin_collections.iter().map(|global_id| {
+                    catalog
+                        .state()
+                        .try_get_entry_by_global_id(global_id)
+                        .expect("new builtin collections have catalog entries")
+                        .id()
+                });
                 let frozen_migrated_mvs = migrated_storage_collections_0dt
                     .iter()
                     .copied()
                     .filter(|_| !hydrate_migrated_mvs)
                     .filter(|id| catalog.state().get_entry(id).is_materialized_view());
-                let mut todo: Vec<_> = new_builtin_mvs.chain(frozen_migrated_mvs).collect();
+                let mut todo: Vec<_> = new_builtin_items.chain(frozen_migrated_mvs).collect();
                 while let Some(item_id) = todo.pop() {
                     let entry = catalog.state().get_entry(&item_id);
                     exclude_collections.extend(entry.global_ids());

src/adapter/src/coord/caught_up.rs

@@ -271,6 +271,11 @@ impl Coordinator {
         // `snapshot_latest` requires that the collection consolidates to a
         // set. `mz_cluster_replica_frontiers` is a controller-managed builtin
         // written with ±1 diffs, so it satisfies that invariant.
+        //
+        // NOTE: these are the leader's frontiers only because we read the leader's shard. A
+        // release that `Replacement`-migrates `mz_cluster_replica_frontiers` itself, or a test
+        // forcing replacement across all builtins, hands us a shard we write ourselves, and the
+        // lag check below then compares this deployment against itself.
         let live_frontiers = self
             .controller
             .storage_collections
@@ -634,12 +639,37 @@ impl Coordinator {
                             self.controller.storage.collection_hydrated(id)?
                         }
                     };
+
+                    // Also require the frontier to be within the allowed lag, the bound the
+                    // live-frontier path applies, with `now` standing in for the missing live
+                    // frontier. Hydration is one-shot: a collection that hydrated and then stalled
+                    // would otherwise satisfy this branch forever.
+                    //
+                    // NOTE: there is deliberately no `cutoff` escape hatch here. A frontier frozen
+                    // at the minimum is exactly what this gate must catch, so a collection stuck
+                    // here blocks promotion until `with_0dt_deployment_max_wait` elapses.
+                    let write_frontier_plus_allowed_lag = Antichain::from_iter(
+                        write_frontier
+                            .iter()
+                            .map(|t| t.step_forward_by(&allowed_lag)),
+                    );
+                    let within_lag = PartialOrder::less_equal(
+                        &Antichain::from_elem(now),
+                        &write_frontier_plus_allowed_lag,
+                    );
+
                     tracing::info!(
                         ?write_frontier,
                         %collection_hydrated,
+                        %within_lag,
+                        ?allowed_lag,
+                        ?now,
                         "collection {id} not in live frontiers"
                     );
-                    if write_frontier.less_equal(&Timestamp::minimum()) || !collection_hydrated {
+                    if write_frontier.less_equal(&Timestamp::minimum())
+                        || !collection_hydrated
+                        || !within_lag
+                    {
                         all_caught_up = false;
                     }
                     continue;

src/compute-client/src/controller.rs

@@ -1183,26 +1183,40 @@ impl ComputeController {
 
     /// Like [`Self::allow_writes`], but takes effect even in read-only mode.
     ///
-    /// The caller must guarantee that no other environment writes the collection's output shard.
+    /// The caller must guarantee that no leader environment writes the collection's output shard.
     /// In a 0dt deployment that means a shard this environment created for itself, the replacement
     /// shard of a `Replacement`-migrated builtin collection, never one the leader is still serving
     /// from. `Evolution` migrates in place and reuses the leader's shard, so it must not reach this
     /// path. Violating the guarantee races two writers on one shard.
     ///
+    /// NOTE: ownership is exclusive per (build version, deploy generation), not per process: the
+    /// migration shard entry naming the shard is keyed by that pair and a read-only catalog open is
+    /// a savepoint, so two read-only processes of one generation both write it. Same shape as a
+    /// multi-replica materialized view, which the self-correcting persist sink tolerates (see the
+    /// `mz_compute::sink::materialized_view` module docs).
+    ///
     /// This is the compute-side counterpart to the storage controller's `force_writable` handling
     /// of migrated storage collections. Migrated builtin tables are storage collections that
     /// storage force-writes read-only; migrated builtin MVs are compute collections that only this
     /// path can force-write. Both rest on the same guarantee (this environment exclusively owns the
     /// replacement shard) but run on separate write paths, so each needs its own bypass.
+    ///
+    /// NOTE: the replica-side handler enables persist compaction process-wide on the clusterd
+    /// (`ComputeState::handle_allow_writes`), which this path is the first to trigger inside a
+    /// read-only deployment.
     pub fn allow_writes_in_read_only(
         &mut self,
         instance_id: ComputeInstanceId,
         collection_id: GlobalId,
     ) -> Result<(), CollectionUpdateError> {
-        // Every builtin eligible for the read-only bypass has a system id. A non-system id means
-        // the caller's `Replacement`-only invariant broke, so degrade to the read-only no-op (cold
-        // collection at cut-over) rather than risk writing a shard the leader still serves. Mirrors
-        // the storage-side tripwire in `StorageController::register_introspection_collection`.
+        // Every builtin eligible for this bypass has a system id, so a non-system id means the
+        // caller's `Replacement`-only invariant broke. No-op rather than risk writing a shard the
+        // leader still serves. The collection then sits in the caught-up gate on an unwritten
+        // shard and blocks promotion, which is the loud, safe direction to fail.
+        //
+        // Storage asserts the same invariant hard, in
+        // `StorageController::register_introspection_collection`. The asymmetry is deliberate: a
+        // soft panic keeps a caller bug visible in CI and Sentry without downing production.
         if self.read_only && !collection_id.is_system() {
             soft_panic_or_log!(
                 "allow_writes_in_read_only called for non-system collection {collection_id}; \