compute: split hydration into dataflow and durable stages

The hydration lifecycle has four stages, installed, started, hydrated and
written, where written is the moment the output is durable in persist.
Indexes do not write, so for them the last two coincide. Materialized
views do, and the gap is the initial snapshot write.

hydrated_at was stamped off the reported output frontier, which is the
meet of the write and compute frontiers. That made it a measure of
durability rather than of computation, and worse, not uniform across
workers: the sink's mint operator maintains the shared sink frontier on
one elected worker and clears it on all the others, so the same dataflow
reported hydration at two different stages depending on which worker's
log a consumer read.

Both stages are wanted. A replacement materialized view runs in read-only
mode and does not write until a cutover, so folding durability into
hydrated_at would measure how long a human took to promote, where the
dataflow stage is the earliest moment the replacement could have been
promoted. Durability is what readiness needs, and readiness is served by
time_ns and the relations built on it.

So the signal is split rather than moved. The Hydration event still fires
on the output frontier crossing and still owns time_ns, which is what
keeps mz_compute_hydration_times, mz_compute_hydration_statuses,
mz_hydration_statuses and the blue-green readiness query reporting
exactly what they reported before. A new DataflowHydrated event fires on
the dataflow's own progress frontier and owns hydrated_at, along with the
started_at backfill, since that is now the handler responsible for the
installed_at <= started_at <= hydrated_at invariant.

report_frontiers derives the progress frontier as the compute probe where
one exists and the write frontier otherwise, so an index keeps stamping
from its trace upper and the index degeneracy falls out rather than being
special-cased. Subscribes report from their own frontier through the same
observer, unchanged.

Also drops the doc's claim that a REFRESH materialized view reports a
refresh interval rather than hydration work. REFRESH EVERY takes an
implicit refresh at creation, so its first write is prompt, and the claim
was wrong for that variant. What is verified, and kept, is that the
compute probe sits before apply_refresh deliberately, so hydration reads
the pre-rounding frontier.

Testdrive coverage asserts both workers report hydrated_at for a
multi-worker materialized view, that they agree closely, and that
hydration never postdates the durable moment time_ns measures. A small
view writes in milliseconds, so these lock the invariants rather than
proving the stages are separate.

Part of CPU-226

Co-Authored-By: Claude <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_018ZVCMBSLdxzGus78ZKWhZz

Author: claude

doc/developer/design/20260817_compute_hydration_timestamps.md

@@ -103,12 +103,26 @@ them.
 | --- | --- | --- |
 | `installed_at` | the `Export` event, from `CreateDataflow` | the dataflow exists on this worker, suspended, so this is also the start of queueing |
 | `started_at` | the dataflow is unsuspended | hydration is actually running |
-| `hydrated_at` | the output frontier passes the as-of | hydration is complete |
+| `hydrated_at` | the dataflow's own progress frontier passes the as-of | the dataflow has computed its output |
 
 `hydrated_at - started_at` is hydration time as users mean it, and
 `started_at - installed_at` is the queueing interval. Today's `time_ns` conflates
 the two.
 
+The full lifecycle has a fourth stage, `installed` to `started` to `hydrated` to
+`written`, where `written` is the moment the output is durable in persist. Indexes
+do not write, so for them the last two coincide. Materialized views do, and the
+gap is the initial snapshot write.
+
+Only the first three are stamped here. `written` is tracked separately, and the
+reason it is a distinct stage rather than the same one is that a replacement
+materialized view runs in read-only mode and does not write until a cutover, so
+folding durability into `hydrated_at` would measure how long a human took to
+promote. `hydrated_at` is the earliest moment a replacement could have been
+promoted, which is the actionable number. Readiness is the question durability
+answers, and that is served by `time_ns` and the relations built on it, which this
+design leaves alone.
+
 Two choices shape everything else, each argued in its own section below. The
 timestamps are stamped by the replica rather than by the compute controller, for
 the reasons in "Why the replica and not the compute controller". And `time_ns` is
@@ -163,6 +177,27 @@ compute logging and has not been observed to be severe, but it is a real risk an
 this design is the first to invite direct comparison of absolute times, so it is
 acknowledged rather than designed around.
 
