adapter: record durable object hydration history

### Motivation

Materialize exposes current hydration state, but it disappears when a dataflow
or replica restarts. A user can see whether an object is hydrated now, but not
how long the last hydration took, which is what they need in order to size a
cluster for the next one.

Ref: SQL-632

### Description

Compute stamps three replica-local wallclock times per export and worker:
dataflow installation, hydration start, and hydration completion. They are added
to the existing `mz_compute_hydration_times_per_worker` log next to `time_ns`,
which keeps that relation's name, OID, and object kind, so its generated
per-replica index and every relation derived from it are untouched.

A new `mz_catalog.mz_object_hydration_history` table records completed episodes
for indexes and materialized views, keyed by object, replica, and installation
time. A background sweep visits one managed user replica per interval, installs a
replica-targeted internal subscribe that diffs that replica's live timestamps
against the history table, and appends what is missing through the timestamped
OCC write path. Including the target table in the read expression is what makes
the write idempotent across concurrent `environmentd` processes: two collectors
racing for one write timestamp leave the loser observing the winner's append and
finding nothing to write. Retention retracts aged-out rows in bounded batches
through the same path.

Rows survive `environmentd` and replica restarts: the table is excluded from the
bootstrap reset and from replacement schema migration.

Collection is off by default in production and on in CI. Recording is best
effort, and the limits are documented rather than implied away: it samples
current state, so an episode whose object is dropped or whose replica restarts
before that replica's next cycle is not recorded, unmanaged replicas are not
sampled, and failed or canceled hydration is not recorded at all. `started_at`
is NULL when an object was never suspended, because no start was observed.
Replica-wide episodes and resource peaks need signals that do not exist yet and
are scoped out in the design doc, not approximated from sampled metrics.

### Verification

`test/testdrive/hydration-status.td` covers lifecycle ordering, several
replicas, deduplication across sweeps, survival of a replica drop, the
deliberate absence of an episode that cleans itself up, and retention. A
`test/restart` workflow covers restart durability without duplication. Unit
tests cover the sweep cursor and the two query invariants that are easy to
regress silently: the retention bound must sit inside a derived table, because
the OCC path discards the plan's `RowSetFinishing`, and a completed row requires
consistent reports from every worker.

Author: aljoscha

Cargo.lock

