fix: keep Stream metric scrapes off durable storage

docs/admin/api/overview-probes-metrics.md

@@ -220,6 +220,12 @@ GET /metrics
 **Listener**: `FITZ_METRICS_BIND_ADDR:FITZ_METRICS_PORT`
 **Authentication**: None; keep this listener private to the scrape network.
 **Response**: Prometheus text format
+
+Scrapes read an in-process Stream metrics projection initialized during startup
+and advanced by successful commits and persisted watermark updates. They do not
+scan durable Stream inventory or enqueue admin work on a family actor, so a slow
+storage backend cannot turn observability polling into data-plane backpressure.
+
 ```
 # HELP fitz_connections_total Total number of active connections
 # TYPE fitz_connections_total gauge

docs/development/architecture.md

@@ -109,6 +109,9 @@ Raw Prometheus is served by the dedicated unauthenticated
 listener returns `404` for `/metrics`. Admin consumers use structured JSON at
 `/api/v1/{family}/metrics`; broker-global samples are available only at
 `/api/v1/all/metrics` with wildcard authority.
+Prometheus rendering reads Stream counts and durable progress from in-process
+metric projections initialized during startup and advanced on committed work; a
+scrape must not scan storage or enqueue admin work onto a domain actor.
 ### Critical Invariant: Ephemeral Sessions
 > **Fitz sessions are ephemeral. The broker never restores session state after disconnect. Clients are responsible for rebuilding all state including subscriptions, transactions, workers, leases, and stream resume position.**
 

docs/development/routing-design.md

@@ -505,7 +505,8 @@ Posting indexes preserve the order of their parent scope:
   contending on a mutable tail page.
 - Background compaction may merge adjacent fragments into larger pages after
   they are below the governing watermark. Readers accept both representations.
-- One synchronous maintenance slice examines at most eight buckets or 4 MiB.
+- One synchronous maintenance slice examines at most one bucket or 4 MiB so
+  strict storage commits yield to client commands between buckets.
   Successful commits enqueue only their touched bucket prefixes. The first
   maintenance slice after restart rebuilds pending work with one lazy family
   scan; later slices consume the queue without rescanning the family history.

src/api/admin/metrics/domains/stream.rs

@@ -4,15 +4,25 @@ use std::fmt::Write as _;
 use super::super::rendering::encode_prometheus_label_value;
 
 pub(super) fn append_metrics(output: &mut String, runtime: &Runtime) {
-    append_core_metrics(output, runtime);
-    append_lag_bucket_metrics(output, runtime);
-    append_watermark_metrics(output, runtime);
+    let durable = runtime.stream_durable_metrics_snapshot();
+    append_core_metrics(output, runtime, durable.as_ref());
+    append_lag_bucket_metrics(output, runtime, durable.as_ref());
+    append_watermark_metrics(output, runtime, durable.as_ref());
 }
 
-fn append_core_metrics(output: &mut String, runtime: &Runtime) {
+fn append_core_metrics(
+    output: &mut String,
+    runtime: &Runtime,
+    durable: Option<&crate::domains::stream::metrics::StreamDurableMetricsSnapshot>,
+) {
+    let metrics = crate::observability::metrics();
     output.push_str("# HELP fitz_stream_active Active streams\n");
     output.push_str("# TYPE fitz_stream_active gauge\n");
-    let _ = writeln!(output, "fitz_stream_active {}", runtime.stream_active());
+    let _ = writeln!(
+        output,
+        "fitz_stream_active {}",
+        metrics.gauge_get(crate::domains::stream::metrics::METRIC_ACTIVE_GAUGE)
+    );
     output.push('\n');
 
     output.push_str("# HELP fitz_stream_response_drops_total Total Stream responses dropped by this broker process\n# TYPE fitz_stream_response_drops_total counter\n");
@@ -30,7 +40,7 @@ fn append_core_metrics(output: &mut String, runtime: &Runtime) {
     let _ = writeln!(
         output,
         "fitz_stream_append_sessions_active {}",
-        runtime.stream_append_sessions_active()
+        metrics.gauge_get(crate::domains::stream::metrics::METRIC_APPEND_SESSIONS_GAUGE)
     );
     output.push('\n');
 
@@ -39,7 +49,10 @@ fn append_core_metrics(output: &mut String, runtime: &Runtime) {
     let _ = writeln!(
         output,
         "fitz_stream_events_total {}",
-        runtime.stream_events_total()
+        durable.map_or_else(
+            || runtime.admin_read_model().stream_events_total(),
+            |snapshot| snapshot.events_total,
+        )
     );
     output.push('\n');
 
@@ -81,7 +94,7 @@ fn append_core_metrics(output: &mut String, runtime: &Runtime) {
     let _ = writeln!(
         output,
         "fitz_stream_subscriptions_active {}",
-        runtime.stream_subscriptions_active()
+        metrics.gauge_get(crate::domains::stream::metrics::METRIC_SUBSCRIPTIONS_GAUGE)
     );
     output.push('\n');
 
@@ -95,8 +108,15 @@ fn append_core_metrics(output: &mut String, runtime: &Runtime) {
     output.push('\n');
 }
 
-fn append_lag_bucket_metrics(output: &mut String, runtime: &Runtime) {
-    let watermark_lag_buckets = runtime.stream_watermark_lag_buckets();
+fn append_lag_bucket_metrics(
+    output: &mut String,
+    runtime: &Runtime,
+    durable: Option<&crate::domains::stream::metrics::StreamDurableMetricsSnapshot>,
+) {
+    let watermark_lag_buckets = durable.map_or_else(
+        || runtime.stream_watermark_lag_buckets(),
+        crate::domains::stream::metrics::StreamDurableMetricsSnapshot::watermark_lag_buckets,
+    );
     output.push_str("# HELP fitz_stream_watermark_lag_bucket_caught_up Stream family watermarks aligned with the fastest family in their area\n");
     output.push_str("# TYPE fitz_stream_watermark_lag_bucket_caught_up gauge\n");
     let _ = writeln!(
@@ -134,28 +154,66 @@ fn append_lag_bucket_metrics(output: &mut String, runtime: &Runtime) {
     output.push('\n');
 }
 
-fn append_watermark_metrics(output: &mut String, runtime: &Runtime) {
+fn append_watermark_metrics(
+    output: &mut String,
+    runtime: &Runtime,
+    durable: Option<&crate::domains::stream::metrics::StreamDurableMetricsSnapshot>,
+) {
     output.push_str(
         "# HELP fitz_stream_realm_watermark Highest committed realm watermark per Stream route family and realm\n",
     );
     output.push_str("# TYPE fitz_stream_realm_watermark gauge\n");
-    for detail in runtime.stream_list_realm_watermark_details() {
-        let realm = encode_prometheus_label_value(&detail.realm);
-        for watermark in detail.family_watermarks {
+    if let Some(snapshot) = durable {
+        for metric in &snapshot.realm_watermarks {
             let _ = writeln!(
                 output,
                 "fitz_stream_realm_watermark{{realm=\"{}\",family=\"{}\"}} {}",
-                realm, watermark.family, watermark.watermark
+                encode_prometheus_label_value(&metric.realm),
+                metric.family,
+                metric.watermark
             );
         }
+    } else {
+        append_cached_realm_watermarks(output, runtime);
     }
     output.push('\n');
 
     output.push_str(
         "# HELP fitz_stream_area_watermark Highest committed area watermark per Stream route family, realm, and area\n",
     );
     output.push_str("# TYPE fitz_stream_area_watermark gauge\n");
-    for detail in runtime.stream_list_area_watermark_details() {
+    if let Some(snapshot) = durable {
+        for metric in &snapshot.area_watermarks {
+            let _ = writeln!(
+                output,
+                "fitz_stream_area_watermark{{realm=\"{}\",area=\"{}\",family=\"{}\"}} {}",
+                encode_prometheus_label_value(&metric.realm),
+                encode_prometheus_label_value(&metric.area),
+                metric.family,
+                metric.watermark
+            );
+        }
+    } else {
+        append_cached_area_watermarks(output, runtime);
+    }
+    output.push('\n');
+}
+
+fn append_cached_realm_watermarks(output: &mut String, runtime: &Runtime) {
+    for detail in runtime.admin_read_model().stream_realm_watermarks() {
+        let realm = encode_prometheus_label_value(&detail.realm);
+        for watermark in detail.family_watermarks {
+            let _ = writeln!(
+                output,
+                "fitz_stream_realm_watermark{{realm=\"{}\",family=\"{}\"}} {}",
+                realm, watermark.family, watermark.watermark
+            );
+        }
+    }
+}
+
+fn append_cached_area_watermarks(output: &mut String, runtime: &Runtime) {
+    for detail in runtime.admin_read_model().stream_area_watermarks() {
         let realm = encode_prometheus_label_value(&detail.realm);
         let area = encode_prometheus_label_value(&detail.area);
         for watermark in detail.family_watermarks {
@@ -166,5 +224,4 @@ fn append_watermark_metrics(output: &mut String, runtime: &Runtime) {
             );
         }
     }
-    output.push('\n');
 }

src/api/admin/metrics/mod.rs

@@ -44,6 +44,7 @@ pub(crate) struct StructuredMetricSample {
 
 /// Handle the authenticated structured metrics contract.
 pub(crate) fn handle_structured_metrics(runtime: &Runtime, family: Option<u64>) -> Response {
+    runtime.refresh_stream_admin_snapshot();
     let mut samples = structured_samples(&generate_prometheus_metrics(runtime), family);
     if let Some(family) = family {
         samples.extend(family_attributable_samples(runtime, family));

src/api/admin/metrics/tests.rs

@@ -218,6 +218,36 @@ fn should_export_schedule_metrics_given_preloaded_schedule_runtime() {
     assert_metric_exported(&metrics, "fitz_notice_wildcard_limit_rejects_total");
 }
 
+#[test]
+fn should_render_stream_metrics_from_cached_observability_state() {
+    // Arrange
+    let metrics = crate::observability::metrics();
+    metrics.gauge_set(crate::domains::stream::metrics::METRIC_ACTIVE_GAUGE, 7);
+    metrics.gauge_set(
+        crate::domains::stream::metrics::METRIC_APPEND_SESSIONS_GAUGE,
+        5,
+    );
+    metrics.gauge_set(
+        crate::domains::stream::metrics::METRIC_SUBSCRIPTIONS_GAUGE,
+        3,
+    );
+    let read_model = crate::control::admin::read_model::AdminReadModel::new();
+    read_model.replace_stream_events_total(11);
+    let runtime = Arc::new(Runtime::with_admin_read_model(
+        Arc::new(Router::new()),
+        read_model,
+    ));
+
+    // Act
+    let payload = generate_prometheus_metrics(&runtime);
+
+    // Assert
+    assert!(payload.contains("fitz_stream_active 7"));
+    assert!(payload.contains("fitz_stream_append_sessions_active 5"));
+    assert!(payload.contains("fitz_stream_subscriptions_active 3"));
+    assert!(payload.contains("fitz_stream_events_total 11"));
+}
+
 #[test]
 fn should_export_type_metadata_for_every_metric_family() {
     // Arrange

src/boot/domains.rs

@@ -261,6 +261,12 @@ impl DomainHandles {
         self.stream.refresh_admin_snapshot_if_dirty();
     }
 
+    pub(crate) fn stream_durable_metrics_snapshot(
+        &self,
+    ) -> crate::domains::stream::metrics::StreamDurableMetricsSnapshot {
+        self.stream.durable_metrics_snapshot()
+    }
+
     pub(crate) fn kv_active_transaction_count(&self) -> usize {
         self.kv.active_transaction_count()
     }
@@ -560,6 +566,7 @@ pub fn setup(
         &route_families,
         &metrics,
     )?;
+    stream_sink.initialize_admin_snapshot();
     register_domain_sink(DomainKind::Stream, router, stream_sink.clone());
 
     let rpc_sink = Arc::new(

src/boot/stats/admin_queries.rs

@@ -41,6 +41,15 @@ impl Runtime {
         }
     }
 
+    pub(crate) fn stream_durable_metrics_snapshot(
+        &self,
+    ) -> Option<crate::domains::stream::metrics::StreamDurableMetricsSnapshot> {
+        self.domains
+            .read()
+            .as_ref()
+            .map(|domains| domains.stream_durable_metrics_snapshot())
+    }
+
     #[must_use]
     pub fn kv_list_transactions(
         &self,
@@ -201,13 +210,6 @@ impl Runtime {
         domains.stream_admin_read_resource_records(request)
     }
 
-    pub(crate) fn stream_list_realm_watermark_details(
-        &self,
-    ) -> Vec<crate::control::admin::StreamRealmWatermarkDetail> {
-        self.refresh_stream_admin_snapshot();
-        self.admin_read_model.stream_realm_watermarks()
-    }
-
     pub(crate) fn stream_realm_watermark_detail(
         &self,
         realm: &str,

src/boot/storage/tests.rs

@@ -137,8 +137,13 @@ fn should_apply_cloud_throughput_defaults_when_memtable_is_auto() {
         "tests",
     )
     .memory_budget(MemoryBudget::Bytes(512 * 1024 * 1024));
-    let expected_memtable_bytes =
-        (512 * 1024 * 1024usize).saturating_sub((512 * 1024 * 1024usize) / 10) / 2;
+    let memory_budget_bytes = 512 * 1024 * 1024usize;
+    let transaction_pool_bytes = memory_budget_bytes / 10;
+    let compaction_pool_bytes = memory_budget_bytes / 10;
+    let expected_memtable_bytes = memory_budget_bytes
+        .saturating_sub(transaction_pool_bytes)
+        .saturating_sub(compaction_pool_bytes)
+        / 2;
 
     // Act
     let tuned = build_midge_open_options(open_options, &config).expect("build cloud options");

src/domains/stream/area_actor.rs

@@ -31,6 +31,8 @@ pub struct AreaActor {
     /// Storage layer for watermark persistence
     store: Arc<StreamStore>,
 
+    durable_metrics: Arc<super::metrics::StreamDurableMetrics>,
+
     /// Area watermark (highest contiguous committed offset).
     ///
     /// `None` means no offset has committed yet. Offset 0 is a valid committed
@@ -57,6 +59,7 @@ impl AreaActor {
         realm: String,
         area: String,
         store: Arc<StreamStore>,
+        durable_metrics: Arc<super::metrics::StreamDurableMetrics>,
     ) -> Self {
         let (area_watermark, watermark_initialized) = store
             .get_persisted_area_watermark(family_id.as_u64(), &realm, &area)
@@ -80,6 +83,7 @@ impl AreaActor {
             realm,
             area,
             store,
+            durable_metrics,
             area_watermark,
             watermark_initialized,
             committed_ranges: BTreeMap::new(),
@@ -188,6 +192,12 @@ impl AreaActor {
     fn apply_persisted_watermark(&mut self, current_watermark: u64, ctx: &mut Context<Self>) {
         let previous_watermark = self.area_watermark.unwrap_or(0);
         self.area_watermark = Some(current_watermark);
+        self.durable_metrics.set_area_watermark(
+            self.family_id.as_u64(),
+            &self.realm,
+            &self.area,
+            current_watermark,
+        );
         self.committed_ranges
             .retain(|_, last_offset| *last_offset > current_watermark);
 
@@ -301,7 +311,13 @@ mod tests {
                 .set_watermark(family.as_u64(), "realm1", "area1", watermark)
                 .expect("persist area watermark");
         }
-        let actor = AreaActor::new(family, "realm1".to_string(), "area1".to_string(), store);
+        let actor = AreaActor::new(
+            family,
+            "realm1".to_string(),
+            "area1".to_string(),
+            store,
+            Arc::new(crate::domains::stream::metrics::StreamDurableMetrics::default()),
+        );
         let ctx = Context::new(addr, router);
         (actor, ctx)
     }
@@ -325,7 +341,13 @@ mod tests {
             .expect("Failed to open store"),
         );
         let store = Arc::new(StreamStore::new(db));
-        let actor = AreaActor::new(family, "realm1".to_string(), "area1".to_string(), store);
+        let actor = AreaActor::new(
+            family,
+            "realm1".to_string(),
+            "area1".to_string(),
+            store,
+            Arc::new(crate::domains::stream::metrics::StreamDurableMetrics::default()),
+        );
         let ctx = Context::new(addr, router);
         (actor, ctx, stream_mailbox)
     }
@@ -340,6 +362,10 @@ mod tests {
 
         // Assert
         assert_eq!(actor.watermark(), 3);
+        assert_eq!(
+            actor.durable_metrics.snapshot().area_watermarks[0].watermark,
+            3
+        );
     }
 
     #[test]