+### Which frontier hydration reads
+
+Hydration is driven by the dataflow's own progress frontier, not by the reported
+output frontier. The output frontier is the meet of write and compute frontier,
+which makes it a measure of durability rather than of computation, and for a
+sink-backed collection it is not even uniform across workers: the sink's `mint`
+operator maintains the shared sink frontier on one elected worker and clears it on
+all the others, so the same dataflow would report hydration at two different times
+depending on which worker's log a consumer read.
+
+So a collection with a compute frontier hydrates on that frontier. A collection
+without one, an index into its own trace, produces its output by writing it, so
+there the write frontier is the progress and hydration coincides with durability.
+That is the index degeneracy of the lifecycle, and it falls out rather than being
+special-cased.
+
+The reported output frontier is unchanged and still the meet. The controller needs
+durability for its caught-up and autoscaling checks, and `time_ns` still fires on
+the meet crossing, which is what keeps every relation built on it reporting
+exactly what it reported before.
+
 ### A new hydration start event
 
 There is no event for hydration start today. Add
@@ -274,9 +309,12 @@ change nor the rename would have touched that relation.
 **`time_ns` is kept rather than replaced.** It is the reason the existing columns
 keep their exact values: retained rather than derived, so nothing is recomputed,
 no precision is lost, and no cross-worker arithmetic is introduced.
-Deriving `time_ns` as `hydrated_at - installed_at` would have moved it to
-microsecond precision, since `timestamptz` caps there, where today it is true
-nanoseconds. Deriving it after aggregation would additionally have absorbed
+It could not be derived from the timestamps in any case. `time_ns` runs to the
+output frontier crossing and `hydrated_at` to the dataflow's own, which for a
+materialized view are different stages separated by the snapshot write, so
+`hydrated_at - installed_at` is not the same interval. Even for an index, where the
+two coincide, deriving it would have moved it to microsecond precision, since
+`timestamptz` caps there, where today it is true nanoseconds. Deriving it after aggregation would additionally have absorbed
 cross-worker install skew and the per-worker anchor skew described above. So
 `time_ns` remains the authoritative per-worker duration, measured from a single
 `Instant` inside one worker, and the timestamps carry episode identity, which
@@ -312,7 +350,7 @@ in one controller turn and one replica turn.
 | 6 | replica | inserts the suspension token and renders the dataflow, whose operators park on the `StartSignal` | |
 | 7 | replica | `handle_schedule` drops the token and the operators start | **`started_at`** |
 | 8 | replica | the dataflow reads its inputs from the as-of forward and builds arrangements. Nothing is stamped here, this interval is the hydration | |
-| 9 | replica | the output frontier passes the as-of and `set_reported_output_frontier` calls `set_hydrated` | **`hydrated_at`** |
+| 9 | replica | the dataflow's progress frontier passes the as-of and `observe_hydration_frontier` calls `set_hydrated` | **`hydrated_at`** |
 | 10 | replica | the demux writes the retract and insert pair, so the per-worker relation carries all three | |
 | 11 | controller | separately, a `Frontiers` response arrives and `update_output_frontier` flips the controller's own hydration view, which is what the 0dt caught-up check and the autoscaling signal read. One round trip later, and it stamps nothing | |
 
@@ -391,21 +429,18 @@ restarting. Consumers must gate on introspection freshness, as
 `mz_object_arrangement_size_history` already does via
 `fresh_introspection_replicas`.
 
