adapter: exclude dependents of every new builtin, and bound the weak caught-up path

Two gaps in the 0dt caught-up gate, both found in review of MaterializeInc#38399.

Seed the transitive-dependent walk from all of `new_builtin_collections`,
not just the materialized views among them. A brand-new builtin table or
source has no read-only writer either, since `register_table_collections`
retains only migrated tables while read-only, so a `Replacement`-migrated
MV reading one can never advance past that empty frontier. With migrated
MVs now kept in the gate, such an MV would block promotion until an
operator intervened. No builtin table or source has been added since
v26.38.0, so this is latent today.

Require the write frontier to be within the allowed lag of `now` on the
no-live-frontier path, the same bound the live-frontier path applies.
Hydration is a one-shot property, so a collection that hydrated and then
stalled satisfied that branch forever.

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

Author: mtabebe

src/adapter/src/coord.rs

@@ -5172,32 +5172,34 @@ 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
+                // The walk is seeded from all of `new_builtin_collections`, not just the MVs among
+                // them. A brand-new builtin table or source has no read-only writer either
+                // (`register_table_collections` retains only *migrated* tables while read-only),
+                // so an MV reading one can never advance past that empty frontier.
+                //
+                // A *migrated* builtin 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 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());
+                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

@@ -634,12 +634,42 @@ impl Coordinator {
                             self.controller.storage.collection_hydrated(id)?
                         }
                     };
+
+                    // Also require the write frontier to be within the allowed lag of `now`, the
+                    // same bound the live-frontier path applies. Hydration alone is a one-shot
+                    // property: a collection that hydrated and then stalled keeps satisfying the
+                    // gate forever. With no live frontier to compare against, `now` stands in for
+                    // it. An empty write frontier means the collection is done advancing, which
+                    // trivially satisfies this.
+                    //
+                    // NOTE: unlike the live-frontier path there is no `cutoff` escape hatch for a
+                    // collection that is hopelessly behind, and there deliberately cannot be: a
+                    // frontier frozen at the minimum is exactly the state this gate exists to
+                    // catch, so ignoring the worst offenders would defeat it. A collection stuck
+                    // here blocks promotion until `with_0dt_deployment_max_wait` elapses or an
+                    // operator intervenes.
+                    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;