@@ -0,0 +1,260 @@
+# Durable Object Hydration History
+
+## Context
+
+Materialize exposes current hydration state, but that state disappears when a
+dataflow or replica restarts. A user can tell whether an object is hydrated now,
+but cannot tell how long the previous hydration took.
+
+This design records completed compute-object hydration episodes in a durable
+catalog table. It is the first stage of broader hydration history. Replica-wide
+episodes, resource peaks, failed episodes, and storage objects require signals
+that do not exist yet and are described under Future Work.
+
+## Goals
+
+- Record successful hydration of indexes and materialized views per replica.
+- Preserve history across environmentd and replica restarts.
+- Make writes idempotent across concurrent environmentd processes.
+- Preserve the shape and values of existing hydration relations.
+- Bound storage with configurable retention.
+- Keep collection disabled by default in production while exercising it in CI.
+
+## Non-Goals
+
+- Failed or canceled hydration. A replica cannot report its own crash, and the
+  current-state compute log retracts an object without recording why.
+- Replica-wide episodes. Correct episode boundaries require the initial object
+  set and explicit transitions between fully hydrated and hydrating states.
+- Resource peaks. The existing metrics history contains samples, not true
+  high-water values. Reporting those samples as peaks would be misleading.
+- Source and sink hydration. Storage does not publish equivalent lifecycle
+  timestamps.
+- Unmanaged replicas. Their Timely worker count is unknown, so the collector
+  cannot prove that a cross-worker aggregate is complete.
+
+## Overview
+
+Compute publishes three replica-stamped timestamps for every export and worker:
+
+- `installed_at`, when the dataflow is installed
+- `started_at`, when the dataflow is unsuspended
+- `hydrated_at`, when the output frontier passes the as-of
+
+The three columns are added to the existing
+`mz_introspection.mz_compute_hydration_times_per_worker` log, next to the
+existing `time_ns`. The relation keeps its name, OID, and object kind, so its
+auto-generated per-replica index is unchanged and every aggregate relation built
+on it keeps reading the same columns with the same values. The change is additive
+rather than byte-identical: a consumer doing `SELECT *` sees three new columns.
+
+A rename plus a compatibility view was considered, since that is what the compute
+half of this project proposes. It is not done here. It would move OID 16977 to a
+differently shaped relation, flip the old name from a log to a view, and rename
+the generated per-replica index. Naming that relation is the compute team's
+decision on their own change, and this one does not need it.
+
+A background task visits one managed user replica per interval. It installs an
+internal subscribe on that replica. The subscribe aggregates complete worker
+rows, maps runtime export IDs to catalog object IDs, and anti-joins the result
+against the history table. The task writes the resulting rows at the subscribe
+frontier using the timestamped OCC write path.
+
+## What is and is not recorded
+
+Collection samples current state. It is not an event log, and the compute log it
+reads retracts an object's row when the export goes away. So an episode is
+recorded only if its row is still live when its replica's turn comes around:
+
+- An object dropped before the next sweep of its replica is not recorded.
+- A replica process that restarts before the next sweep loses the episode that
+  preceded the restart.
+- An object that hydrates and immediately advances to the empty frontier, such as
+  a constant materialized view, is never recorded, because its replica-side state
+  is cleaned up as soon as it completes. `test/testdrive/hydration-status.td`
+  asserts this absence explicitly so the limit is visible rather than surprising.
+
+Making these cases durable requires compute to emit hydration transitions into an
+append-only collection that survives until an observer acknowledges them. That is
+compute-side work and is listed under Future Work. Until then the table is
+documented as best effort, which is the honest description of a sampler.
+
+## History Table
+
+`mz_catalog.mz_object_hydration_history` has this shape:
+
+```text
+object_id     text         not null
+cluster_id    text         not null
+replica_id    text         not null
+installed_at  timestamptz  not null
+started_at    timestamptz  null
+finished_at   timestamptz  null
+status        text         not null
+key (object_id, replica_id, installed_at)
+```
+
+The initial implementation writes only terminal `hydrated` rows, so its rows
+have non-null start and finish timestamps. Nullable columns reserve a compatible
+representation for an episode that is canceled before it starts or never
+finishes once those events become observable.
+
+`installed_at` is the episode identity. It is stamped when the replica creates
+the export and remains stable across an environmentd restart. `started_at` is
+not suitable as the identity because it is null while the object waits to run.
+
+## Compute Timestamps
+
+Compute logging event times use a Unix epoch anchor advanced by a monotonic
+clock. Timestamp values come from the event time, not the rounded differential
+timestamp used to publish the update. Logging cadence can delay visibility but
+does not reduce timestamp accuracy.
+
+The row state moves through these forms:
+
+```text
+(installed_at, null, null)
+(installed_at, started_at, null)
+(installed_at, started_at, hydrated_at)
+```
+
+An import-free dataflow is never suspended, so its `Schedule` can arrive after it
+has already hydrated. Recording that late arrival as the start of hydration would
+invent an interval nobody observed, so `started_at` stays NULL instead. A NULL
+start means "not observed", and `installed_at <= started_at <= hydrated_at` holds
+for every non-NULL pair.
+
+Each worker has its own wall-clock anchor. A completed object row uses the
+minimum installation and start times and the maximum completion time across
+workers. The collector requires exactly the configured worker count and requires
+every worker to have reported completion. It additionally requires
+`max(installed_at) <= min(hydrated_at)`, which rejects worker rows that cannot
+belong to one episode: a replica whose processes restarted at different times can
+otherwise present one worker from before the restart and one from after, satisfy
+the count check, and produce a row whose duration includes the downtime between
+two episodes. The resulting interval can still include clock skew between worker
+processes.
+
+Multi-export dataflows would need one more step. All exports of a dataflow share
+a suspension token, so the dataflow only starts once every export is scheduled,
+while the stamp is per export. Every compute dataflow has exactly one export
+today and `sequential_hydration.rs` asserts it, so the two coincide. The code
+records what has to change if that stops holding.
+
+## OCC Collector
+
+The subscribe reads both the replica-local timestamp log and the physical
+history table. Including the target table in the read expression is required for
+distributed idempotence.
+
+Assume two environmentd processes compute the same missing row at frontier `T`:
+
+1. Both submit a timestamped write for `T`.
+2. One write commits and advances the table upper and timestamp oracle.
+3. The other write receives `TimestampPassed`.
+4. Its subscribe observes the committed history row.
+5. The anti-join retracts the candidate, leaving no row to write.
+
+The losing writer never retries stale diffs at the next eligible timestamp. It
+waits for subscribe progress and retries only with state known to be valid at
+the observed frontier.
+
+Background subscribes use the ordinary active-compute-sink lifecycle but have a
+background owner rather than a synthetic SQL session. They do not write an
+`mz_subscriptions` row. Dropping the handle retires the compute dataflow on
+success, error, timeout, or replica failure.
+
+## Scheduling and Failure Handling
+
+`hydration_history_collection_interval` controls cadence. A zero duration
+disables collection. The task waits for each replica attempt to finish before
+scheduling the next one, which bounds compute load and avoids self-contention in
+the globally serialized timestamped-write path.
+
+Fires are aligned to interval boundaries, and each scheduler sleep is capped so
+that a configuration change takes effect within the cap rather than after the
+previous interval has elapsed. Without the cap, lowering a long interval at
+runtime, which tests do, would appear to do nothing.
+
+Each mutation is bounded by its own timeout. The bound has to be generous: a
+mutation that finds nothing to write still waits for its read to linearize, which
+can take a full `default_timestamp_interval`, and that parameter has no upper
+bound. A shared, tighter bound would let a large timestamp interval starve
+retention permanently.
+
+The replica list is refreshed before every attempt. Replica drop, cluster drop,
+dependency replacement, replica failure, and timeout skip the attempt. A later
+cycle recomputes from current state. Read-only environmentd generations do not
+install subscribes or buffer writes.
+
+Managed replicas expose their total worker count. Unmanaged replicas are skipped,
+because without that count the collector cannot distinguish a complete
+cross-worker aggregate from a partial one, and the public documentation says so.
+
+## Retention and Restart
+
+`hydration_history_retention_period` defaults to 30 days. Retention is another
+OCC mutation. It subscribes to rows with `finished_at` before the cutoff and
+writes their retractions at the observed frontier. The insertion query applies
+the same cutoff so a current-state log cannot resurrect a row just removed by
+retention.
+
+Retention deletes a bounded batch per sweep, and converges over as many sweeps as
+it takes. It has to be bounded: 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 on every sweep and never shrink the table.
+The bound lives inside a derived table, because a top-level `LIMIT` lands in the
+plan's `RowSetFinishing`, which the OCC path deliberately discards.
+
+Retention runs on the catalog server cluster, so it keeps working when there are
+no user replicas at all, and it runs even when that sweep's replica collection
+failed. A crash-looping replica must not be able to stop the table from shrinking
+back to its bound.
+
+Disabling collection also suspends retention. The alternative is an always-on
+background subscribe in the default configuration, where the table is empty and
+there is nothing to retain. The consequence is that rows already collected are
+kept while collection is off, which the user-facing documentation states.
+
+Builtin tables are normally reset during environmentd bootstrap. The history
+table is explicitly retained and is protected from replacement schema
+migrations, which would otherwise allocate a fresh shard and discard its data.
+
+## Rollout
+
+The collection interval defaults to zero in production. Test and CI defaults
+enable collection so hydration, restart, retention, and catalog tests exercise
+the path. Runtime configuration can enable the collector without restarting
+environmentd.
+
+## Open Questions
+
+Two decisions were made narrowly and are worth confirming.
+
+Whether best-effort sampling is enough to ship. The sampler cannot see an episode
+whose evidence is already gone, so a user watching a churning cluster sees gaps.
+The alternative is to wait for compute to publish durable hydration transitions
+and build this on those instead. Shipping the sampler first gives the common case
+(an object that hydrates and stays hydrated) with no compute-side protocol work,
+at the cost of a table that is honest but incomplete.
+
+Where collection should be enabled in tests. It is on in the mzcompose
+configuration at a 60 second interval, and off in the sqllogictest runner's
+defaults. Turning it on for sqllogictest as well would exercise the path more
+broadly, but the collector installs subscribes and writes to a builtin table, and
+those runs assert on catalog contents and plans, so it risks plan churn and
+timing flakiness in files that have nothing to do with hydration.
+
+## Future Work
+
+The full hydration visibility surface needs additional durable events:
+
+- Record installation and start observations before completion, then finalize
+  canceled and failed episodes by joining replica lifecycle events.
+- Define a replica episode state machine using the object set present when the
+  replica transitions from fully hydrated to hydrating.
+- Publish resettable per-process high-water values for RAM, swap, and scratch
+  disk. Define replica aggregation without pretending that sampled maxima are
+  simultaneous peaks.
+- Give storage objects equivalent lifecycle timestamps.
+- Build replica history and progress views from those authoritative signals.

