adapter: collect durable object hydration history

doc/developer/design/20260817_durable_object_hydration_history.md

@@ -50,6 +50,11 @@ It installs an internal subscribe on that replica which aggregates every worker'
 completed rows, anti-joins them against the history table, and writes the missing
 ones through the timestamped OCC read-then-write path.
 
+Collection has no explicit batch bound. It returns at most one row per
+not-yet-recorded dataflow, and the OCC path rejects a result that exceeds
+`max_result_size` or `max_query_result_size`. At their 1 GiB defaults that ceiling
+only matters at millions of dataflows per replica.
+
 Two dyncfgs control it. `hydration_history_collection_interval` sets the sweep
 cadence and disables collection at zero, which is the production default.
 `hydration_history_retention_period` bounds how long rows live, defaulting to 30
@@ -148,6 +153,13 @@ sampling race rather than depending on replica configuration for an expected wor
 count. In that case the durable finish can precede the latest worker's finish. This
 is separate from restart behavior, which discards the replica collection as a unit.
 
+The missing worker cannot later change the episode key. Its logging clock stamps
+both the Differential update and `installed_at`, so appearing after the read means
+its installation stamp is later than the visible minimum. A later sweep therefore
+keeps the same `min(installed_at)` and the anti-join matches the row already written.
+The worker can raise `max(hydrated_at)`, but the durable row is not repaired after
+its episode key has been recorded.
+
 Compute is adding an append-only lifecycle log for the same stages, currently
 proposed in #38403: one row per export, worker and event, with a reason and the
 dataflow's as-of. It leaves `mz_compute_hydration_times_per_worker` alone, so
@@ -190,16 +202,21 @@ writes their retractions at the observed frontier. Collection applies the same
 cutoff, so a still-live log row cannot resurrect an episode retention just
 retracted.
 
-Retention deletes successive bounded batches until one is not full. The fixed
-cutoff makes the eligible set finite, and collection refuses to insert rows behind
-that cutoff, so a successful sweep drains the backlog even when more than one batch
-expires at once. The bound is not a nicety. The OCC path refuses a selection larger
-than `max_result_size` before submitting any write, so one unbounded delete over a
-large backlog would fail identically forever and never shrink the table. The bound
-has to sit inside a derived table, because a top-level `LIMIT` lands in the plan's
+Retention deletes one bounded batch per sweep. The fixed cutoff makes the eligible
+set finite, and collection refuses to insert rows behind that cutoff, so later
+sweeps continue draining the same backlog without starving collection. The batch
+bound is not a nicety. The OCC path refuses a selection larger than
+`max_result_size` before submitting any write, so one unbounded delete over a large
+backlog would fail identically forever and never shrink the table. The bound has to
+sit inside a derived table, because a top-level `LIMIT` lands in the plan's
 `RowSetFinishing`, which the OCC path deliberately discards, and the delete would be
 silently unbounded again.
 
+`mz_hydration_history_retention_batch_full_total` increments when the batch deletes
+all 1,000 rows. Repeated increments mean retention may not be keeping up. An operator
+can lower `hydration_history_collection_interval` to schedule sweeps more often,
+then compare appended and deleted row rates to confirm the backlog is shrinking.
+
 Retention runs on the catalog server, so it keeps working when there are no user
 replicas at all, and it runs even when that sweep's collection failed. A
 crash-looping replica must not be able to stop the table from shrinking. The
@@ -238,6 +255,11 @@ the interval is one fleet-wide setting and an unshifted grid would have every
 environment sweep at the same instant. Using the full id also separates regions and
 ordinals belonging to one organization.
 
+One replica per interval means an environment with `N` eligible replicas revisits
+each one approximately every `N * interval`. Freshness therefore degrades linearly
+with replica count. Lowering the interval improves freshness at the cost of more
+replica dataflow installs.
+
 **Background mutations take no OCC write permit.** The permits are one semaphore
 shared by every read-then-write in the process, not one per table. A session's wait
 is bounded by its statement timeout, a sweep's is not, and a sweep's subscribe has
