compute: expose replica peak memory and disk usage

Replica resource usage is only visible as periodic orchestrator samples,
roughly one per minute, so a spike between two samples is invisible. A
hydration episode that starts and finishes inside one sampling gap cannot
be recovered at all.

Track a high-water mark instead. `mz_metrics::usage` samples memory, heap
and disk on the existing periodic metrics task and folds monotonic peaks
into process-global state, which a new compute logging dataflow reports as
`mz_introspection.mz_cluster_peak_usage`, one row per replica process. The
peaks are also registered as Prometheus gauges.

Peaks are monotonic for the lifetime of the process and are never reset. A
peak that resets is not composable: whoever reads it first consumes it, a
retried read loses it, and two consumers reading at different times
disagree about the same episode. Monotone peaks still answer the hydration
question, because a replica starts fresh, so at the moment it finishes
hydrating its since-start peak is its hydration peak.

`memory_bytes` is exact, taken from the kernel's own high-water mark
(`getrusage`'s `ru_maxrss`). `heap_bytes` and `disk_bytes` have no
kernel-side equivalent, so they are maxima over samples and are therefore
lower bounds. Sampling runs on the metrics task rather than in the logging
operator, because a timely worker saturated by hydration stops scheduling
its logging operators during exactly the episode whose peak we want.

Sampling cadence is a new dyncfg, `mz_metrics_peak_usage_refresh_interval`,
kept separate from `memory_limiter_interval`, which governs OOM-kill
behavior and must not be retuned for introspection's sake.

Tests: unit tests in `mz_metrics::usage` cover monotonicity, the
`heap_bytes >= memory_bytes` ordering and the unmeasured-value case; a new
`test-peak-usage` workflow in test/cluster asserts one row per process and
that peaks never go backwards.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_016V6ARZQJJ5RpXRci6oY6y9

Author: claude

Cargo.lock

@@ -0,0 +1,77 @@
+# Design: Replica Peak Resource Usage
+
+## Summary
+
+Replica resource usage is visible only as periodic samples taken by the orchestrator, roughly one
+per minute, landing in `mz_cluster_replica_metrics_history`. A memory spike between two samples is
+invisible, so the usage of a hydration episode that starts and finishes inside one sampling gap
+cannot be recovered at all.
+
+This adds a measured high-water mark instead: each replica process tracks the peak memory, heap
+and disk usage it has reached, and compute introspection exposes those peaks as
+`mz_introspection.mz_cluster_peak_usage`.
+
+## Semantics
+
+Peaks are monotonic for the lifetime of the process, and are never reset.
+
+The alternative, max-since-last-flush, makes a window's peak recoverable by taking the maximum of
+the rows in that window. It also makes every reader destructive: whoever reads first consumes the
+value, a retried read loses a peak, and two consumers cannot both see the same episode. In a
+system where the peak is read by ad-hoc SQL, that is not workable.
+
+Monotone peaks still answer the question hydration visibility asks. A replica starts fresh, so at
+the moment it finishes hydrating, its since-start peak *is* its hydration peak. More generally,
+for a monotone series `M`, the peak over a window `(t1, t2]` is `M(t2)` whenever `M(t2) > M(t1)`,
+and is otherwise bounded above by `M(t1)`.
+
+The cost is that peaks live and die with the process. A restarted replica reports the peaks of its
+new process, and the old ones are gone. Persisting peaks across replica lifetimes is deliberately
+not part of this work.
+
+## Precision
+
+`memory_bytes` is exact. `getrusage`'s `ru_maxrss` is a high-water mark the kernel maintains
+itself, so no spike can pass between two of our observations unseen.
+
+`heap_bytes` and `disk_bytes` have no such kernel-side counter, so they are maxima over samples
+and are therefore lower bounds: a spike shorter than the sampling interval can be missed. The
+sampling interval is `mz_metrics_peak_usage_refresh_interval` (5s by default), separate from
+`memory_limiter_interval`, which governs OOM-kill behavior and must not be retuned for
+introspection's sake.
+
+`heap_bytes` folds in the `memory_bytes` peak, since peak heap is at least peak memory. That keeps
+`heap_bytes >= memory_bytes` true even when `ru_maxrss` catches a spike that sampling missed.
+
+## Where the peaks are measured
+
+In `mz_metrics::usage`, on the periodic task that already samples `rusage` and lgalloc stats. That
+task runs on the tokio runtime, independent of the timely workers.
+
+The tempting alternative, folding the maximum inside the compute logging operator, samples exactly
+where sampling is least reliable: a logging operator only runs when its worker schedules it, and a
+worker saturated by hydration is precisely the case whose peak we want. Keeping the fold in the
+sampler also means a slow reader cannot lose a peak, because it reads an already-monotone value
+rather than a series of instantaneous ones.
+
+## How the peaks reach SQL
+
+A new `ComputeLog::PeakUsage` logging dataflow reads the process-global peaks and emits one row
+per process, following `ComputeLog::PrometheusMetrics`: the usage is per-process, not per-worker,
+so one worker per process reports and the rest drop their capability.
+
+The peaks are also registered as Prometheus gauges (`mz_metrics_peak_*_bytes`), which costs
+nothing extra since the sampler already holds the values, and which makes them scrapable without
+going through SQL.
+
+## Alternatives considered
+
+- **Extend `/api/usage-metrics` and `mz_cluster_replica_metrics_history`.** Gets 30-day retention
+  and after-the-fact queryability for free. Rejected as the primary surface because the transport
+  is a poll of an HTTP endpoint by the orchestrator, so reporting a peak means either making the
+  endpoint destructive or duplicating peaks across polls, and because the peaks then inherit the
+  orchestrator's poll cadence. Worth revisiting as the persistence story.
+- **Reuse `mz_cluster_prometheus_metrics`.** The gauges appear there automatically, so this needs
+  no catalog change at all. Rejected as the primary surface: values are untyped `double`, the
+  relation is a debugging escape hatch rather than a documented contract, and it is gated by its
+  own scrape-interval dyncfg, so a config change would silently remove the peaks.

doc/developer/design/20260817_replica_peak_usage.md

@@ -153,6 +153,26 @@ The `mz_compute_operator_durations_histogram` view describes a histogram of the
 <!-- RELATION_SPEC_UNDOCUMENTED mz_introspection.mz_compute_operator_durations_histogram_per_worker -->
 <!-- RELATION_SPEC_UNDOCUMENTED mz_introspection.mz_compute_operator_durations_histogram_raw -->
 
+## `mz_cluster_peak_usage`
+
+The `mz_cluster_peak_usage` source describes the peak resource usage of each process of a cluster
+replica, as a high-water mark measured since the process started. Peaks never decrease and are
+never reset, so a replica that has just finished hydrating reports the peak usage of its
+hydration.
+
+Peaks are measured by sampling, so a spike shorter than the sampling interval can be missed and
+the reported values are lower bounds. The exception is `memory_bytes`, which the operating system
+tracks itself and which is therefore exact. A `NULL` means the value could not be measured, for
+example because the replica has no disk.
+
+<!-- RELATION_SPEC mz_introspection.mz_cluster_peak_usage NO_COMMENTS -->
+| Field          | Type      | Meaning                                                                       |
+|----------------|-----------|-------------------------------------------------------------------------------|
+| `process_id`   | [`uint8`] | The ID of the process within the replica.                                     |
+| `memory_bytes` | [`uint8`] | Peak memory (RAM) usage, in bytes.                                            |
+| `heap_bytes`   | [`uint8`] | Peak heap (RAM + swap) usage, in bytes. Always at least `memory_bytes`.       |
+| `disk_bytes`   | [`uint8`] | Peak disk usage, in bytes.                                                    |
+
 ## `mz_cluster_prometheus_metrics`
 
 The `mz_cluster_prometheus_metrics` source exposes Prometheus metrics collected from each cluster replica process's internal metrics registry.

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

@@ -1371,6 +1371,18 @@ metrics:
   help: user CPU time used
   source: src/metrics/src/rusage.rs
   visibility: internal
+- name: mz_metrics_peak_disk_bytes
+  help: Peak disk usage since process start.
+  source: src/metrics/src/usage.rs
+  visibility: internal
+- name: mz_metrics_peak_heap_bytes
+  help: Peak heap (RAM + swap) usage since process start.
+  source: src/metrics/src/usage.rs
+  visibility: internal
+- name: mz_metrics_peak_memory_bytes
+  help: Peak memory (RAM) usage since process start.
+  source: src/metrics/src/usage.rs
+  visibility: internal
 - name: mz_metrics_update_duration_bucket
   help: The time it took to update lgalloc stats
   labels:

doc/user/data/metrics.yml

@@ -751,6 +751,7 @@ def get_default_system_parameters(
     "mz_metrics_lgalloc_map_refresh_interval",
     "mz_metrics_lgalloc_refresh_interval",
     "mz_metrics_rusage_refresh_interval",
+    "mz_metrics_peak_usage_refresh_interval",
     "compute_peek_response_stash_batch_max_runs",
     "compute_peek_response_stash_read_batch_size_bytes",
     "compute_peek_response_stash_read_memory_budget_bytes",

misc/python/materialize/mzcompose/__init__.py

@@ -3284,6 +3284,7 @@ def __init__(
             "mz_metrics_lgalloc_map_refresh_interval",
             "mz_metrics_lgalloc_refresh_interval",
             "mz_metrics_rusage_refresh_interval",
+            "mz_metrics_peak_usage_refresh_interval",
             "compute_peek_stash_num_batches",
             "compute_peek_stash_batch_size",
             "compute_peek_response_stash_batch_max_runs",

misc/python/materialize/parallel_workload/action.py

@@ -392,6 +392,23 @@ static MIGRATIONS: LazyLock<Vec<MigrationStep>> = LazyLock::new(|| {
             MZ_CATALOG_SCHEMA,
             "mz_audit_events",
         ),
+        // Required because we added the `mz_cluster_peak_usage` builtin log.
+        // make_mz_indexes and make_mz_sources inline the builtin-log set as
+        // VALUES, so adding one changes both MVs' SQL fingerprints. See the
+        // NOTE above: this version must stay at the workspace's current dev
+        // version until the change ships.
+        MigrationStep::replacement(
+            "26.39.0-dev.0",
+            CatalogItemType::MaterializedView,
+            MZ_CATALOG_SCHEMA,
+            "mz_indexes",
+        ),
+        MigrationStep::replacement(
+            "26.39.0-dev.0",
+            CatalogItemType::MaterializedView,
+            MZ_CATALOG_SCHEMA,
+            "mz_sources",
+        ),
     ]
 });
 

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

@@ -1090,6 +1090,7 @@ pub static BUILTINS_STATIC: LazyLock<Vec<Builtin<NameReference>>> = LazyLock::ne
         Builtin::Log(&MZ_COMPUTE_EXPORTS_PER_WORKER),
         Builtin::Log(&MZ_COMPUTE_DATAFLOW_GLOBAL_IDS_PER_WORKER),
         Builtin::Log(&MZ_CLUSTER_PROMETHEUS_METRICS),
+        Builtin::Log(&MZ_CLUSTER_PEAK_USAGE),
         Builtin::Log(&MZ_MESSAGE_COUNTS_RECEIVED_RAW),
         Builtin::Log(&MZ_MESSAGE_COUNTS_SENT_RAW),
         Builtin::Log(&MZ_MESSAGE_BATCH_COUNTS_RECEIVED_RAW),

src/catalog/src/builtin.rs

@@ -248,6 +248,21 @@ pub static MZ_CLUSTER_PROMETHEUS_METRICS: LazyLock<BuiltinLog> = LazyLock::new(|
     }),
 });
 
