compute: delay retracting a dropped export's lifecycle rows [deferred, unsound]

doc/developer/design/20260817_compute_hydration_timestamps.md

@@ -199,11 +199,13 @@ details      jsonb       nullable
 | `write_blocked` | per object | `read_only` |
 | `write_unblocked` | per object | none |
 | `written` | per object | none |
+| `dropped` | per worker | none |
 
-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. Subscribes and
-`COPY TO` stop early for the same reason, and a metric sink folds its output into
-the metrics registry rather than into a shard, so it has no write stages either.
+An index emits the first three and then `dropped`, stopping short of the write
+stages, which is the index degeneracy of the lifecycle falling out of the model
+rather than being special-cased. Subscribes and `COPY TO` stop early for the same
+reason, and a metric sink folds its output into the metrics registry rather than
+into a shard, so it has no write stages either.
 
 **Grain.** `worker_id` is the worker that observed the event and is never NULL. The
 per-object events are observed by the single worker that maintains the sink's
@@ -331,9 +333,37 @@ earn a column and is worth having in the row. Invariant tests must assert on
 `event`, `reason` and `occurred_at` and never on `details`, or the first test that
 pins a field removes the extensibility it exists for.
 
-**Bounds.** At most six rows per object, times workers for the first three events,
-all retracted when the object is dropped. This is in-memory introspection, so
-there is no durable growth to reason about.
+**Bounds.** At most seven rows per object, times workers for the four per-worker
+events, all retracted a delay after the object is dropped. This is in-memory
+introspection, so there is no durable growth to reason about, but for the length of
+that delay a dropped object's rows are still resident.
+
+The delay is what makes a dropped object's history readable at all, and the two
+populations churn at rates too different to share one value. A transient export
+exists per peek and per subscribe, so its rate is the query rate: at eight workers
+and two hundred peeks per second, a five minute window holds around 1.4 million
+rows, hundreds of megabytes. A few seconds is enough for a reader to observe them
+and costs single digit megabytes. A user object is dropped by DDL, so a thousand
+drops inside a five minute window is under ten megabytes, and there the history is
+the thing worth keeping. Hence `compute_lifecycle_retraction_delay` and
+`compute_lifecycle_retraction_delay_transient`, defaulting to five minutes and five
+seconds.
+
+Both are floored at the logging interval in code rather than by convention. The
+demux rounds update timestamps up to that interval, so a shorter delay can round to
+the same timestamp as the insertion, retracting the rows before any reader can
+observe them and defeating the point. Reading the delays per batch rather than at
+construction is what lets a `dyncfg` change take effect without recreating the
+logging dataflow, and it makes them eligible for per-replica overrides, so a
+replica under investigation can retain longer than the rest.
+
+**`dropped` is a stage, not just a retraction.** Without it a lingering row says
+only that an export reached some stage, not whether it still exists, so "hydrated
+but never written" cannot be told apart from "dropped before it wrote". With it the
+last event names the object's fate and the delay is pure retention. It also makes
+the log self-describing while a delay window is open: a row whose `export_id` no
+longer appears in `mz_objects` is explained by its own `dropped` event rather than
+reading as a leak.
 
 **Only `read_only` is attributed.** It is the one cause of a write block that
 compute can observe. Two further attributions would be useful and are not
@@ -572,15 +602,21 @@ downstream of it. Those objects get fresh timestamps on a replica that has been
 for a long time. That is arguably correct, since the dataflows really were
 rebuilt, but it is not obvious from the outside.
 
-**Very short episodes may not be observable at all.** The demux assigns updates a
-timestamp rounded up to the logging interval, and the introspection write path
-consolidates: the subscribe handler discards the subscribe timestamp and flattens
-a batch into one append, and the storage write task consolidates on its own batch
-interval. So an object whose transitions all fall inside one such window presents
-only its final state, and an object that dies before its rows reach persist leaves
-no record. In the limit this is unavoidable, and stamping event time rather than
-the rounded update timestamp is what keeps it to a visibility limit rather than an
-accuracy one: an episode that is recorded is recorded accurately.
+**Very short episodes are recorded, but not as a sequence of states.** The demux
+assigns updates a timestamp rounded up to the logging interval, and the
+introspection write path consolidates: the subscribe handler discards the subscribe
+timestamp and flattens a batch into one append, and the storage write task
+consolidates on its own batch interval. An object whose transitions all fall inside
+one such window therefore appears with its whole history at once rather than
+progressing through it. Stamping event time rather than the rounded update
+timestamp is what keeps that a limit on snapshots rather than on accuracy: the
+sequence and the intervals survive the rounding even where the intermediate states
+are never separately visible.
+
+What is not lost is the episode itself, and that is the retraction delay's doing. A
+dataflow that comes and goes inside one interval still leaves its rows behind for a
+reader to find, which is the case that matters most, since a dataflow dropped
+before it hydrated is exactly the one someone is looking for.
 
 **A crash is not a compute event and is not reported here.** A replica that dies
 cannot report its own death. What the relation shows is that a dropped object's
