adapter: collect durable replica hydration history

doc/developer/design/20260827_durable_replica_hydration_history.md

@@ -0,0 +1,79 @@
+# Durable Replica Hydration History
+
+## Context
+
+The [durable object hydration history](20260817_durable_object_hydration_history.md)
+records successful hydration for individual dataflows. This design extends that
+collector with replica-wide hydration episodes and the resource high-water marks
+visible when each episode is recorded.
+
+The extension reuses the object collector's scheduling, replica-targeted
+read-then-write path, exact-timestamp OCC, retention, and migration protections.
+This document describes only the replica-level additions.
+
+## Episode boundaries
+
+Each live non-transient compute export contributes an interval from its earliest
+worker installation to its latest worker hydration. The collector waits until
+every visible worker of every live non-transient export has hydrated. Transient
+query dataflows are excluded because the collection query itself creates one.
+
+A replica episode is a connected component in the union of those export
+intervals. Two intervals belong to one episode if they overlap directly or
+through a chain of overlapping intervals. A gap means the replica was fully
+hydrated before the next export was installed, so the next interval starts a
+new episode.
+
+Each sweep records only the latest completed episode visible in its snapshot.
+Episodes that complete between sweeps and exports that retract before a sweep
+leave no evidence. The current inputs can therefore record successful episodes
+only. Failed, canceled, and OOM-killed outcomes need an additional durable
+replica signal.
+
+Process-local clocks stamp both interval endpoints. Clock skew can merge
+episodes that did not overlap in real time. Once an episode is recorded, a
+monotonic history guard prevents a later snapshot from interpreting retracted
+intervals as an earlier or overlapping episode.
+
+## Resource interpretation
+
+`peak_memory_bytes` is the maximum `cgroup memory_peak` across replica
+processes. `peak_disk_bytes` is the maximum sampled `statvfs fs_used_peak` when a
+scratch filesystem is present. Otherwise it is the maximum kernel-maintained
+`cgroup swap_peak`.
+
+Replica memory and disk limits apply independently to each process. The maximum
+process peak therefore answers whether any process approached its limit. Adding
+process maxima would combine peaks that may not have occurred simultaneously.
+
+The operating system's peaks cover the process lifetime through the collector's
+observation. They are not bounded by `finished_at`, so work after hydration and
+before collection can raise them. Later episodes can also include an earlier
+high-water mark. A true episode peak requires a reset or a separately retained
+interval maximum at the replica.
+
+The collector requires at least one resource observation from every configured
+process before writing. Individual peak metrics can still be absent, which is
+represented by `NULL` rather than a zero sentinel.
+
+## History table
+
+```text
+mz_internal.mz_replica_hydration_history
+  replica_id         text         not null
+  cluster_id         text         not null
+  started_at         timestamptz  not null
+  finished_at        timestamptz  null
+  object_count       uint8        not null
+  peak_memory_bytes  uint8        null
+  peak_disk_bytes    uint8        null
+  status             text         not null
+```
+
+An episode is identified operationally by `(replica_id, started_at)`. The table
+does not declare a key or index. Collection runs on the selected replica, so a
+catalog-server index would not avoid importing and arranging the history there.
+
+Rows currently have a populated `finished_at` and the status `hydrated`.
+Resource columns are nullable because the available kernel and filesystem
+observations depend on the replica platform.

doc/user/content/reference/system-catalog/mz_internal.md

@@ -743,6 +743,31 @@ logical timestamp, so the recorded finish can precede the latest process's finis
 | `hydrated_at`  | [`timestamp with time zone`] | When hydration finished.                                                                                                 |
 | `status`       | [`text`]                     | The terminal status. Currently always `hydrated`.                                                                        |
 