+pub static MZ_CLUSTER_PEAK_USAGE: LazyLock<BuiltinLog> = LazyLock::new(|| BuiltinLog {
+    name: "mz_cluster_peak_usage",
+    schema: MZ_INTROSPECTION_SCHEMA,
+    oid: oid::LOG_MZ_CLUSTER_PEAK_USAGE_OID,
+    variant: LogVariant::Compute(ComputeLog::PeakUsage),
+    access: vec![PUBLIC_SELECT],
+    ontology: Some(Ontology {
+        entity_name: "cluster_peak_usage",
+        description: "Peak resource usage of each process of the cluster replica, \
+                      measured since the process started.",
+        links: &const { [] },
+        column_semantic_types: &[],
+    }),
+});
+
 pub static MZ_COMPUTE_FRONTIERS_PER_WORKER: LazyLock<BuiltinLog> = LazyLock::new(|| BuiltinLog {
     name: "mz_compute_frontiers_per_worker",
     schema: MZ_INTROSPECTION_SCHEMA,

src/catalog/src/builtin/mz_introspection.rs

@@ -949,6 +949,7 @@ impl<'a> Transaction<'a> {
             LogVariant::Compute(ComputeLog::DataflowGlobal) => 31,
             LogVariant::Compute(ComputeLog::OperatorHydrationStatus) => 32,
             LogVariant::Compute(ComputeLog::PrometheusMetrics) => 33,
+            LogVariant::Compute(ComputeLog::PeakUsage) => 34,
         };
 
         let mut id: u64 = u64::from(cluster_variant) << 56;