@@ -308,6 +330,12 @@ the sqllogictest runner defaults, against the usual preference for enabling new
 paths in tests: the collector installs subscribes and writes a builtin table, while
 those runs assert on catalog contents and plans.
 
+Background collection always uses frontend OCC, independently of the session
+`frontend_read_then_write` rollout flag. This is safe while the lock path remains
+available because the target is a system table, which user DML can neither read nor
+write. The background OCC entry point enforces that target contract rather than
+relying on each maintenance caller to remember the rollout constraint.
+
 ## Future Work
 
 - Record installation and start before completion, then finalize canceled and

doc/developer/guide-adapter.md

@@ -242,6 +242,15 @@ keep the change invisible to session-visible catalog reads (name resolution,
 planning). Otherwise sessions serve stale catalogs where today they would see
 the change.
 
+### Background OCC must stay disjoint from the lock path
+
+The frontend OCC and coordinator lock paths do not synchronize with each other.
+Background maintenance that bypasses the frontend rollout flag must therefore
+target a system table and read only system relations. User DML can neither read
+nor write system relations, so the two paths cannot interleave on a target or
+dependency. A background caller that needs a user relation requires a separate
+contract that establishes every writer uses OCC across the fleet.
+
 ### Group commits and generation handover
 
 At runtime, one group committer per `environmentd` serializes txns-shard operations:

doc/user/data/metrics.yml

@@ -97,7 +97,7 @@ metrics:
   source: src/adapter/src/metrics.rs
   visibility: internal
 - name: mz_active_internal_subscribes
-  help: The number of active internal subscribes, which serve frontend-sequenced read-then-write.
+  help: The number of active internal subscribes used by read-then-write operations and background maintenance.
   labels:
   - session_type
   source: src/adapter/src/metrics.rs
@@ -955,6 +955,37 @@ metrics:
   help: The time it takes to advance the catalog shard upper for a txns-shard write (group commits and table register/forget).
   source: src/adapter/src/metrics.rs
   visibility: internal
+- name: mz_hydration_history_mutations_total
+  help: Total hydration-history collection and retention mutations since process start.
+  labels:
+  - operation
+  - outcome
+  source: src/adapter/src/metrics.rs
+  visibility: internal
+- name: mz_hydration_history_retention_batch_full_total
+  help: Total hydration-history sweeps whose retention batch was full. Repeated increments mean retention may not be keeping up with its schedule.
+  source: src/adapter/src/metrics.rs
+  visibility: internal
+- name: mz_hydration_history_rows_affected_total
+  help: Total rows changed by hydration-history maintenance since process start.
+  labels:
+  - action
+  source: src/adapter/src/metrics.rs
+  visibility: internal
+- name: mz_hydration_history_sweep_duration_seconds_bucket
+  help: Wall time of a complete hydration-history collection and retention sweep.
+  labels:
+  - le
+  source: src/adapter/src/metrics.rs
+  visibility: internal
+- name: mz_hydration_history_sweep_duration_seconds_count
+  help: Wall time of a complete hydration-history collection and retention sweep.
+  source: src/adapter/src/metrics.rs
+  visibility: internal
+- name: mz_hydration_history_sweep_duration_seconds_sum
+  help: Wall time of a complete hydration-history collection and retention sweep.
+  source: src/adapter/src/metrics.rs
+  visibility: internal
 - name: mz_index_peek_cursor_setup_seconds_bucket
   help: Time setting up cursor and literal constraints.
   labels:

misc/python/materialize/mzcompose/__init__.py