doc/developer/design/20260817_durable_object_hydration_history.md

@@ -370,6 +370,33 @@ Field          | Type                 | Meaning
 `create_sql`   | [`text`]             | The `CREATE` SQL statement for the materialized view.
 `redacted_create_sql` | [`text`]      | The redacted `CREATE` SQL statement for the materialized view.
 
+### `mz_object_hydration_history`
+
+The `mz_object_hydration_history` table contains completed hydration episodes
+for indexes and materialized views. Rows survive environmentd and replica
+restarts.
+
+Collection is off by default. It is enabled by setting the
+`hydration_history_collection_interval` configuration parameter to a non-zero
+interval, and rows are retained for 30 days by default while it stays enabled.
+
+Recording is best effort. Each cycle samples one managed replica's current
+hydration state, so an episode whose object is dropped, or whose replica process
+restarts, before that replica's next cycle is not recorded. Replicas of
+unmanaged clusters are not sampled at all. Only successful hydration is
+recorded, so there are no rows for failed or canceled hydration.
+
+<!-- RELATION_SPEC mz_catalog.mz_object_hydration_history -->
+Field          | Type                          | Meaning
+---------------|-------------------------------|--------
+`object_id`    | [`text`]                      | The ID of the index or materialized view.
+`cluster_id`   | [`text`]                      | The ID of the object's cluster.
+`replica_id`   | [`text`]                      | The ID of the cluster replica.
+`installed_at` | [`timestamp with time zone`]  | When the object's dataflow was installed on the replica.
+`started_at`   | [`timestamp with time zone`]  | When hydration work began, or `NULL` if the object was never suspended and so no start was observed.
+`finished_at`  | [`timestamp with time zone`]  | When hydration finished.
+`status`       | [`text`]                      | The terminal status. Currently always `hydrated`.
+
 ### `mz_objects`
 
 The `mz_objects` view contains a row for each table, source, view, materialized

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

