adapter: collect durable object hydration history

doc/developer/design/20260210_incremental_occ_read_then_write.md

@@ -235,6 +235,11 @@ The correctness argument has two parts: (1) the OCC loop produces the right
 diffs, and (2) the timestamped write mechanism ensures they are applied at the
 right timestamp.
 
+User statements require every relation leaf to be a user table owned by the
+table-write transaction domain. Source-backed tables, including webhook tables,
+receive updates outside that domain, so the target table's oracle cannot make
+their updates atomic with the resulting write.
+
 ### The subscribe produces the right diffs
 
 The subscribe starts at the oracle read timestamp and emits the current state

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,15 +202,19 @@ 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
-`RowSetFinishing`, which the OCC path deliberately discards, and the delete would be
-silently unbounded again.
+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 this OCC stage cannot apply.
+
+`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
@@ -238,6 +254,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 +271,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 +304,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
 

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"] = [

misc/python/materialize/sqlancer.py

@@ -108,7 +108,7 @@
     r"null character not permitted",
     r"expressions must appear in select list",
     r"expressions must match initial",
-    r"invalid selection: operation may only",
+    r"invalid selection: every relation leaf",
     r"array size exceeds the maximum allowed",
     r"does not allow subqueries",
     r"must use value within",

misc/python/materialize/sqlsmith.py

@@ -83,7 +83,7 @@
     "result exceeds max size of",  # Seems expected with huge queries
     "expected expression, but found reserved keyword",  # Should fix, but only happens rarely with subqueries
     "Expected right parenthesis, found left parenthesis",  # Should fix, but only happens rarely with cast+coalesce
-    "invalid selection: operation may only (transitively) refer to non-source, non-system tables",  # Seems expected when using catalog tables
+    "invalid selection: every relation leaf must be a writable user table",  # Expected when using catalog tables
     "Unsupported temporal predicate",  # Expected, see https://github.com/MaterializeInc/database-issues/issues/5288
     "Unsupported temporal operation: NotEq",
     "Unsupported binary temporal operation: NotEq",

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);