@@ -147,6 +147,9 @@ def get_minimal_system_parameters(
     if version < MzVersion.parse_mz("v26.25.0-dev"):
         config["enable_multi_replica_sources"] = "true"
 
+    if version >= MzVersion.parse_mz("v26.40.0-dev"):
+        config["hydration_history_collection_interval"] = "60s"
+
     if sanitizer_enabled():
         config["with_0dt_deployment_max_wait"] = "18000s"
 
@@ -495,6 +498,17 @@ def get_variable_system_parameters(
         VariableSystemParameter(
             "arrangement_size_history_retention_period", "7d", ["1min", "1h", "7d"]
         ),
+        *(
+            [
+                VariableSystemParameter(
+                    "hydration_history_retention_period",
+                    "30d",
+                    ["1min", "1h", "30d"],
+                )
+            ]
+            if version >= MzVersion.parse_mz("v26.40.0-dev")
+            else []
+        ),
         VariableSystemParameter(
             "persist_validate_part_bounds_on_read", "false", ["true", "false"]
         ),

misc/python/materialize/parallel_workload/action.py

@@ -3009,6 +3009,16 @@ def __init__(
             "'1h'",
             "'7d'",
         ]
+        self.flags_with_values["hydration_history_collection_interval"] = [
+            "'0s'",
+            "'1s'",
+            "'1min'",
+        ]
+        self.flags_with_values["hydration_history_retention_period"] = [
+            "'1min'",
+            "'1h'",
+            "'30d'",
+        ]
         # Keep these generous: a tight timeout would abort the oracle's own
         # queries (they are retried, but it adds noise). "0s" leaves it unset.
         self.flags_with_values["pg_timestamp_oracle_statement_timeout"] = [

src/adapter-types/src/dyncfgs.rs

@@ -421,6 +421,22 @@ pub const ARRANGEMENT_SIZE_HISTORY_RETENTION_PERIOD: Config<Duration> = Config::
     ParameterScope::Environment,
 );
 
+/// How often to sweep replicas for completed object hydration episodes.
+pub const HYDRATION_HISTORY_COLLECTION_INTERVAL: Config<Duration> = Config::new(
+    "hydration_history_collection_interval",
+    Duration::ZERO,
+    "How often to record completed object hydration episodes. A zero duration disables collection.",
+    ParameterScope::Environment,
+);
+
+/// How long to retain completed object hydration episodes.
+pub const HYDRATION_HISTORY_RETENTION_PERIOD: Config<Duration> = Config::new(
+    "hydration_history_retention_period",
+    Duration::from_hours(30 * 24),
+    "How long to retain rows in mz_internal.mz_object_hydration_history.",
+    ParameterScope::Environment,
+);
+
 /// How frequently the catalog `*_info` metrics (`mz_object_info`,
 /// `mz_cluster_info`, …) are reconciled with the catalog. A zero duration
 /// disables reconciliation.
@@ -557,6 +573,8 @@ pub fn all_dyncfgs(configs: ConfigSet) -> ConfigSet {
         .add(&CONSOLE_OIDC_SCOPES)
         .add(&ARRANGEMENT_SIZE_HISTORY_COLLECTION_INTERVAL)
         .add(&ARRANGEMENT_SIZE_HISTORY_RETENTION_PERIOD)
+        .add(&HYDRATION_HISTORY_COLLECTION_INTERVAL)
+        .add(&HYDRATION_HISTORY_RETENTION_PERIOD)
         .add(&CATALOG_INFO_METRICS_RECONCILE_INTERVAL)
         .add(&PG_TIMESTAMP_ORACLE_STATEMENT_TIMEOUT)
         .add(&FRONTEND_READ_THEN_WRITE)

src/adapter/src/active_compute_sink.rs

@@ -51,10 +51,10 @@ impl ActiveComputeSink {
     }
 
     /// Reports the ID of the connection which created the sink.
-    pub fn connection_id(&self) -> &ConnectionId {
+    pub fn connection_id(&self) -> Option<&ConnectionId> {
         match &self {
-            ActiveComputeSink::Subscribe(subscribe) => &subscribe.conn_id,
-            ActiveComputeSink::CopyTo(copy_to) => &copy_to.conn_id,
+            ActiveComputeSink::Subscribe(subscribe) => subscribe.connection_id(),
+            ActiveComputeSink::CopyTo(copy_to) => Some(&copy_to.conn_id),
         }
     }
 
@@ -147,13 +147,26 @@ impl SubscribeBacklogAccounting {
     }
 }
 
+/// Ownership and cleanup scope of an active subscribe.
+#[derive(Debug)]
+pub enum ActiveSubscribeOwner {
+    /// The subscribe belongs to a SQL session.
+    Session {
+        conn_id: ConnectionId,
+        session_uuid: Uuid,
+    },
+    /// The subscribe belongs to a coordinator background task.
+    ///
+    /// Always `internal`, since there is no session to attribute a
+    /// `mz_subscriptions` row to.
+    Background,
+}
+
 /// A description of an active subscribe from coord's perspective
 #[derive(Debug)]
 pub struct ActiveSubscribe {
-    /// The ID of the connection which created the subscribe.
-    pub conn_id: ConnectionId,
-    /// The UUID of the session which created the subscribe.
-    pub session_uuid: Uuid,
+    /// The owner responsible for retiring the subscribe.
+    pub owner: ActiveSubscribeOwner,
     /// The ID of the cluster on which the subscribe is running.
     pub cluster_id: ClusterId,
     /// The IDs of the objects on which the subscribe depends.
@@ -189,6 +202,33 @@ pub struct ActiveSubscribe {
 }
 
 impl ActiveSubscribe {
+    /// The session uuid for this subscribe's `mz_subscriptions` row, or `None`
+    /// if it does not appear there.
+    pub fn introspection_session_uuid(&self) -> Option<Uuid> {
+        match &self.owner {
+            ActiveSubscribeOwner::Session { session_uuid, .. } if !self.internal => {
+                Some(*session_uuid)
+            }
+            _ => None,
+        }
+    }
+
+    /// Returns the owning connection, if this is a session subscribe.
+    pub fn connection_id(&self) -> Option<&ConnectionId> {
+        match &self.owner {
+            ActiveSubscribeOwner::Session { conn_id, .. } => Some(conn_id),
+            ActiveSubscribeOwner::Background => None,
+        }
+    }
+
+    /// Returns the owning session UUID, if this is a session subscribe.
+    pub fn session_uuid(&self) -> Option<Uuid> {
+        match self.owner {
+            ActiveSubscribeOwner::Session { session_uuid, .. } => Some(session_uuid),
+            ActiveSubscribeOwner::Background => None,
+        }
+    }
+
     /// Initializes the subscription.
     ///
     /// This method must be called exactly once, after constructing an

src/adapter/src/catalog/builtin_table_updates.rs

@@ -51,6 +51,7 @@ use mz_sql::names::SchemaSpecifier;
 use mz_sql_parser::ast::display::AstDisplay;
 use mz_storage_client::client::TableData;
 use smallvec::smallvec;
+use uuid::Uuid;
 
 // DO NOT add any more imports from `crate` outside of `crate::catalog`.
 use crate::active_compute_sink::ActiveSubscribe;
@@ -811,12 +812,13 @@ impl CatalogState {
         &self,
         id: GlobalId,
         subscribe: &ActiveSubscribe,
+        session_uuid: Uuid,
         diff: Diff,
     ) -> BuiltinTableUpdate<&'static BuiltinTable> {
         let mut row = Row::default();
         let mut packer = row.packer();
         packer.push(Datum::String(&id.to_string()));
-        packer.push(Datum::Uuid(subscribe.session_uuid));
+        packer.push(Datum::Uuid(session_uuid));
         packer.push(Datum::String(&subscribe.cluster_id.to_string()));
 
         let start_dt = mz_ore::now::to_datetime(subscribe.start_time);