compute: add a lifecycle event log for compute exports

Record each compute export's lifecycle as an append-only log,
`mz_introspection.mz_compute_lifecycle_events_per_worker`, rather than as more
timestamp columns on the hydration time relation.

Two things stop timestamp columns from carrying the lifecycle. The stages do not
share a grain: `installed`, `started` and `hydrated` are per-worker facts, since
each worker hydrates its own fragment of the dataflow, while whether the output
is durable is a property of the sink as a whole, maintained on one elected
worker. And a timestamp cannot say why the next stage has not happened, so a
NULL cannot tell a replacement materialized view waiting for a cutover apart
from an index that will never write.

    export_id    text        not null
    worker_id    uint8       not null
    event        text        not null
    occurred_at  timestamptz not null
    reason       text        nullable
    details      jsonb       nullable

`installed`, `started` and `hydrated` are logged by every worker. The write
stages are logged only by the worker that maintains the sink's shared write
frontier, so they appear once per object and the row records which worker was
elected. An index emits the first three and stops, which is the index degeneracy
of the lifecycle falling out of the model rather than being special-cased.

`hydrated` reads the dataflow's own progress frontier, the compute probe, not
the reported output frontier. The output frontier folds in the write frontier,
which makes it a measure of durability, and for a sink-backed collection it is
not even uniform across workers: `mint` clears the shared frontier on every
non-elected worker, where it is the empty antichain and contributes nothing to
the meet. `mz_compute_hydration_times_per_worker` is unchanged, so `hydrated_at`
and `time_ns` keep reporting exactly what they reported before, and the new
relation carries the dataflow reading alongside.

The write stages are gated on hydration. Before it the sink has produced nothing
and read-only mode is holding nothing back, and every collection starts
read-only, so reporting a block from installation would put a `write_blocked`
and a `write_unblocked` on essentially every materialized view, both ahead of
`hydrated`. Gating also keeps `written` ordered after `hydrated`, which it is not
otherwise: `apply_refresh` rounds a `REFRESH` materialized view's frontier up to
the next refresh time off its input frontier, before the dataflow computes
anything, so the sink writes an empty batch for the pre-refresh window and the
shard's upper passes the as-of while the dataflow is still hydrating. That also
means a refresh schedule advances writing rather than blocking it, so there is no
`refresh` cause for `write_blocked` to report.

`details` carries the dataflow's as-of, which every stage is defined relative to:
without it the interval between two stages says nothing about how much work was
done, since a replacement materialized view with a far behind as-of is a
completely different amount of work at the same duration.

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

@@ -464,6 +464,7 @@ The `mz_scheduling_parks_histogram` view describes a histogram of [dataflow] wor
 [query hints]: /sql/select/#query-hints
 
 <!-- RELATION_SPEC_UNDOCUMENTED mz_introspection.mz_compute_hydration_times_per_worker -->
+<!-- RELATION_SPEC_UNDOCUMENTED mz_introspection.mz_compute_lifecycle_events_per_worker -->
 <!-- RELATION_SPEC_UNDOCUMENTED mz_introspection.mz_compute_operator_hydration_statuses_per_worker -->
 <!-- RELATION_SPEC_UNDOCUMENTED mz_introspection.mz_dataflow_operator_reachability -->
 <!-- RELATION_SPEC_UNDOCUMENTED mz_introspection.mz_dataflow_operator_reachability_per_worker -->

doc/user/content/reference/system-catalog/mz_introspection.md

@@ -416,6 +416,28 @@ static MIGRATIONS: LazyLock<Vec<MigrationStep>> = LazyLock::new(|| {
             MZ_CATALOG_SCHEMA,
             "mz_views",
         ),
+        // Required because we added the `mz_compute_lifecycle_events_per_worker` builtin log.
+        // make_mz_indexes inlines one VALUES row per builtin log, naming the log and its
+        // `index_by` columns, so adding or removing a log changes the SQL fingerprint of
+        // `mz_indexes` just as adding a builtin index does. See the NOTE above: this version
+        // must stay at the workspace's current dev version until the change ships.
+        MigrationStep::replacement(
+            "26.40.0-dev.0",
+            CatalogItemType::MaterializedView,
+            MZ_CATALOG_SCHEMA,
+            "mz_indexes",
+        ),
+        // Adding a builtin log moves two generated materialized views, not one:
+        // make_mz_sources inlines a VALUES row per builtin log alongside the builtin sources,
+        // so `mz_sources` needs the same treatment. Without it, an upgrade from a released
+        // version reaches `update_fingerprints` with a mismatch for a builtin that is neither
+        // migrated nor ephemeral, which panics and blocks catalog open.
+        MigrationStep::replacement(
+            "26.40.0-dev.0",
+            CatalogItemType::MaterializedView,
+            MZ_CATALOG_SCHEMA,
+            "mz_sources",
+        ),
     ]
 });
 