@@ -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

doc/user/data/metrics.yml

@@ -137,6 +137,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.39.0-dev"):
+        config["hydration_history_collection_interval"] = "60s"
+
     if sanitizer_enabled():
         config["with_0dt_deployment_max_wait"] = "18000s"
 
@@ -457,6 +460,9 @@ def get_variable_system_parameters(
         VariableSystemParameter(
             "arrangement_size_history_retention_period", "7d", ["1min", "1h", "7d"]
         ),
+        VariableSystemParameter(
+            "hydration_history_retention_period", "30d", ["1min", "1h", "30d"]
+        ),
         VariableSystemParameter(
             "persist_validate_part_bounds_on_read", "false", ["true", "false"]
         ),

misc/python/materialize/mzcompose/__init__.py

@@ -2949,6 +2949,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/parallel_workload/action.py

@@ -343,6 +343,20 @@ pub const ARRANGEMENT_SIZE_HISTORY_RETENTION_PERIOD: Config<Duration> = Config::
     "How long to retain rows in mz_internal.mz_object_arrangement_size_history.",
 );
 
+/// 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.",
+);
+
+/// 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_catalog.mz_object_hydration_history.",
+);
+
 /// How frequently the catalog `*_info` metrics (`mz_object_info`,
 /// `mz_cluster_info`, …) are reconciled with the catalog. A zero duration
 /// disables reconciliation.
@@ -468,6 +482,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)