feat(location): sync every state table without waking a suspended warehouse

benchmarks/duckshop/edge_cases/orchestration/column_migration_history.json

@@ -0,0 +1,46 @@
+{
+  "record": {
+    "state_database": null,
+    "state_schema": "main",
+    "event_id": "rename-000",
+    "target_name": null,
+    "model_name": "orders",
+    "relation_database": null,
+    "relation_schema": "main",
+    "relation_name": "orders_🧾",
+    "origin_column": "customer_🧾",
+    "destination_column": "customer_id",
+    "discovery": "manual",
+    "decision": "rename",
+    "run_id": "external-rename",
+    "created_at": "2026-02-03T04:05:06Z"
+  },
+  "source_rows": 104,
+  "duplicate_rows": 3,
+  "later_time": "2026-02-03T04:05:06.000001Z",
+  "page_lengths": [
+    100,
+    4
+  ],
+  "later_index": 103,
+  "source_count": 104,
+  "model": "MODEL (description 'Orders.', materialized table); SELECT 1 AS order_id",
+  "retained_result": "rename-103",
+  "kind": "column_migrations",
+  "source_key": "event_id",
+  "timestamp_key": "created_at",
+  "id_prefix": "rename",
+  "later_record": {
+    "event_id": "rename-103",
+    "created_at": "2026-02-03T04:05:06.000001Z"
+  },
+  "last_record": {
+    "event_id": "rename-103",
+    "created_at": "2026-02-03T04:05:06.000001Z"
+  },
+  "record_overrides": {
+    "rename-103": {
+      "created_at": "2026-02-03T04:05:06.000001Z"
+    }
+  }
+}

benchmarks/duckshop/edge_cases/orchestration/consumption_history.json

@@ -0,0 +1,63 @@
+{
+  "record": {
+    "state_database": null,
+    "state_schema": "main",
+    "execution_id": "execution-000",
+    "phase": "succeeded",
+    "run_id": "external-run",
+    "node_type": "model",
+    "node_name": "orders",
+    "target_database": null,
+    "target_schema": "main",
+    "target_name": "orders_🧾",
+    "version_hash": "orders-version",
+    "input_count": 1,
+    "input_kind": "source",
+    "input_name": "shop.orders_🧾",
+    "input_version": "source-version",
+    "recorded_at": "2026-02-03T04:05:06Z"
+  },
+  "source_rows": 104,
+  "duplicate_rows": 3,
+  "later_time": "2026-02-03T04:05:06.000001Z",
+  "page_lengths": [
+    100,
+    7
+  ],
+  "later_index": 106,
+  "source_count": 107,
+  "model": "MODEL (description 'Orders.', materialized table); SELECT 1 AS order_id",
+  "retained_result": "execution-103",
+  "kind": "consumption",
+  "source_key": "execution_id",
+  "timestamp_key": "recorded_at",
+  "id_prefix": "execution",
+  "later_record": {
+    "execution_id": "execution-103",
+    "phase": "started",
+    "input_count": 0,
+    "input_kind": null,
+    "input_name": null,
+    "input_version": null,
+    "recorded_at": "2026-02-03T04:05:06.000001Z"
+  },
+  "last_record": {
+    "execution_id": "execution-103",
+    "phase": "started",
+    "input_count": 0,
+    "input_kind": null,
+    "input_name": null,
+    "input_version": null,
+    "recorded_at": "2026-02-03T04:05:06.000001Z"
+  },
+  "record_overrides": {
+    "execution-103": {
+      "phase": "started",
+      "input_count": 0,
+      "input_kind": null,
+      "input_name": null,
+      "input_version": null,
+      "recorded_at": "2026-02-03T04:05:06.000001Z"
+    }
+  }
+}

benchmarks/duckshop/edge_cases/orchestration/node_result_history.json