-**`REFRESH` materialized views report a refresh interval, not hydration work.**
-The reported output frontier is the meet of write and compute frontier, and a
-REFRESH MV's write frontier sits at the as-of until the first refresh lands, so
-hydration is not considered complete until then. For `REFRESH EVERY '1 day'`,
-`hydrated_at - started_at` can be most of a day, nearly all of it idle. The
-per-object stamps are still internally consistent, so this is not a defect in the
-relation, but any rollup must exclude these objects or it will never close an
-episode. The controller already receives `refresh_schedule` in `add_collection`,
-so it can mark them.
-
-**Read-only mode changes what the output frontier means.** In read-only mode the
-write frontier is deliberately excluded from the reported output frontier, because
-a read-only dataflow cannot push it forward. So `hydrated_at` during a 0dt
-read-only window reflects compute progress only, which is the intended reading but
-differs from the steady-state one.
+**`REFRESH` materialized views hydrate on their computation.** The compute probe
+is attached before the `apply_refresh` operator, deliberately, with the comment in
+`src/compute/src/sink/materialized_view.rs` explaining that rounding frontiers up
+"makes it impossible to accurately track the progress of the computation". So the
+frontier hydration reads is the pre-rounding one, and a REFRESH MV reports when
+its computation caught up rather than anything derived from its schedule.
+
+How a refresh schedule affects the *written* stage is deliberately not claimed
+here. It varies by variant, `REFRESH EVERY` takes an implicit refresh at creation
+while a `REFRESH AT` in the future does not, and getting it wrong in either
+direction is easy. Whoever adds the `written` stamp should establish it against
+the refresh tests rather than inherit a claim from this document.
 
 ### Why the replica and not the compute controller
 

src/compute/src/compute_state.rs