@@ -670,7 +706,14 @@ The lifecycle log:
 - `src/compute/src/logging/compute.rs`: the `Lifecycle` event and the
   `LifecycleStage` vocabulary, an as-of field on `Export`, a demux output and
   packer including the `jsonb` column, and the emitted rows kept on `ExportState`
-  so that they can be retracted verbatim when the export is dropped.
+  so that they can be retracted verbatim, at a delayed timestamp, when the export
+  is dropped.
+- `src/compute-types/src/dyncfgs.rs`: the two retraction delays. They reach the
+  replica through `ComputeCommand::UpdateConfiguration`, which applies them in place
+  to the `ConfigSet` the logging dataflow holds, so no plumbing beyond registration
+  is needed. `misc/python/materialize/mzcompose/__init__.py` and
+  `misc/python/materialize/parallel_workload/action.py` must both list the flag
+  names, or `check-test-flags` fails.
 - `src/compute/src/compute_state.rs`: the stage bookkeeping on `CollectionState`
   and the observation of both frontiers in `report_frontiers`.
 - `src/compute/src/sink/materialized_view.rs` and `materialized_view_v2.rs`: the
@@ -694,9 +737,15 @@ Goldens that hardcode a log relation's identity, columns, OIDs or indexes:
 `test/workload-replay/system_catalog_identifiers.txt` and `objects.txt`. Docs: the
 `mz_introspection` system catalog reference page.
 
-`catalog_server_explain.slt` and `test/cluster/mzcompose.py` need no change. The
-former's query filters `o.id NOT LIKE 'si%'`, which excludes per-replica
-introspection log indexes, and the latter queries named relations.
+`catalog_server_explain.slt` needs its `Constant (N rows)` counts bumped. Its query
+filters `o.id NOT LIKE 'si%'`, so no new EXPLAIN entry appears for the per-replica
+log index, but the existing plans embed the inlined builtin `VALUES` sets as
+constant nodes, so every count over a catalog relation that gained a row moves,
+including the two ontology counts moved by the new entity and link. The question a
+new builtin raises here is not whether a plan is added but whether the existing
+plans change.
+
+`test/cluster/mzcompose.py` needs no change, since it queries named relations.
 
 Not touched, and deliberately so: the introspection subscribe,
 `mz_internal.mz_compute_hydration_times`,
@@ -712,6 +761,8 @@ New testdrive coverage worth adding:
 - An import-free dataflow carrying `started_at` from creation, equal to its
   `installed_at`, and satisfying the ordering invariant.
 - Log collections having all three timestamps set.
+- A dropped object's rows outliving the drop by the configured delay, carrying a
+  `dropped` event, and then disappearing.
 - All timestamps surviving an environmentd restart unchanged.
 - A replica restart yielding entirely fresh values.
 

misc/python/materialize/mzcompose/__init__.py