@@ -0,0 +1,52 @@
+{
+  "record": {
+    "state_database": null,
+    "state_schema": "main",
+    "node_type": "task",
+    "node_name": "result-000",
+    "target_database": null,
+    "target_schema": "main",
+    "target_name": null,
+    "run_id": "external-run",
+    "status": "success",
+    "payload_json_b64": "eyJub3RlIjoiUmV0YWluZWQgJ29yZGVycycgXHVkODNlXHVkZGZlIiwicm93cyI6N30=",
+    "metadata_json_b64": "e30=",
+    "error_message": null,
+    "materialized": "true",
+    "ts": "2026-02-03T04:05:06Z"
+  },
+  "source_rows": 104,
+  "duplicate_rows": 3,
+  "later_time": "2026-02-03T04:05:06.000001Z",
+  "page_lengths": [
+    100,
+    7
+  ],
+  "later_index": 106,
+  "source_count": 107,
+  "model": "MODEL (description 'Orders.', materialized table); SELECT 1 AS order_id",
+  "retained_result": "result-103",
+  "kind": "node_results",
+  "source_key": "node_name",
+  "timestamp_key": "ts",
+  "id_prefix": "result",
+  "later_record": {
+    "node_name": "result-103",
+    "status": "failed",
+    "error_message": "Rejected 'orders' 🧾",
+    "ts": "2026-02-03T04:05:06.000001Z"
+  },
+  "last_record": {
+    "node_name": "result-103",
+    "status": "failed",
+    "error_message": "Rejected 'orders' 🧾",
+    "ts": "2026-02-03T04:05:06.000001Z"
+  },
+  "record_overrides": {
+    "result-103": {
+      "status": "failed",
+      "error_message": "Rejected 'orders' 🧾",
+      "ts": "2026-02-03T04:05:06.000001Z"
+    }
+  }
+}

clients/typescript/src/schema.d.ts