@@ -921,6 +921,8 @@ impl<'a> ActiveComputeState<'a> {
 
         // Maintain a single allocation for `new_frontier` to avoid allocating on every iteration.
         let mut new_frontier = Antichain::new();
+        // Same, for the frontier that measures dataflow progress.
+        let mut hydration_frontier = Antichain::new();
 
         for (&id, collection) in self.compute_state.collections.iter_mut() {
             // The compute protocol does not allow `Frontiers` responses for subscribe and copy-to
@@ -950,6 +952,30 @@ impl<'a> ActiveComputeState<'a> {
                 .allows_reporting(&new_frontier)
                 .then(|| new_frontier.clone());
 
+            // Collect the frontier that measures the dataflow's own progress, which is what
+            // hydration is about.
+            //
+            // This is deliberately not the output frontier collected below. That folds in the
+            // write frontier, which makes it a measure of durability rather than of dataflow
+            // progress, and for a collection that sinks to persist it is not even uniform across
+            // workers: the sink's `mint` operator maintains the shared sink frontier on one
+            // elected worker and clears it on all the others, so the same dataflow would report
+            // hydration at two different times depending on which worker's log you read.
+            //
+            // A collection with a compute frontier produces its output before writing it, so that
+            // frontier is its progress. A collection without one produces its output *by* writing
+            // it, an index into its own trace, so there the write frontier is the progress and
+            // hydration coincides with durability.
+            hydration_frontier.clear();
+            match &collection.compute_probe {
+                Some(probe) => {
+                    probe.with_frontier(|frontier| {
+                        hydration_frontier.extend(frontier.iter().copied())
+                    });
+                }
+                None => hydration_frontier.clone_from(&new_frontier),
+            }
+
             // Collect the output frontier and check for progress.
             //
             // By default, the output frontier equals the write frontier (which is still stored in
@@ -996,6 +1022,8 @@ impl<'a> ActiveComputeState<'a> {
                     .set_reported_output_frontier(ReportedFrontier::Reported(frontier.clone()));
             }
 
+            collection.observe_hydration_frontier(&hydration_frontier);
+
             let response = FrontiersResponse {
                 write_frontier: new_write_frontier,
                 input_frontier: new_input_frontier,
@@ -1209,6 +1237,7 @@ impl<'a> ActiveComputeState<'a> {
                     .set_reported_write_frontier(ReportedFrontier::Reported(new_frontier.clone()));
                 collection
                     .set_reported_input_frontier(ReportedFrontier::Reported(new_frontier.clone()));
+                collection.observe_hydration_frontier(&new_frontier);
                 collection.set_reported_output_frontier(ReportedFrontier::Reported(new_frontier));
             } else {
                 // Presumably tracking state for this subscribe was already dropped by
@@ -1998,6 +2027,13 @@ pub struct CollectionState {
     logging: Option<CollectionLogging>,
     /// Metrics tracked for this collection.
     metrics: CollectionMetrics,
+    /// Whether the dataflow-progress frontier has been observed to pass the as-of, and logged.
+    ///
+    /// Only ever set, never cleared. Reconciliation resets the reported frontiers of a retained
+    /// dataflow, so without this the collection would look unhydrated again and re-log a stage it
+    /// already reported. The `time_ns` path does not need the same guard because the demux drops
+    /// repeats itself.
+    dataflow_hydrated: bool,
     /// Send-side to transition a dataflow from read-only mode to read-write mode.
     ///
     /// All dataflows start in read-only mode. Only after receiving a
@@ -2036,6 +2072,7 @@ impl CollectionState {
             compute_probe: None,
             logging: None,
             metrics,
+            dataflow_hydrated: false,
             read_only_tx,
             read_only_rx,
         }
@@ -2094,13 +2131,35 @@ impl CollectionState {
     }
 
     /// Return whether this collection is hydrated.
+    ///
+    /// This is the output-frontier reading, which folds in the write frontier and so reports
+    /// durability for a collection that sinks to persist. `observe_hydration_frontier` reports the
+    /// dataflow-progress reading instead. Both are wanted, and they differ for a materialized view
+    /// by the time its snapshot takes to reach persist.
     fn hydrated(&self) -> bool {
         match &self.reported_frontiers.output_frontier {
             ReportedFrontier::Reported(frontier) => PartialOrder::less_than(&self.as_of, frontier),
             ReportedFrontier::NotReported { .. } => false,
         }
     }
 
+    /// Observe the frontier measuring this collection's dataflow progress, and log the dataflow
+    /// having hydrated if it has passed the as-of for the first time.
+    ///
+    /// The caller decides which frontier measures progress; see the comment at the call site in
+    /// `report_frontiers`. An empty as-of never hydrates, which is consistent with no dataflow
+    /// being created for one.
+    fn observe_hydration_frontier(&mut self, frontier: &Antichain<Timestamp>) {
+        if self.dataflow_hydrated || !PartialOrder::less_than(&self.as_of, frontier) {
+            return;
+        }
+
+        self.dataflow_hydrated = true;
+        if let Some(logging) = &mut self.logging {
+            logging.set_dataflow_hydrated();
+        }
+    }
+
     /// Allow writes for this collection.
     fn allow_writes(&self) {
         info!(

src/compute/src/logging/compute.rs

@@ -158,6 +158,16 @@ pub struct ErrorCount {
     pub diff: Diff,
 }
 
+/// A dataflow's own progress frontier passed its as-of.
+///
+/// Distinct from [`Hydration`], which fires on the reported output frontier and therefore reports
+/// durability for a collection that sinks to persist.
+#[derive(Debug, Clone, PartialOrd, PartialEq, Columnar)]
+pub struct DataflowHydrated {
+    /// The ID of the export whose dataflow hydrated.
+    pub export_id: GlobalId,
+}
+
 /// An export started hydrating.
 #[derive(Debug, Clone, PartialOrd, PartialEq, Columnar)]
 pub struct HydrationStart {
@@ -236,6 +246,8 @@ pub enum ComputeEvent {
     ErrorCount(ErrorCount),
     /// A dataflow export started hydrating, i.e. its dataflow was unsuspended.
     HydrationStart(HydrationStart),
+    /// A dataflow's progress frontier passed its as-of.
+    DataflowHydrated(DataflowHydrated),
     /// A dataflow export was hydrated.
     Hydration(Hydration),
     /// A dataflow operator's hydration status changed.
@@ -887,6 +899,7 @@ impl DemuxHandler<'_, '_, '_> {
             DataflowShutdown(shutdown) => self.handle_dataflow_shutdown(shutdown),
             ErrorCount(error_count) => self.handle_error_count(error_count),
             HydrationStart(hydration) => self.handle_hydration_start(hydration),
+            DataflowHydrated(hydration) => self.handle_dataflow_hydrated(hydration),
             Hydration(hydration) => self.handle_hydration(hydration),
             OperatorHydration(hydration) => self.handle_operator_hydration(hydration),
             LirMapping(mapping) => self.handle_lir_mapping(mapping),
@@ -1082,8 +1095,6 @@ impl DemuxHandler<'_, '_, '_> {
 
     fn handle_hydration(&mut self, HydrationReference { export_id }: Ref<'_, Hydration>) {
         let ts = self.ts();
-        // Stamp the event time rather than `ts`, as in `handle_export`.
-        let hydrated_at = self.time;
         let export_id = Columnar::into_owned(export_id);
 
         let Some(export) = self.state.exports.get_mut(&export_id) else {
@@ -1100,12 +1111,46 @@ impl DemuxHandler<'_, '_, '_> {
         let nanos = u64::try_from(duration.as_nanos()).expect("must fit");
         export.hydration_time_ns = Some(nanos);
 
+        let timestamps = export.hydration_timestamps;
+
+        let retraction = self
+            .state
+            .pack_hydration_time_update(export_id, None, &timestamps);
+        self.output
+            .hydration_time
+            .give((retraction, ts, Diff::MINUS_ONE));
+        let insertion = self
+            .state
+            .pack_hydration_time_update(export_id, Some(nanos), &timestamps);
+        self.output.hydration_time.give((insertion, ts, Diff::ONE));
+    }
+
+    fn handle_dataflow_hydrated(
+        &mut self,
+        DataflowHydratedReference { export_id }: Ref<'_, DataflowHydrated>,
+    ) {
+        let ts = self.ts();
+        // Stamp the event time rather than `ts`, as in `handle_export`.
+        let hydrated_at = self.time;
+        let export_id = Columnar::into_owned(export_id);
+
+        let Some(export) = self.state.exports.get_mut(&export_id) else {
+            error!(%export_id, "dataflow hydrated event for unknown export");
+            return;
+        };
+        if export.hydration_timestamps.hydrated_at.is_some() {
+            // Reconciliation resets a retained dataflow's reported frontiers, so it can report
+            // progress past its as-of a second time. Ignore the repeats, as `handle_hydration`
+            // does.
+            return;
+        }
+
         let old_timestamps = export.hydration_timestamps;
         export.hydration_timestamps.hydrated_at = Some(hydrated_at);
-        // A dataflow can reach hydration before its `Schedule` arrives, and not only when it has
-        // no imports to suspend: an index over an already-hydrated arrangement reports hydration
-        // while still suspended, which happens for a handful of `mz_catalog_server` indexes on
-        // every bootstrap. So this is a normal path, not a repair for an exotic one.
+        // A dataflow can hydrate before its `Schedule` arrives, and not only when it has no
+        // imports to suspend: an index over an already-hydrated arrangement reports progress while
+        // still suspended, which happens for a handful of `mz_catalog_server` indexes on every
+        // bootstrap. So this is a normal path, not a repair for an exotic one.
         //
         // Stamp `started_at` from `installed_at`, which keeps `installed_at <= started_at <=
         // hydrated_at` total and reports the queueing interval as zero. Stamping `hydrated_at`
@@ -1115,16 +1160,17 @@ impl DemuxHandler<'_, '_, '_> {
             export.hydration_timestamps.started_at = Some(export.hydration_timestamps.installed_at);
         }
         let new_timestamps = export.hydration_timestamps;
+        let time_ns = export.hydration_time_ns;
 
         let retraction = self
             .state
-            .pack_hydration_time_update(export_id, None, &old_timestamps);
+            .pack_hydration_time_update(export_id, time_ns, &old_timestamps);
         self.output
             .hydration_time
             .give((retraction, ts, Diff::MINUS_ONE));
-        let insertion =
-            self.state
-                .pack_hydration_time_update(export_id, Some(nanos), &new_timestamps);
+        let insertion = self
+            .state
+            .pack_hydration_time_update(export_id, time_ns, &new_timestamps);
         self.output.hydration_time.give((insertion, ts, Diff::ONE));
     }
 
@@ -1544,6 +1590,18 @@ impl CollectionLogging {
             }));
     }
 
+    /// Record that the collection's dataflow has hydrated, meaning its own progress frontier
+    /// passed the as-of.
+    ///
+    /// Distinct from [`Self::set_hydrated`], which reports the *output* frontier doing so, and
+    /// therefore reports durability for a collection that sinks to persist.
+    pub fn set_dataflow_hydrated(&self) {
+        self.logger
+            .log(&ComputeEvent::DataflowHydrated(DataflowHydrated {
+                export_id: self.export_id,
+            }));
+    }
+
     /// Set the collection as hydrated.
     pub fn set_hydrated(&self) {
         self.logger.log(&ComputeEvent::Hydration(Hydration {