feat(location): sync a location only after runs end or on request

.gitattributes

@@ -0,0 +1,11 @@
+# GitHub's language bar counts shipped code only: test suites, development scripts and
+# benchmarks are documentation to Linguist, in every language.
+tests/** linguist-documentation
+scripts/** linguist-documentation
+benchmarks/** linguist-documentation
+crates/**/tests/** linguist-documentation
+crates/**/tests.rs linguist-documentation
+ui/src/**/*.test.ts linguist-documentation
+ui/scripts/** linguist-documentation
+clients/typescript/tests/** linguist-documentation
+clients/typescript/scripts/** linguist-documentation

clients/typescript/src/schema.d.ts

@@ -818,6 +818,22 @@ export interface paths {
         patch?: never;
         trace?: never;
     };
+    "/api/control/locations/{location_id}/targets/{target}/warehouse-syncs": {
+        parameters: {
+            query?: never;
+            header?: never;
+            path?: never;
+            cookie?: never;
+        };
+        get?: never;
+        put?: never;
+        post: operations["requestWarehouseSync"];
+        delete?: never;
+        options?: never;
+        head?: never;
+        patch?: never;
+        trace?: never;
+    };
     "/api/control/platform/events": {
         parameters: {
             query?: never;
@@ -1010,6 +1026,22 @@ export interface paths {
         patch?: never;
         trace?: never;
     };
+    "/api/locations/{location_id}/targets/{target}/metadata/sync-due": {
+        parameters: {
+            query?: never;
+            header?: never;
+            path?: never;
+            cookie?: never;
+        };
+        get?: never;
+        put?: never;
+        post: operations["sync_due"];
+        delete?: never;
+        options?: never;
+        head?: never;
+        patch?: never;
+        trace?: never;
+    };
     "/api/locations/{location_id}/targets/{target}/metadata/watermark": {
         parameters: {
             query?: never;
@@ -1524,11 +1556,6 @@ export interface components {
             rejected_sha256: string;
             table: components["schemas"]["Table"];
         };
-        ConnectionTestRequest: {
-            reason: string;
-            /** Format: uuid */
-            request_id: string;
-        };
         /** @description What the UI needs first: how people sign in, and what this person may see and do. */
         ControlBootstrap: {
             access: components["schemas"]["PersonAccess"];
@@ -1776,6 +1803,9 @@ export interface components {
         LocationWork: {
             /** @enum {string} */
             kind: "check_connection";
+        } | {
+            /** @enum {string} */
+            kind: "sync_warehouse";
         } | {
             evaluation: Record<string, never>;
             /** @enum {string} */
@@ -1810,6 +1840,9 @@ export interface components {
             document: Record<string, never>;
             /** @enum {string} */
             kind: "plan";
+        } | {
+            /** @enum {string} */
+            kind: "sync_started";
         } | {
             /** @enum {string} */
             kind: "superseded";
@@ -2193,6 +2226,12 @@ export interface components {
             /** @description A running schedule's next due times from `next_due_at` on, at most five; empty otherwise. */
             upcoming: string[];
         };
+        /** @description A person asks a location for one named read-only operation, such as a connection test. */
+        PersonWorkRequest: {
+            reason: string;
+            /** Format: uuid */
+            request_id: string;
+        };
         /** @description Observed physical evidence is not a live warehouse-existence guarantee. */
         PhysicalEvidence: {
             observation: components["schemas"]["CatalogObservation"];
@@ -3107,6 +3146,20 @@ export interface components {
             state_database?: string | null;
             state_schema: string;
         };
+        /** @description Both times come from the server clock, so the reader compares them without clock skew. */
+        WarehouseSyncDue: {
+            /** Format: date-time */
+            now: string;
+            /** Format: date-time */
+            runs_ended_at?: string | null;
+        };
+        /** @description A registered reader asks whether a run ended since its last synchronization began. */
+        WarehouseSyncDueRead: {
+            /** Format: int64 */
+            generation: number;
+            /** Format: uuid */
+            session_id: string;
+        };
         /**
          * @description Closed failure stages keep warehouse errors and credentials out of health observations.
          * @enum {string}
@@ -4104,7 +4157,7 @@ export interface operations {
         };
         requestBody: {
             content: {
-                "application/json": components["schemas"]["ConnectionTestRequest"];
+                "application/json": components["schemas"]["PersonWorkRequest"];
             };
         };
         responses: {
@@ -5308,6 +5361,57 @@ export interface operations {
             };
         };
     };
+    requestWarehouseSync: {
+        parameters: {
+            query?: never;
+            header: {
+                "x-sqb-csrf": string;
+            };
+            path: {
+                location_id: string;
+                target: string;
+            };
+            cookie?: never;
+        };
+        requestBody: {
+            content: {
+                "application/json": components["schemas"]["PersonWorkRequest"];
+            };
+        };
+        responses: {
+            200: {
+                headers: {
+                    [name: string]: unknown;
+                };
+                content: {
+                    "application/json": components["schemas"]["MutationResponse"];
+                };
+            };
+            400: {
+                headers: {
+                    [name: string]: unknown;
+                };
+                content: {
+                    "application/json": components["schemas"]["ApiErrorBody"];
+                };
+            };
+            /** @description Session rejected */
+            401: {
+                headers: {
+                    [name: string]: unknown;
+                };
+                content?: never;
+            };
+            403: {
+                headers: {
+                    [name: string]: unknown;
+                };
+                content: {
+                    "application/json": components["schemas"]["ApiErrorBody"];
+                };
+            };
+        };
+    };
     listPersonPlatformEvents: {
         parameters: {
             query?: {
@@ -5836,6 +5940,61 @@ export interface operations {
             };
         };
     };
+    sync_due: {
+        parameters: {
+            query?: never;
+            header?: never;
+            path: {
+                location_id: string;
+                target: string;
+            };
+            cookie?: never;
+        };
+        requestBody: {
+            content: {
+                "application/json": components["schemas"]["WarehouseSyncDueRead"];
+            };
+        };
+        responses: {
+            200: {
+                headers: {
+                    [name: string]: unknown;
+                };
+                content: {
+                    "application/json": components["schemas"]["WarehouseSyncDue"];
+                };
+            };
+            400: {
+                headers: {
+                    [name: string]: unknown;
+                };
+                content: {
+                    "application/json": components["schemas"]["ApiErrorBody"];
+                };
+            };
+            /** @description Invalid credential */
+            401: {
+                headers: {
+                    [name: string]: unknown;
+                };
+                content?: never;
+            };
+            /** @description Location not authorized */
+            403: {
+                headers: {
+                    [name: string]: unknown;
+                };
+                content?: never;
+            };
+            /** @description Reader generation fenced */
+            409: {
+                headers: {
+                    [name: string]: unknown;
+                };
+                content?: never;
+            };
+        };
+    };
     watermark: {
         parameters: {
             query?: never;

crates/sqlbuild-cli/src/argv/command_table.json

@@ -82,12 +82,12 @@
      "name": "control",
      "parser": {
       "prog": "sqb control",
-      "description": "Read one exact node's current published SQL using your saved sqb login session. Results are JSON; progress goes to stderr. No project or warehouse connection is opened.",
+      "description": "Use your saved sqb login session to read one exact node's current published SQL, or to ask a location to sync from the warehouse now. Results are JSON; progress goes to stderr. No project or warehouse connection is opened.",
       "groups": [{"title": "positional arguments"}, {"title": "options"}],
       "actions": [
        {"kind": "help", "options": ["-h", "--help"], "dest": "help", "nargs": 0, "default": {"suppress": true}, "help": "show this help message and exit", "group": 1},
-       {"kind": "store", "dest": "person_action", "choices": ["sql"], "help": "inspect published raw and compiled SQL", "group": 0},
-       {"kind": "store", "dest": "person_node", "help": "exact published node key, for example model:orders (not a selector)", "group": 0},
+       {"kind": "store", "dest": "person_action", "choices": ["sql", "sync"], "help": "sql: inspect published raw and compiled SQL; sync: ask the location to read SQLBuild's state and catalog from the warehouse now", "group": 0},
+       {"kind": "store", "dest": "person_node", "nargs": "?", "help": "for sql: exact published node key, for example model:orders (not a selector)", "group": 0},
        {"kind": "store", "options": ["--server"], "dest": "login_server", "required": true, "help": "HTTPS SQLBuild server origin used by sqb login", "group": 1},
        {"kind": "store", "options": ["--ca-cert"], "dest": "login_ca", "help": "additional trusted PEM CA certificate", "group": 1},
        {"kind": "store", "options": ["--location-id"], "dest": "person_location", "required": true, "help": "published location UUID", "group": 1},

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

@@ -37,7 +37,7 @@ use sqlbuild_orchestration::store::models::ControlStore;
 use sqlbuild_orchestration::warehouse_metadata::constants::MAX_HISTORY_RECORDS;
 use sqlbuild_orchestration::warehouse_metadata::models::{
     MetadataBatch, MetadataScan, StatisticsDueRead, StatisticsRead, StatisticsRetention,
-    WarehouseHistoryKind, WarehouseHistoryPage, WarehouseStateLocation,
+    WarehouseHistoryKind, WarehouseHistoryPage, WarehouseStateLocation, WarehouseSyncDueRead,
 };
 use sqlbuild_orchestration::webhooks::constants::SECURITY_AUDIT_WEBHOOK_VARIABLE;
 use sqlbuild_orchestration::webhooks::main::{
@@ -70,7 +70,8 @@ use crate::native_commands::_helpers::terminal::{
 };
 use crate::native_commands::constants::{
     ADMIN_COMMAND, COLUMNS_VARIABLE, FALLBACK_COLUMNS, METADATA_POLL_SECONDS,
-    METADATA_RECONCILE_SECONDS, SCHEDULER_COMMAND, UNCAUGHT_EXCEPTION_STATUS, UNKNOWN_COMMAND,
+    METADATA_RECONCILE_SECONDS, METADATA_SYNC_CHECK_SECONDS, SCHEDULER_COMMAND,
+    UNCAUGHT_EXCEPTION_STATUS, UNKNOWN_COMMAND,
 };
 use crate::native_commands::errors::PythonNeeded;
 use crate::native_commands::main::_run_admin::run_admin;
@@ -349,6 +350,8 @@ pub(crate) fn metadata_sync(
         statistics_limit: MAX_HISTORY_RECORDS,
         pending: None,
         next_at: Instant::now(),
+        idle: true,
+        cycle_started_at: None,
     }
 }
 
@@ -447,6 +450,34 @@ pub(crate) fn freeze_history_page(
     Ok(state)
 }
 
+/// An idle location asks the server, never the warehouse, whether a run ended since it last began.
+pub(crate) async fn begin_after_run(
+    mut state: LocationMetadataSync,
+    (client, poll): (&LocationWorkClient, &LocationWorkPoll),
+    target: &str,
+) -> (LocationMetadataSync, Result<bool, ClientError>) {
+    state.next_at = Instant::now() + Duration::from_secs(METADATA_SYNC_CHECK_SECONDS);
+    let request = WarehouseSyncDueRead {
+        session_id: poll.session_id,
+        generation: state.generation,
+    };
+    let due = match client.metadata_sync_due(target, &request).await {
+        Ok(due) => due,
+        Err(error) => return (state, Err(error)),
+    };
+    let ran = match (state.cycle_started_at, due.runs_ended_at) {
+        (None, _) => true,
+        (Some(started_at), Some(ended_at)) => ended_at >= started_at,
+        (Some(_), None) => false,
+    };
+    if ran {
+        state.idle = false;
+        state.cycle_started_at = Some(due.now);
+        state.next_at = Instant::now();
+    }
+    (state, Ok(ran))
+}
+
 pub(crate) fn acknowledge_metadata_page(mut state: LocationMetadataSync) -> LocationMetadataSync {
     let Some(pending) = state.pending.take() else {
         return state;
@@ -470,7 +501,8 @@ pub(crate) fn acknowledge_metadata_page(mut state: LocationMetadataSync) -> Loca
                 state.reconcile_at =
                     Instant::now() + Duration::from_secs(u64::from(state.reconcile_seconds));
             }
-            state.next_at = Instant::now() + Duration::from_secs(METADATA_POLL_SECONDS);
+            state.idle = true;
+            state.next_at = Instant::now();
             drop(write_output(&[(
                 Stream::Stderr,
                 "Warehouse history synchronization completed.\n",

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

@@ -93,8 +93,10 @@ pub(crate) const UNCAUGHT_EXCEPTION_STATUS: i32 = 1;
 /// Reserve half the location work claim for acknowledgement retries after killing a callback.
 pub(crate) const SENSOR_EVALUATION_SECONDS: u64 = 30;
 pub(crate) const SENSOR_RESULT_BYTES: u64 = 1_048_576;
-/// Independent read timeout and delay between automatic history polls.
+/// Read timeout, and the delay before retrying a failed page of an unfinished synchronization.
 pub(crate) const METADATA_POLL_SECONDS: u64 = 30;
+/// How often an idle location asks the server, never the warehouse, whether a run ended.
+pub(crate) const METADATA_SYNC_CHECK_SECONDS: u64 = 5;
 /// Full key-set reconciliation is independent of the incremental poll and overlap window.
 pub(crate) const METADATA_RECONCILE_SECONDS: u32 = 300;
 pub(crate) const METADATA_OVERLAP_SECONDS: i64 = 300;

crates/sqlbuild-cli/src/native_commands/main/_location_metadata.rs

@@ -16,8 +16,8 @@ use tokio::time::Instant;
 
 use crate::argv::models::Stream;
 use crate::native_commands::_helpers::execution::{
-    acknowledge_metadata_page, freeze_history_page, history_request, location_result,
-    metadata_health, statistics_page,
+    acknowledge_metadata_page, begin_after_run, freeze_history_page, history_request,
+    location_result, metadata_health, statistics_page,
 };
 use crate::native_commands::_helpers::paths::location_inputs;
 use crate::native_commands::_helpers::terminal::write_output;
@@ -41,6 +41,18 @@ pub(super) async fn sync_location_metadata(
     }
     let started = Instant::now();
     tokio::pin!(stopped);
+    if state.idle {
+        let (next, due) = tokio::select! {
+            result = &mut stopped => { result?; return Ok(None); },
+            result = begin_after_run(state, (client, poll), &options.target) => result,
+        };
+        state = next;
+        match due {
+            Ok(true) => {}
+            Ok(false) => return Ok(Some(state)),
+            Err(error) => return retry_metadata(state, error),
+        }
+    }
     if state.pending.is_none() {
         if state.kind() == WarehouseHistoryKind::Statistics && state.statistics.is_none() {
             state.statistics = Some(tokio::select! {