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
@@ -250,10 +272,10 @@ nothing else stops it outliving the coordinator, and the teardown that follows d
 the timestamp oracle's worker task. A sweep still running would then read a
 timestamp from a dead oracle and panic, so the coordinator owns the task handle.
 
-**Each mutation's timeout is deliberately generous.** Even a mutation with nothing
-to write waits for its read to linearize, which can take a full
-`default_timestamp_interval`, a parameter with no upper bound. A tighter bound would
-let a large timestamp interval starve retention permanently.
+**Each mutation's timeout is deliberately generous.** Subscribe installation,
+replica-side progress, OCC conflict retries, and the external commit can all wait
+indefinitely. The bound keeps one unavailable replica or service from starving
+retention and every later replica in the single-flight rotation.
 
 Replica drop, cluster drop, dependency replacement, replica failure, and timeout all
 skip the attempt, and a later sweep recomputes from current state. Read-only
@@ -283,17 +305,23 @@ frontier reaches the target. One of the subscribe's inputs is a replica-local
 introspection log whose frontier the *replica* advances from its own clock, rounded
 up to its introspection interval.
 
-So a replica whose clock trails environmentd by more than one introspection
-interval produces a frontier that keeps sitting below the target the oracle hands
-out. The mutation waits, and the sweep's own timeout ends it. The consequence is
-bounded. Nothing wrong is recorded, that replica records nothing at all, and the
-sweep stretches while it waits, which delays others in the rotation. It resolves
-itself when the clocks converge.
+For a non-empty selection, a replica whose clock trails environmentd by more than
+one introspection interval repeatedly loses the same race. Its frontier eventually
+certifies the current target, but by then keepalives or other writes have advanced
+the oracle past it. The committer refuses the stale target, the OCC loop adopts a
+new one, and the replica has to catch up again. The sweep's timeout normally ends
+that loop. A lower `max_occ_retries` can end it first with a contention error.
+
+Nothing wrong is recorded in either case. That replica records nothing at all, and
+the sweep stretches while it retries, which delays others in the rotation. Empty
+selections exit without entering the conflict loop. The condition resolves itself
+when the clocks converge.
 
 Accepted for now: skew between processes is normally milliseconds while the
-tolerance is a whole introspection interval, and the failure mode is waiting rather
-than wrong data. Because the symptom is otherwise hard to attribute, a step that
-times out logs that the replica's introspection frontier may be trailing.
+tolerance is a whole introspection interval, and the failure mode is retrying rather
+than wrong data. Because the symptom is otherwise hard to attribute, both a timeout
+and retry-budget exhaustion log that the replica's introspection frontier may be
+trailing.
 
 ## Rollout
 
@@ -308,6 +336,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,25 @@ 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,
+        }
+    }
+
     /// 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);