src/adapter/src/catalog/open/builtin_schema_migration.rs

@@ -1106,6 +1106,7 @@ pub static BUILTINS_STATIC: LazyLock<Vec<Builtin<NameReference>>> = LazyLock::ne
         Builtin::Log(&MZ_COMPUTE_IMPORT_FRONTIERS_PER_WORKER),
         Builtin::Log(&MZ_COMPUTE_ERROR_COUNTS_RAW),
         Builtin::Log(&MZ_COMPUTE_HYDRATION_TIMES_PER_WORKER),
+        Builtin::Log(&MZ_COMPUTE_LIFECYCLE_EVENTS_PER_WORKER),
         Builtin::Log(&MZ_COMPUTE_OPERATOR_HYDRATION_STATUSES_PER_WORKER),
         Builtin::MaterializedView(&MZ_KAFKA_SINKS),
         Builtin::MaterializedView(&MZ_KAFKA_CONNECTIONS),
@@ -2220,6 +2221,21 @@ mod tests {
             Fingerprint::fingerprint(&&mv_extra),
             "mz_sources fingerprint must change when a builtin source is added"
         );
+
+        // Adding an extra log must also change the fingerprint, because the log set is inlined
+        // alongside the source set. Without this case, adding a builtin log moves the
+        // `mz_sources` fingerprint with nothing on the PR path to announce that it needs a
+        // migration step, and catalog open panics on the upgrade.
+        let extra_log = logs[0];
+        let mv_extra_log = builtin::make_mz_sources(
+            sources.iter().copied(),
+            logs.iter().copied().chain(std::iter::once(extra_log)),
+        );
+        assert_ne!(
+            fp_base,
+            Fingerprint::fingerprint(&&mv_extra_log),
+            "mz_sources fingerprint must change when a builtin log is added"
+        );
     }
 
     /// Verifies that the `mz_indexes` materialized view fingerprint changes

src/catalog/src/builtin.rs

@@ -342,6 +342,41 @@ pub static MZ_COMPUTE_HYDRATION_TIMES_PER_WORKER: LazyLock<BuiltinLog> =
         }),
     });
 