+## `mz_replica_hydration_history`
+
+The `mz_replica_hydration_history` table records successful replica hydration
+episodes. An episode begins when a maintained compute dataflow is installed on
+a fully hydrated replica and finishes when every running maintained compute
+dataflow has hydrated.
+
+By default, rows are retained for 30 days while collection is enabled. Recording
+is best effort, and only the latest completed episode visible in each collection
+is recorded. Resource peaks cover the replica processes' lifetimes through
+collection, not only the hydration episode. On a multi-process replica, the
+table records the largest peak reported by any process.
+
+<!-- RELATION_SPEC mz_internal.mz_replica_hydration_history -->
+| Field               | Type                         | Meaning                                                                                                                  |
+| ------------------- | ---------------------------- | ------------------------------------------------------------------------------------------------------------------------ |
+| `replica_id`        | [`text`]                     | The ID of the cluster replica. May name a replica that no longer exists.                                                 |
+| `cluster_id`        | [`text`]                     | The ID of the replica's cluster.                                                                                          |
+| `started_at`        | [`timestamp with time zone`] | The earliest maintained compute dataflow installation in the hydration episode.                                         |
+| `finished_at`       | [`timestamp with time zone`] | The latest maintained compute dataflow hydration in the hydration episode.                                               |
+| `object_count`      | [`uint8`]                    | The number of maintained compute dataflows in the hydration episode.                                                     |
+| `peak_memory_bytes` | [`uint8`]                    | The largest process-lifetime cgroup memory high-water mark reported by any process when the collector recorded the episode. `NULL` if the platform reports no cgroup memory peak. |
+| `peak_disk_bytes`   | [`uint8`]                    | The largest process-lifetime scratch-filesystem or swap high-water mark reported by any process when the collector recorded the episode. Filesystem peaks are sampled lower bounds. `NULL` if neither measurement is available. |
+| `status`            | [`text`]                     | The hydration episode's status. Currently always `hydrated`.                                                             |
+
 ## `mz_object_transitive_dependencies`
 
 The `mz_object_transitive_dependencies` view describes the transitive dependency structure between

doc/user/data/metrics.yml

@@ -963,7 +963,7 @@ metrics:
   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.
+  help: Total hydration-history retention batches that were 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

src/adapter-types/src/dyncfgs.rs

@@ -429,11 +429,11 @@ pub const HYDRATION_HISTORY_COLLECTION_INTERVAL: Config<Duration> = Config::new(
     ParameterScope::Environment,
 );
 
-/// How long to retain completed object hydration episodes.
+/// How long to retain completed object and replica 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.",
+    "How long to retain rows in mz_internal.mz_object_hydration_history and mz_internal.mz_replica_hydration_history.",
     ParameterScope::Environment,
 );
 

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

@@ -39,6 +39,7 @@ use mz_catalog::builtin::{
     BUILTIN_LOOKUP, Builtin, Fingerprint, MZ_CATALOG_RAW, MZ_CATALOG_RAW_DESCRIPTION,
     MZ_CLUSTER_REPLICA_FRONTIERS_DESCRIPTION, MZ_OBJECT_ARRANGEMENT_SIZE_HISTORY_DESCRIPTION,
     MZ_OBJECT_HYDRATION_HISTORY, MZ_OBJECT_HYDRATION_HISTORY_DESCRIPTION,
+    MZ_REPLICA_HYDRATION_HISTORY, MZ_REPLICA_HYDRATION_HISTORY_DESCRIPTION,
     MZ_STORAGE_USAGE_BY_SHARD, MZ_STORAGE_USAGE_BY_SHARD_DESCRIPTION,
     RUNTIME_ALTERABLE_FINGERPRINT_SENTINEL,
 };