@@ -3039,7 +3039,7 @@ export interface components {
          * @description Catalog observations require full scans; only immutable history supports a watermark.
          * @enum {string}
          */
-        WarehouseHistoryKind: "fingerprints" | "source_freshness" | "audit_results" | "janitor_events" | "migrations" | "old_name_views" | "microbatches" | "catalog" | "statistics";
+        WarehouseHistoryKind: "fingerprints" | "source_freshness" | "audit_results" | "janitor_events" | "migrations" | "old_name_views" | "microbatches" | "column_migrations" | "node_results" | "consumption" | "catalog" | "statistics";
         /** @description Warehouse cleanup facts do not authorize control-plane deletion or worker recovery. */
         WarehouseJanitorRecord: {
             archive_name: string;

crates/sqlbuild-analysis/src/janitor_planning/constants.rs

@@ -1,25 +1,9 @@
 //! Janitor planning names, prefixes and skip reasons.
 
-use sqlbuild_core::constants::COLUMN_MIGRATION_TABLE_NAME;
-use sqlbuild_core::constants::CONSUMPTION_TABLE_NAME;
-use sqlbuild_core::constants::FINGERPRINT_TABLE_NAME;
-use sqlbuild_core::constants::MIGRATION_TABLE_NAME;
-use sqlbuild_core::constants::OLD_NAME_VIEW_TABLE_NAME;
-use sqlbuild_core::constants::SOURCE_FRESHNESS_TABLE_NAME;
+use sqlbuild_core::constants::STATE_TABLE_NAMES;
 
 /// SQLBuild state tables the janitor never acts on.
-pub const BUILT_IN_EXCLUDE_PATTERNS: [&str; 10] = [
-    FINGERPRINT_TABLE_NAME,
-    "_sqlbuild_microbatches",
-    SOURCE_FRESHNESS_TABLE_NAME,
-    CONSUMPTION_TABLE_NAME,
-    "_sqlbuild_node_results",
-    "_sqlbuild_audit_results",
-    "_sqlbuild_janitor_events",
-    MIGRATION_TABLE_NAME,
-    COLUMN_MIGRATION_TABLE_NAME,
-    OLD_NAME_VIEW_TABLE_NAME,
-];
+pub const BUILT_IN_EXCLUDE_PATTERNS: [&str; 10] = STATE_TABLE_NAMES;
 /// Prefix of every strict janitor archive name.
 pub const ARCHIVE_NAME_PREFIX: &str = "_SQB_ARCHIVE__";
 /// Separator between archive name segments.

crates/sqlbuild-cli/src/native_commands/_helpers/execution.rs

@@ -346,6 +346,7 @@ pub(crate) fn metadata_sync(
         reconcile_at: Instant::now(),
         reconcile_seconds,
         limit: MAX_HISTORY_RECORDS,
+        statistics_limit: MAX_HISTORY_RECORDS,
         pending: None,
         next_at: Instant::now(),
     }
@@ -389,7 +390,7 @@ pub(crate) async fn statistics_page(
                 generation: state.generation,
                 location: state.location().clone(),
                 after,
-                limit: state.limit,
+                limit: state.statistics_limit,
             },
         )
         .await
@@ -423,8 +424,13 @@ pub(crate) fn freeze_history_page(
         > BODY_BYTES
     {
         state.statistics = None;
-        if state.limit > 1 {
-            state.limit = (state.limit / 2).max(1);
+        let limit = if state.kind() == WarehouseHistoryKind::Statistics {
+            &mut state.statistics_limit
+        } else {
+            &mut state.limit
+        };
+        if *limit > 1 {
+            *limit /= 2;
         } else {
             state.next_at = Instant::now() + Duration::from_secs(METADATA_POLL_SECONDS);
             drop(write_output(&[(
@@ -450,6 +456,9 @@ pub(crate) fn acknowledge_metadata_page(mut state: LocationMetadataSync) -> Loca
         state.statistics = None;
     }
     if state.source_after.is_none() {
+        if state.kind() == WarehouseHistoryKind::Statistics {
+            state.statistics_limit = MAX_HISTORY_RECORDS;
+        }
         state.index += 1;
         state.scan_id = Uuid::now_v7();
         state.observed_since = None;

crates/sqlbuild-cli/src/native_commands/models.rs

@@ -200,21 +200,25 @@ pub(crate) struct LocationMetadataSync {
     pub(crate) reconcile_at: tokio::time::Instant,
     pub(crate) reconcile_seconds: u32,
     pub(crate) limit: usize,
+    /// Survives a failed statistics read's rewind through catalog, which resets `limit`.
+    pub(crate) statistics_limit: usize,
     pub(crate) pending: Option<LocationMetadataPage>,
     pub(crate) next_at: tokio::time::Instant,
 }
 
 impl LocationMetadataSync {
     /// Failed samples revisit catalog before retrying; no frozen delivery is discarded.
     pub(crate) fn retry_read(&mut self) {
-        self.limit = (self.limit / 2).max(1);
         self.statistics = None;
         if self.kind() == WarehouseHistoryKind::Statistics {
+            self.statistics_limit = (self.statistics_limit / 2).max(1);
             self.index -= 1;
             self.source_after = None;
             self.scan_id = Uuid::now_v7();
             self.observed_since = None;
             self.watermark_loaded = false;
+        } else {
+            self.limit = (self.limit / 2).max(1);
         }
         self.next_at = tokio::time::Instant::now()
             + std::time::Duration::from_secs(super::constants::METADATA_POLL_SECONDS);

crates/sqlbuild-cli/src/native_commands/tests.rs

@@ -6,6 +6,8 @@ mod decision;
 mod helpers;
 #[path = "tests/test_init.rs"]
 mod init;
+#[path = "tests/test_location_metadata.rs"]
+mod location_metadata;
 #[path = "tests/test_location_plan.rs"]
 mod location_plan;
 #[path = "tests/test_paths.rs"]

crates/sqlbuild-cli/src/native_commands/tests/helpers.rs

@@ -1,8 +1,13 @@
 use std::path::Path;
 
+use sqlbuild_orchestration::warehouse_metadata::models::MetadataBatch;
+use uuid::Uuid;
+
 use crate::argv::main::parse_cli_arguments::parse_cli_arguments;
 use crate::argv::models::{ArgvOutcome, ArgvSettings};
 use crate::native_commands::_helpers::argv_host::{NativeArgvHost, namespace_defaults};
+use crate::native_commands::_helpers::execution::acknowledge_metadata_page;
+use crate::native_commands::models::{LocationMetadataPage, LocationMetadataSync};
 
 /// Parse `argv` onto the `CliNamespace` defaults with the native host.
 pub(super) fn parse(argv: &[&str]) -> (ArgvOutcome, NativeArgvHost) {
@@ -35,3 +40,23 @@ pub(super) fn plan_output(diff_line: &str, count: usize) -> Vec<u8> {
     serde_json::to_vec(&serde_json::json!({"selected_count": count, "models": models}))
         .expect("plan output")
 }
+
+/// Acknowledges `pages` complete, empty pages, each finishing the kind it was read for.
+pub(super) fn acknowledge_complete_pages(
+    state: LocationMetadataSync,
+    pages: usize,
+) -> LocationMetadataSync {
+    (0..pages).fold(state, |mut state, _| {
+        state.pending = Some(LocationMetadataPage {
+            batch: MetadataBatch {
+                kind: state.kind(),
+                session_id: Uuid::nil(),
+                generation: 1,
+                records: Vec::new(),
+                scan: None,
+            },
+            next_source_after: None,
+        });
+        acknowledge_metadata_page(state)
+    })
+}

crates/sqlbuild-cli/src/native_commands/tests/test_location_metadata.rs

@@ -0,0 +1,55 @@
+use sqlbuild_orchestration::warehouse_metadata::constants::MAX_HISTORY_RECORDS;
+use sqlbuild_orchestration::warehouse_metadata::models::{
+    WarehouseHistoryKind, WarehouseStateLocation,
+};
+
+use crate::native_commands::_helpers::execution::metadata_sync;
+use crate::native_commands::tests::helpers::acknowledge_complete_pages;
+use crate::native_commands::tests::test_types::MetadataRetryTestCase;
+
+#[test]
+fn given_failed_metadata_read_when_back_at_statistics_then_page_shrinks_until_scan_completes() {
+    let test_cases = [
+        MetadataRetryTestCase {
+            description: "a timed-out statistics page is retried smaller after catalog is revisited",
+            failed_kind: WarehouseHistoryKind::Statistics,
+            acknowledged_pages: 1,
+            expected_statistics_limit: MAX_HISTORY_RECORDS / 2,
+        },
+        MetadataRetryTestCase {
+            description: "a failed catalog page leaves the statistics page size alone",
+            failed_kind: WarehouseHistoryKind::Catalog,
+            acknowledged_pages: 1,
+            expected_statistics_limit: MAX_HISTORY_RECORDS,
+        },
+        MetadataRetryTestCase {
+            description: "a completed statistics scan restores the full statistics page",
+            failed_kind: WarehouseHistoryKind::Statistics,
+            acknowledged_pages: 1 + WarehouseHistoryKind::ALL.len(),
+            expected_statistics_limit: MAX_HISTORY_RECORDS,
+        },
+    ];
+    for test_case in test_cases {
+        let location = WarehouseStateLocation {
+            database: Some("analytics".into()),
+            schema: "state".into(),
+        };
+        let mut state = metadata_sync((vec![location], 1), 3600);
+        state.index = WarehouseHistoryKind::ALL
+            .iter()
+            .position(|kind| *kind == test_case.failed_kind)
+            .expect("kind is listed");
+        state.retry_read();
+        let state = acknowledge_complete_pages(state, test_case.acknowledged_pages);
+        assert_eq!(
+            (state.kind(), state.statistics_limit, state.limit),
+            (
+                WarehouseHistoryKind::Statistics,
+                test_case.expected_statistics_limit,
+                MAX_HISTORY_RECORDS,
+            ),
+            "{description}",
+            description = test_case.description
+        );
+    }
+}

crates/sqlbuild-cli/src/native_commands/tests/test_types.rs

@@ -1,3 +1,5 @@
+use sqlbuild_orchestration::warehouse_metadata::models::WarehouseHistoryKind;
+
 use crate::argv::models::{TypeFailure, Value};
 use crate::native_commands::errors::PythonNeeded;
 use crate::native_commands::models::{CommandOutput, NativeCommand, RulesAction};
@@ -89,3 +91,11 @@ pub(super) struct PlanDocumentTestCase {
     pub(super) planned: bool,
     pub(super) expected_first_model: serde_json::Value,
 }
+
+/// One failed metadata read and the statistics page size the sync uses afterwards.
+pub(super) struct MetadataRetryTestCase {
+    pub(super) description: &'static str,
+    pub(super) failed_kind: WarehouseHistoryKind,
+    pub(super) acknowledged_pages: usize,
+    pub(super) expected_statistics_limit: usize,
+}