@@ -609,6 +609,8 @@ def get_default_system_parameters(
     "compute_dataflow_max_inflight_bytes_cc",
     "compute_flat_map_fuel",
     "compute_temporal_bucketing_summary",
+    "compute_lifecycle_retraction_delay",
+    "compute_lifecycle_retraction_delay_transient",
     "consolidating_vec_growth_dampener",
     "copy_to_s3_parquet_row_group_file_ratio",
     "copy_to_s3_arrow_builder_buffer_ratio",

misc/python/materialize/parallel_workload/action.py

@@ -3146,6 +3146,8 @@ def __init__(
             "compute_replica_expiration_offset",
             "enable_compute_render_fueled_as_specific_collection",
             "compute_temporal_bucketing_summary",
+            "compute_lifecycle_retraction_delay",
+            "compute_lifecycle_retraction_delay_transient",
             "enable_compute_logical_backpressure",
             "enable_replica_targeted_materialized_views",
             "compute_logical_backpressure_max_retained_capabilities",

src/compute-types/src/dyncfgs.rs

@@ -245,6 +245,32 @@ pub const TEMPORAL_BUCKETING_SUMMARY: Config<Duration> = Config::new(
     "The summary to apply to frontiers in temporal bucketing in compute.",
 );
 
+/// How long to keep a dropped export's lifecycle rows before retracting them.
+///
+/// The rows of `mz_compute_lifecycle_events_per_worker` are retracted when an export is dropped,
+/// which without a delay makes an object's history vanish the moment it goes away. Delaying the
+/// retraction leaves the history readable for a while after the fact.
+///
+/// Clamped up to the logging interval, since a shorter delay can round to the same update
+/// timestamp as the insertion and leave the rows never separately visible.
+pub const LIFECYCLE_RETRACTION_DELAY: Config<Duration> = Config::new(
+    "compute_lifecycle_retraction_delay",
+    Duration::from_secs(300),
+    "How long to keep a dropped export's rows in the compute lifecycle log.",
+);
+
+/// As [`LIFECYCLE_RETRACTION_DELAY`], for transient exports.
+///
+/// Transient exports are created per peek and per subscribe, so their churn is driven by query
+/// rate rather than by DDL. Retaining them for the same duration as user objects would cost
+/// hundreds of megabytes on a busy replica, where a few seconds is enough for a reader to observe
+/// them at all.
+pub const LIFECYCLE_RETRACTION_DELAY_TRANSIENT: Config<Duration> = Config::new(
+    "compute_lifecycle_retraction_delay_transient",
+    Duration::from_secs(5),
+    "How long to keep a dropped transient export's rows in the compute lifecycle log.",
+);
+
 /// The yielding behavior with which linear joins should be rendered.
 pub const LINEAR_JOIN_YIELDING: Config<&str> = Config::new(
     "linear_join_yielding",
@@ -575,6 +601,8 @@ pub fn all_dyncfgs(configs: ConfigSet) -> ConfigSet {
         .add(&CORRECTION_V2_CHUNK_SIZE)
         .add(&ENABLE_COMPUTE_TEMPORAL_BUCKETING)
         .add(&TEMPORAL_BUCKETING_SUMMARY)
+        .add(&LIFECYCLE_RETRACTION_DELAY)
+        .add(&LIFECYCLE_RETRACTION_DELAY_TRANSIENT)
         .add(&LINEAR_JOIN_YIELDING)
         .add(&ENABLE_LGALLOC)
         .add(&LGALLOC_BACKGROUND_INTERVAL)

src/compute/src/logging/compute.rs

@@ -21,7 +21,9 @@ use columnar::{Columnar, Index, Ref};
 use differential_dataflow::VecCollection;
 use differential_dataflow::collection::AsCollection;
 use differential_dataflow::trace::{BatchReader, Cursor, Navigable};
+use mz_compute_types::dyncfgs::{LIFECYCLE_RETRACTION_DELAY, LIFECYCLE_RETRACTION_DELAY_TRANSIENT};
 use mz_compute_types::plan::LirId;
+use mz_dyncfg::ConfigSet;
 use mz_ore::cast::CastFrom;
 use mz_repr::adt::timestamp::CheckedTimestamp;
 use mz_repr::{Datum, Diff, GlobalId, Row, RowRef, Timestamp};
@@ -208,6 +210,11 @@ pub enum LifecycleStage {
     WriteUnblocked,
     /// The export's sink advanced the output shard's upper past the as-of.
     Written,
+    /// The export was dropped.
+    ///
+    /// Terminal. Its rows are retracted after a delay, so a reader that sees this stage is looking
+    /// at the history of an object that no longer exists.
+    Dropped,
 }
 
 impl LifecycleStage {
@@ -223,6 +230,7 @@ impl LifecycleStage {
             Self::WriteBlockedReadOnly => ("write_blocked", Some("read_only")),
             Self::WriteUnblocked => ("write_unblocked", None),
             Self::Written => ("written", None),
+            Self::Dropped => ("dropped", None),
         }
     }
 }
@@ -375,8 +383,11 @@ pub(super) fn construct<'scope>(
     config: &mz_compute_client::logging::LoggingConfig,
     event_queue: EventQueue<Column<(Duration, ComputeEvent)>>,
     shared_state: Rc<RefCell<SharedLoggingState>>,
+    worker_config: Rc<ConfigSet>,
 ) -> Return {
     let logging_interval_ms = std::cmp::max(1, config.interval.as_millis());
+    let logging_interval =
+        Duration::from_millis(u64::try_from(logging_interval_ms).expect("must fit"));
 
     scope.scoped("compute logging", move |scope| {
         let enable_logging = config.enable_logging;
@@ -465,6 +476,17 @@ pub(super) fn construct<'scope>(
                         dataflow_global_ids: dataflow_global_ids.session_with_builder(&cap),
                     };
 
+                    // Read the delays per batch rather than once, so that a configuration change
+                    // takes effect without recreating the logging dataflow. The floor keeps a
+                    // delay from rounding to the same update timestamp as the insertion, which
+                    // would leave a dropped export's rows never separately visible.
+                    let retraction_delay = LIFECYCLE_RETRACTION_DELAY
+                        .get(&worker_config)
+                        .max(logging_interval);
+                    let retraction_delay_transient = LIFECYCLE_RETRACTION_DELAY_TRANSIENT
+                        .get(&worker_config)
+                        .max(logging_interval);
+
                     let shared_state = &mut shared_state.borrow_mut();
                     for (time, event) in data.borrow().into_index_iter() {
                         DemuxHandler {
@@ -473,6 +495,8 @@ pub(super) fn construct<'scope>(
                             output: &mut output_sessions,
                             logging_interval_ms,
                             time,
+                            retraction_delay,
+                            retraction_delay_transient,
                         }
                         .handle(event);
                     }
@@ -980,13 +1004,25 @@ struct DemuxHandler<'a, 'b, 'c> {
     logging_interval_ms: u128,
     /// The current event time.
     time: Duration,
+    /// How long to delay retracting a dropped export's lifecycle rows.
+    retraction_delay: Duration,
+    /// Same, for transient exports.
+    retraction_delay_transient: Duration,
 }
 
 impl DemuxHandler<'_, '_, '_> {
     /// Return the timestamp associated with the current event, based on the event time and the
     /// logging interval.
     fn ts(&self) -> Timestamp {
-        let time_ms = self.time.as_millis();
+        self.ts_at(Duration::ZERO)
+    }
+
+    /// Return the timestamp `offset` past the current event's, rounded as [`Self::ts`] rounds.
+    ///
+    /// Giving an update at a time beyond the session's capability is allowed, so this is how a
+    /// retraction is scheduled into the future.
+    fn ts_at(&self, offset: Duration) -> Timestamp {
+        let time_ms = (self.time + offset).as_millis();
         let interval = self.logging_interval_ms;
         let rounded = (time_ms / interval + 1) * interval;
         rounded.try_into().expect("must fit")
@@ -1094,11 +1130,42 @@ impl DemuxHandler<'_, '_, '_> {
             .hydration_time
             .give((datum, ts, Diff::MINUS_ONE));
 
-        // Remove lifecycle logging for this export.
-        for (key, value) in export.lifecycle_rows.values() {
+        // Record the drop as a lifecycle stage of its own, then retract the whole history at a
+        // later timestamp. Without the `dropped` stage the delay would be unreadable: a lingering
+        // row would say only that an export reached some stage, not whether it still exists, so
+        // "hydrated but never written" could not be told apart from "dropped before it wrote".
+        let details = lifecycle_details(export.as_of);
+        let dropped = {
+            let (key, value) = self.state.pack_lifecycle_update(
+                export_id,
+                dataflow_index,
+                LifecycleStage::Dropped,
+                self.time,
+                details.unpack_first(),
+            );
+            (key.to_owned(), value.to_owned())
+        };
+        self.output
+            .lifecycle
+            .give(((&*dropped.0, &*dropped.1), ts, Diff::ONE));
+
+        // Transient exports are created per peek and per subscribe, so they churn with query rate
+        // rather than with DDL. They keep a short delay, enough for a reader to observe them,
+        // where a user object can afford to linger.
+        let delay = if export_id.is_transient() {
+            self.retraction_delay_transient
+        } else {
+            self.retraction_delay
+        };
+        let retract_ts = self.ts_at(delay);
+        for (key, value) in export
+            .lifecycle_rows
+            .values()
+            .chain(std::iter::once(&dropped))
+        {
             self.output
                 .lifecycle
-                .give(((&**key, &**value), ts, Diff::MINUS_ONE));
+                .give(((&**key, &**value), retract_ts, Diff::MINUS_ONE));
         }
 
         // Remove operator hydration logging for this export.

src/compute/src/logging/initialize.rs

@@ -166,6 +166,7 @@ impl LoggingContext<'_> {
                 self.config,
                 self.c_event_queue.clone(),
                 Rc::clone(&self.shared_state),
+                Rc::clone(&self.worker_config),
             );
             collections.extend(compute_collections);