@@ -689,18 +690,20 @@ fn participates_in_forced_migration(
     match builtin {
         // A forced replacement allocates a fresh shard, which discards the
         // table's contents. Exclude the tables whose contents are the point:
-        // storage usage is retained for billing, and hydration history cannot
+        // storage usage is retained for billing, and hydration histories cannot
         // be rebuilt from any other source.
         //
-        // Hydration history takes part in a forced `Evolution`, which keeps the
-        // rows. It has to: dev upgrades force one for every object, and a table
-        // left out of the plan never gets its new schema registered, so
-        // `update_fingerprints` panics at open as soon as the desc changes. See
-        // the tripwire in `validate_migration_steps` for how to give up the
+        // Hydration history tables take part in a forced `Evolution`, which
+        // keeps the rows. They have to: dev upgrades force one for every object,
+        // and a table left out of the plan never gets its new schema registered.
+        // `update_fingerprints` then panics at open as soon as the desc changes.
+        // See the tripwire in `validate_migration_steps` for how to give up the
         // replacement exemption deliberately.
         Table(table) => {
             **table != *MZ_STORAGE_USAGE_BY_SHARD
-                && (mechanism != Mechanism::Replacement || **table != *MZ_OBJECT_HYDRATION_HISTORY)
+                && (mechanism != Mechanism::Replacement
+                    || (**table != *MZ_OBJECT_HYDRATION_HISTORY
+                        && **table != *MZ_REPLICA_HYDRATION_HISTORY))
         }
         MaterializedView(..) => true,
         Source(source) => **source != *MZ_CATALOG_RAW,
@@ -819,6 +822,10 @@ impl Migration {
                     &*MZ_OBJECT_HYDRATION_HISTORY_DESCRIPTION, object,
                     "replacing mz_object_hydration_history clears it, see the comment above"
                 );
+                assert_ne!(
+                    &*MZ_REPLICA_HYDRATION_HISTORY_DESCRIPTION, object,
+                    "replacing mz_replica_hydration_history clears it, see the comment above"
+                );
             }
 
             // `mz_catalog_raw` cannot be migrated because it contains the durable catalog and it

src/adapter/src/catalog/open/builtin_schema_migration_tests.rs

@@ -29,16 +29,21 @@ use super::*;
 
 #[mz_ore::test]
 fn hydration_history_forced_migration_policy() {
-    let hydration_history = Builtin::Table(&*MZ_OBJECT_HYDRATION_HISTORY);
-
-    assert!(participates_in_forced_migration(
-        &hydration_history,
-        Mechanism::Evolution
-    ));
-    assert!(!participates_in_forced_migration(
-        &hydration_history,
-        Mechanism::Replacement
-    ));
+    for table in [
+        &*MZ_OBJECT_HYDRATION_HISTORY,
+        &*MZ_REPLICA_HYDRATION_HISTORY,
+    ] {
+        let hydration_history = Builtin::Table(table);
+
+        assert!(participates_in_forced_migration(
+            &hydration_history,
+            Mechanism::Evolution
+        ));
+        assert!(!participates_in_forced_migration(
+            &hydration_history,
+            Mechanism::Replacement
+        ));
+    }
 }
 
 #[test] // allow(test-attribute)

src/adapter/src/coord.rs

@@ -100,7 +100,7 @@ use mz_auth::password::Password;
 use mz_build_info::BuildInfo;
 use mz_catalog::builtin::{
     BUILTINS, BUILTINS_STATIC, MZ_OBJECT_ARRANGEMENT_SIZE_HISTORY, MZ_OBJECT_HYDRATION_HISTORY,
-    MZ_STORAGE_USAGE_BY_SHARD,
+    MZ_REPLICA_HYDRATION_HISTORY, MZ_STORAGE_USAGE_BY_SHARD,
 };
 use mz_catalog::config::{AwsPrincipalContext, BuiltinItemMigrationConfig, ClusterReplicaSizeMap};
 use mz_catalog::durable::OpenableDurableCatalogState;
@@ -3186,6 +3186,8 @@ impl Coordinator {
                 .resolve_builtin_table(&MZ_OBJECT_ARRANGEMENT_SIZE_HISTORY),
             self.catalog()
                 .resolve_builtin_table(&MZ_OBJECT_HYDRATION_HISTORY),
+            self.catalog()
+                .resolve_builtin_table(&MZ_REPLICA_HYDRATION_HISTORY),
         ]);
 
         let mut retraction_tasks = Vec::new();
@@ -4337,6 +4339,14 @@ impl Coordinator {
                     }
                 }
             }
+
+            // The sweep can own timestamp-oracle senders through its background
+            // client. Release them before the coordinator runtime starts shutting
+            // down the oracle workers.
+            if let Some(sweep) = self.hydration_history_sweep.take() {
+                sweep.abort_and_wait().await;
+            }
+
             // Try and cleanup as a best effort. There may be some async tasks out there holding a
             // reference that prevents us from cleaning up.
             if let Some(catalog) = Arc::into_inner(self.catalog) {