compute: add a lifecycle event log for compute exports

doc/user/content/reference/system-catalog/mz_introspection.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 -->

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

@@ -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/catalog/src/builtin.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/mz_introspection.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/durable/transaction.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/compute-client/src/logging.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/src/compute_state.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!(