+pub static MZ_COMPUTE_LIFECYCLE_EVENTS_PER_WORKER: LazyLock<BuiltinLog> =
+    LazyLock::new(|| BuiltinLog {
+        name: "mz_compute_lifecycle_events_per_worker",
+        schema: MZ_INTROSPECTION_SCHEMA,
+        oid: oid::LOG_MZ_COMPUTE_LIFECYCLE_EVENTS_PER_WORKER_OID,
+        variant: LogVariant::Compute(ComputeLog::LifecycleEvent),
+        access: vec![PUBLIC_SELECT],
+        ontology: Some(Ontology {
+            entity_name: "compute_lifecycle_event_per_worker",
+            description: "Lifecycle events of each compute export, as observed by the worker \
+                          that logged them. Every event carries the wallclock instant it \
+                          occurred at, so durations between stages are differences of \
+                          `occurred_at`. The `installed`, `started` and `hydrated` events are \
+                          logged by every worker, since each worker hydrates its own fragment \
+                          of the dataflow. The write events are logged only by the worker that \
+                          maintains the sink frontier, so they appear once per export rather \
+                          than once per worker.",
+            links: &const {
+                [OntologyLink {
+                    name: "lifecycle_event_of",
+                    target: "compute_export_per_worker",
+                    properties: LinkProperties::MapsTo {
+                        source_column: "export_id",
+                        target_column: "export_id",
+                        via: None,
+                        from_type: Some(SemanticType::GlobalId),
+                        to_type: Some(SemanticType::GlobalId),
+                        note: None,
+                    },
+                }]
+            },
+            column_semantic_types: &[("export_id", SemanticType::GlobalId)],
+        }),
+    });
+
 pub static MZ_COMPUTE_OPERATOR_HYDRATION_STATUSES_PER_WORKER: LazyLock<BuiltinLog> =
     LazyLock::new(|| BuiltinLog {
         name: "mz_compute_operator_hydration_statuses_per_worker",

src/catalog/src/builtin/mz_introspection.rs

@@ -1021,6 +1021,7 @@ impl<'a> Transaction<'a> {
             LogVariant::Compute(ComputeLog::DataflowGlobal) => 31,
             LogVariant::Compute(ComputeLog::OperatorHydrationStatus) => 32,
             LogVariant::Compute(ComputeLog::PrometheusMetrics) => 33,
+            LogVariant::Compute(ComputeLog::LifecycleEvent) => 34,
         };
 
         let mut id: u64 = u64::from(cluster_variant) << 56;

src/catalog/src/durable/transaction.rs

@@ -176,6 +176,8 @@ pub enum ComputeLog {
     ErrorCount,
     /// Hydration times of exported collections.
     HydrationTime,
+    /// Lifecycle events of exported collections.
+    LifecycleEvent,
     /// Hydration status of dataflow operators.
     OperatorHydrationStatus,
     /// Mappings from `GlobalId`/`LirId`` pairs to dataflow addresses.
@@ -370,6 +372,18 @@ impl LogVariant {
                 .with_key(vec![0, 1])
                 .finish(),
 
+            LogVariant::Compute(ComputeLog::LifecycleEvent) => RelationDesc::builder()
+                .with_column("export_id", SqlScalarType::String.nullable(false))
+                .with_column("worker_id", SqlScalarType::UInt64.nullable(false))
+                .with_column("event", SqlScalarType::String.nullable(false))
+                .with_column(
+                    "occurred_at",
+                    SqlScalarType::TimestampTz { precision: None }.nullable(false),
+                )
+                .with_column("reason", SqlScalarType::String.nullable(true))
+                .with_column("details", SqlScalarType::Jsonb.nullable(true))
+                .finish(),
+
             LogVariant::Compute(ComputeLog::OperatorHydrationStatus) => RelationDesc::builder()
                 .with_column("export_id", SqlScalarType::String.nullable(false))
                 .with_column("lir_id", SqlScalarType::UInt64.nullable(false))

src/compute-client/src/logging.rs

@@ -70,7 +70,7 @@ use uuid::Uuid;
 
 use crate::arrangement::manager::{TraceBundle, TraceManager};
 use crate::logging;
-use crate::logging::compute::{CollectionLogging, ComputeEvent, PeekEvent};
+use crate::logging::compute::{CollectionLogging, ComputeEvent, LifecycleStage, PeekEvent};
 use crate::logging::initialize::LoggingTraces;
 use crate::metrics::{CollectionMetrics, WorkerMetrics};
 use crate::render::{LinearJoinSpec, StartSignal};
@@ -694,6 +694,7 @@ impl<'a> ActiveComputeState<'a> {
                     object_id,
                     logger,
                     *dataflow_index,
+                    as_of.as_option().copied(),
                     dataflow.import_ids(),
                 );
                 if starts_immediately {
@@ -884,12 +885,17 @@ impl<'a> ActiveComputeState<'a> {
             let mut collection = CollectionState::new(
                 Rc::clone(&dataflow_index),
                 is_subscribe_or_copy,
-                as_of,
+                as_of.clone(),
                 metrics,
             );
 
-            let logging =
-                CollectionLogging::new(id, logger.clone(), *dataflow_index, std::iter::empty());
+            let logging = CollectionLogging::new(
+                id,
+                logger.clone(),
+                *dataflow_index,
+                as_of.as_option().copied(),
+                std::iter::empty(),
+            );
             // Log collections are never suspended and the controller marks them scheduled
             // implicitly, so no `Schedule` command ever arrives for them. Record their hydration
             // start here, or they would sit permanently in the illegal state of being hydrated
@@ -921,6 +927,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 +958,36 @@ 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),
+            }
+
+            // Evaluate the lifecycle predicates here, while both frontiers are still in hand.
+            // `new_frontier` is folded into the output frontier below, which loses the write
+            // frontier this one is about.
+            let hydrated = PartialOrder::less_than(&collection.as_of, &hydration_frontier);
+            let written = PartialOrder::less_than(&collection.as_of, &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 +1034,9 @@ impl<'a> ActiveComputeState<'a> {
                     .set_reported_output_frontier(ReportedFrontier::Reported(frontier.clone()));
             }
 
+            collection.observe_hydration(hydrated);
+            collection.observe_writes(written);
+
             let response = FrontiersResponse {
                 write_frontier: new_write_frontier,
                 input_frontier: new_input_frontier,
@@ -1209,6 +1250,12 @@ impl<'a> ActiveComputeState<'a> {
                     .set_reported_write_frontier(ReportedFrontier::Reported(new_frontier.clone()));
                 collection
                     .set_reported_input_frontier(ReportedFrontier::Reported(new_frontier.clone()));
+                // Only a batch upper measures progress. `DroppedAt` reports the empty
+                // antichain, which is the maximum of the order, so a subscribe cancelled while
+                // still hydrating would otherwise read as hydrated at the moment it is dropped.
+                let hydrated = matches!(response, SubscribeResponse::Batch(_))
+                    && PartialOrder::less_than(&collection.as_of, &new_frontier);
+                collection.observe_hydration(hydrated);
                 collection.set_reported_output_frontier(ReportedFrontier::Reported(new_frontier));
             } else {
                 // Presumably tracking state for this subscribe was already dropped by
@@ -1998,6 +2045,21 @@ pub struct CollectionState {
     logging: Option<CollectionLogging>,
     /// Metrics tracked for this collection.
     metrics: CollectionMetrics,
+    /// Whether this worker maintains the authoritative write frontier of this collection's sink.
+    ///
+    /// A persist sink elects one worker to track the output shard's upper and clears the shared
+    /// frontier on all the others, so only the elected worker's copy carries write progress. The
+    /// write lifecycle stages are logged by that worker alone, which also makes them a single
+    /// observation per export rather than one per worker. False for collections whose output
+    /// frontier is not a persist upper at all, such as indexes and metric sinks.
+    pub owns_sink_frontier: bool,
+    /// Which lifecycle stages have been logged for this collection.
+    ///
+    /// Stages are only ever added, never removed. 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 lifecycle relation is append-only, so a repeat would show
+    /// up as a duplicate row rather than being dropped.
+    logged_stages: BTreeSet<LifecycleStage>,
     /// 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 +2098,8 @@ impl CollectionState {
             compute_probe: None,
             logging: None,
             metrics,
+            owns_sink_frontier: false,
+            logged_stages: BTreeSet::new(),
             read_only_tx,
             read_only_rx,
         }
@@ -2094,13 +2158,87 @@ 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` 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,
         }
     }
 
+    /// Log that this collection reached a lifecycle stage, unless it already reported it.
+    fn log_stage(&mut self, stage: LifecycleStage) {
+        if !self.logged_stages.insert(stage) {
+            return;
+        }
+        if let Some(logging) = &self.logging {
+            logging.log_lifecycle(stage);
+        }
+    }
+
+    /// Observe whether this collection's dataflow has progressed past its as-of, and log the
+    /// `hydrated` stage the first time it has.
+    ///
+    /// The caller decides which frontier measures dataflow 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(&mut self, hydrated: bool) {
+        if hydrated {
+            self.log_stage(LifecycleStage::Hydrated);
+        }
+    }
+
+    /// Observe whether this collection's sink has written past its as-of, and log the write
+    /// lifecycle stages it has reached.
+    ///
+    /// Only the worker that maintains the sink frontier reports these stages, which is what makes
+    /// them one observation per export rather than one per worker.
+    ///
+    /// Nothing is reported before the dataflow has hydrated. Until then the sink has produced no
+    /// output, so read-only mode is not holding anything back, and reporting a block there would
+    /// make `write_unblocked - hydrated` negative in the common case rather than zero. Gating here
+    /// also keeps the stages ordered against `written`, which can otherwise arrive first:
+    /// `apply_refresh` rounds a `REFRESH` materialized view's frontier up to the next refresh time
+    /// before the dataflow has computed anything, so its sink writes an empty batch for the
+    /// pre-refresh window and the shard's upper passes the as-of while the dataflow is still
+    /// hydrating.
+    ///
+    /// NOTE: `written` is derived from the output shard's upper, which is a property of the shard
+    /// and not of this replica. The as-of is bounded to one step below that upper for a non-empty
+    /// storage export (`as_of_selection::apply_downstream_storage_constraints`), so for a shard
+    /// that already holds data the predicate is true from the moment the dataflow is installed. A
+    /// replica that may not write can therefore never be the one that advanced it, which is why
+    /// the stage is withheld while writes are blocked. Reporting it there would attribute another
+    /// writer's progress to this replica and put `written` before `write_unblocked`.
+    fn observe_writes(&mut self, written: bool) {
+        if !self.owns_sink_frontier || !self.logged_stages.contains(&LifecycleStage::Hydrated) {
+            return;
+        }
+
+        let read_only = *self.read_only_rx.borrow();
+        if read_only {
+            self.log_stage(LifecycleStage::WriteBlockedReadOnly);
+            return;
+        }
+
+        if self
+            .logged_stages
+            .contains(&LifecycleStage::WriteBlockedReadOnly)
+        {
+            // Only report having been unblocked if we reported being blocked. A collection whose
+            // writes were allowed before we first observed it was never seen to wait.
+            self.log_stage(LifecycleStage::WriteUnblocked);
+        }
+
+        if written {
+            self.log_stage(LifecycleStage::Written);
+        }
+    }
+
     /// Allow writes for this collection.
     fn allow_writes(&self) {
         info!(