diff --git a/.gitattributes b/.gitattributes new file mode 100644 index 0000000000..3dcbe083f9 --- /dev/null +++ b/.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 diff --git a/clients/typescript/src/schema.d.ts b/clients/typescript/src/schema.d.ts index 8a7d06b209..041ea0880a 100644 --- a/clients/typescript/src/schema.d.ts +++ b/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; /** @enum {string} */ @@ -1810,6 +1840,9 @@ export interface components { document: Record; /** @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; diff --git a/crates/sqlbuild-cli/src/argv/command_table.json b/crates/sqlbuild-cli/src/argv/command_table.json index c433d00395..bc15cbc49d 100644 --- a/crates/sqlbuild-cli/src/argv/command_table.json +++ b/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}, diff --git a/crates/sqlbuild-cli/src/native_commands/_helpers/execution.rs b/crates/sqlbuild-cli/src/native_commands/_helpers/execution.rs index 9d5ce63b7e..a7859224da 100644 --- a/crates/sqlbuild-cli/src/native_commands/_helpers/execution.rs +++ b/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) { + 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", diff --git a/crates/sqlbuild-cli/src/native_commands/constants.rs b/crates/sqlbuild-cli/src/native_commands/constants.rs index 6406a3a58b..ab853320e3 100644 --- a/crates/sqlbuild-cli/src/native_commands/constants.rs +++ b/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; diff --git a/crates/sqlbuild-cli/src/native_commands/main/_location_metadata.rs b/crates/sqlbuild-cli/src/native_commands/main/_location_metadata.rs index 7ad9339b76..90bbbf4164 100644 --- a/crates/sqlbuild-cli/src/native_commands/main/_location_metadata.rs +++ b/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! { diff --git a/crates/sqlbuild-cli/src/native_commands/main/_location_work.rs b/crates/sqlbuild-cli/src/native_commands/main/_location_work.rs index d67613c7a2..41b65faff0 100644 --- a/crates/sqlbuild-cli/src/native_commands/main/_location_work.rs +++ b/crates/sqlbuild-cli/src/native_commands/main/_location_work.rs @@ -20,13 +20,13 @@ use crate::native_commands::main::_location_plan::plan_location; use crate::native_commands::main::_location_sensor::evaluate_location_sensor; use crate::native_commands::models::{LocationServe, LocationWatch}; -/// Retry the exact completed result after uncertain delivery; never rerun it to obtain a new result. +/// Retry the exact completed result without rerunning it; `true` asks the metadata loop to sync. pub(super) async fn run_location_work( (options, script): (&LocationServe, &Path), (client, poll, watch): (&LocationWorkClient, &LocationWorkPoll, &LocationWatch), generation: Option, stopped: impl Future>, -) -> Result>, ControlError> { +) -> Result, bool)>, ControlError> { tokio::pin!(stopped); let claim = tokio::select! { result = &mut stopped => { result?; return Ok(None); }, @@ -34,12 +34,12 @@ pub(super) async fn run_location_work( Ok(claim) => claim, Err(error) => { work_error(&error)?; - return Ok(Some(Vec::new())); + return Ok(Some((Vec::new(), false))); }, }, }; let Some(claim) = claim else { - return Ok(Some(Vec::new())); + return Ok(Some((Vec::new(), false))); }; let deadline = Instant::now() + (claim.lease_expires_at - Utc::now()) @@ -49,12 +49,6 @@ pub(super) async fn run_location_work( let outcome = if generation == Some(claim.generation) && location_inputs(&options.project_dir, &watch.rules, &watch.local_files) == watch.observed { - let message = match &claim.request.work { - LocationWork::CheckConnection {} => "Checking requested location connection...\n", - LocationWork::EvaluateSensor { .. } => "Evaluating requested custom sensor...\n", - LocationWork::Plan { .. } => "Planning requested command...\n", - }; - drop(write_output(&[(Stream::Stderr, message)])); let mut shutting_down = false; let expires = async { tokio::select! { @@ -62,34 +56,13 @@ pub(super) async fn run_location_work( () = tokio::time::sleep_until(deadline) => Ok(()), } }; - let checked = match claim.request.work { - LocationWork::CheckConnection {} => check_location_connection(options, script, expires) - .await? - .map(|(healthy, files)| (LocationWorkOutcome::Connection { healthy }, files)), - LocationWork::EvaluateSensor { evaluation } => { - evaluate_location_sensor(options, script, &evaluation, expires) - .await? - .map(|outcome| { - ( - LocationWorkOutcome::SensorEvaluation { outcome }, - Vec::new(), - ) - }) - } - LocationWork::Plan { - selection, - settings, - .. - } => plan_location((options, script), (&selection, &settings), expires) - .await? - .map(|document| (LocationWorkOutcome::Plan { document }, Vec::new())), - }; + let checked = perform_work((options, script), claim.request.work, expires).await?; let Some((outcome, local_files)) = checked else { drop(write_output(&[( Stream::Stderr, "Location work stopped before completion; no result reported.\n", )])); - return Ok((!shutting_down).then_some(files)); + return Ok((!shutting_down).then_some((files, false))); }; files = local_files; let mut expected = watch.observed.clone(); @@ -106,6 +79,7 @@ pub(super) async fn run_location_work( } else { LocationWorkOutcome::Superseded {} }; + let sync = matches!(outcome, LocationWorkOutcome::SyncStarted {}); let result = LocationWorkResult { claim_id: claim.claim_id, session_id: poll.session_id, @@ -123,7 +97,7 @@ pub(super) async fn run_location_work( Stream::Stderr, "Location work completed.\n", )])); - return Ok(Some(files)); + return Ok(Some((files, sync))); } Err(error) => { work_error(&error)?; @@ -141,7 +115,45 @@ pub(super) async fn run_location_work( Stream::Stderr, "Location work acknowledgement expired; check the server result before retrying.\n", )])); - Ok(Some(files)) + Ok(Some((files, sync))) +} + +/// One named operation; `None` means it stopped at shutdown or its lease ended. +async fn perform_work( + (options, script): (&LocationServe, &Path), + work: LocationWork, + expires: impl Future>, +) -> Result)>, ControlError> { + let message = match &work { + LocationWork::CheckConnection {} => "Checking requested location connection...\n", + LocationWork::SyncWarehouse {} => "Synchronizing warehouse history on request...\n", + LocationWork::EvaluateSensor { .. } => "Evaluating requested custom sensor...\n", + LocationWork::Plan { .. } => "Planning requested command...\n", + }; + drop(write_output(&[(Stream::Stderr, message)])); + Ok(match work { + LocationWork::SyncWarehouse {} => Some((LocationWorkOutcome::SyncStarted {}, Vec::new())), + LocationWork::CheckConnection {} => check_location_connection(options, script, expires) + .await? + .map(|(healthy, files)| (LocationWorkOutcome::Connection { healthy }, files)), + LocationWork::EvaluateSensor { evaluation } => { + evaluate_location_sensor(options, script, &evaluation, expires) + .await? + .map(|outcome| { + ( + LocationWorkOutcome::SensorEvaluation { outcome }, + Vec::new(), + ) + }) + } + LocationWork::Plan { + selection, + settings, + .. + } => plan_location((options, script), (&selection, &settings), expires) + .await? + .map(|document| (LocationWorkOutcome::Plan { document }, Vec::new())), + }) } fn work_error(error: &ClientError) -> Result<(), ControlError> { diff --git a/crates/sqlbuild-cli/src/native_commands/main/_run_location.rs b/crates/sqlbuild-cli/src/native_commands/main/_run_location.rs index 43e3a3f2c0..9968de6124 100644 --- a/crates/sqlbuild-cli/src/native_commands/main/_run_location.rs +++ b/crates/sqlbuild-cli/src/native_commands/main/_run_location.rs @@ -155,7 +155,7 @@ async fn poll_location( tokio::pin!(stopped); if generation.is_some() && watch.attempted.as_ref() == Some(&watch.observed) { if Instant::now() >= work_at { - let Some(files) = run_location_work( + let Some((files, sync)) = run_location_work( (options, script), (client, poll, &watch), generation, @@ -167,6 +167,9 @@ async fn poll_location( }; watch = exclude_location_databases(&options.project_dir, files, watch); work_at = Instant::now() + Duration::from_secs(2); + if sync && let Some(state) = metadata.as_mut() { + state.request_sync(); + } } if let Some(state) = metadata.take() { metadata = sync_location_metadata( diff --git a/crates/sqlbuild-cli/src/native_commands/main/_run_warehouse_sync.rs b/crates/sqlbuild-cli/src/native_commands/main/_run_warehouse_sync.rs new file mode 100644 index 0000000000..41a8a2a011 --- /dev/null +++ b/crates/sqlbuild-cli/src/native_commands/main/_run_warehouse_sync.rs @@ -0,0 +1,46 @@ +//! `sqb control sync` asks the location to read the warehouse; the location runs the sync itself. + +use crate::argv::models::{Namespace, Stream}; +use crate::native_commands::_helpers::decision::text; +use crate::native_commands::_helpers::terminal::write_output; +use crate::person_login::main::request_warehouse_sync::request_warehouse_sync; + +pub(super) fn run_warehouse_sync(namespace: &Namespace) -> i32 { + drop(write_output(&[( + Stream::Stderr, + "Requesting a warehouse sync…\n", + )])); + let result = text(namespace, "person_location") + .unwrap_or_default() + .parse() + .map_err(|_| std::io::Error::other("--location-id must be a UUID")) + .and_then(|location| { + let ca = text(namespace, "login_ca").map(std::path::PathBuf::from); + request_warehouse_sync( + ( + &text(namespace, "login_server").unwrap_or_default(), + ca.as_deref(), + ), + (location, &text(namespace, "target").unwrap_or_default()), + ) + }); + match result { + Ok(applied) => { + drop(write_output(&[ + (Stream::Stdout, &format!("{{\"applied\": {applied}}}\n")), + ( + Stream::Stderr, + "Warehouse sync requested; the location starts it within a few seconds.\n", + ), + ])); + 0 + } + Err(error) => { + drop(write_output(&[( + Stream::Stderr, + &format!("Warehouse sync request failed: {error}\n"), + )])); + 1 + } + } +} diff --git a/crates/sqlbuild-cli/src/native_commands/main/mod.rs b/crates/sqlbuild-cli/src/native_commands/main/mod.rs index 813d58cd45..c5ee3267d2 100644 --- a/crates/sqlbuild-cli/src/native_commands/main/mod.rs +++ b/crates/sqlbuild-cli/src/native_commands/main/mod.rs @@ -10,6 +10,7 @@ pub(super) mod _run_control; pub(super) mod _run_cost; pub(super) mod _run_dev; pub(super) mod _run_location; +pub(super) mod _run_warehouse_sync; pub mod effective_target; pub(crate) mod parse_native_invocation; pub(crate) mod run_control_invocation; diff --git a/crates/sqlbuild-cli/src/native_commands/main/run_person_invocation.rs b/crates/sqlbuild-cli/src/native_commands/main/run_person_invocation.rs index 036e68a8d5..558af309fd 100644 --- a/crates/sqlbuild-cli/src/native_commands/main/run_person_invocation.rs +++ b/crates/sqlbuild-cli/src/native_commands/main/run_person_invocation.rs @@ -3,12 +3,29 @@ use crate::argv::models::{Namespace, Stream}; use crate::native_commands::_helpers::decision::text; use crate::native_commands::_helpers::terminal::write_output; +use crate::native_commands::main::_run_warehouse_sync::run_warehouse_sync; use crate::person_login::main::read_node_sql::read_node_sql; pub(crate) fn run_person_invocation(namespace: &Namespace) -> Option { if text(namespace, "command").as_deref() != Some("control") { return None; } + let action = text(namespace, "person_action"); + let misuse = match (action.as_deref(), text(namespace, "person_node").is_some()) { + (Some("sync"), true) => Some("sync takes no node; it syncs the whole target"), + (Some("sql"), false) => Some("sql needs an exact node key, for example model:orders"), + _ => None, + }; + if let Some(message) = misuse { + drop(write_output(&[( + Stream::Stderr, + &format!("sqb control: error: {message}\n"), + )])); + return Some(2); + } + if action.as_deref() == Some("sync") { + return Some(run_warehouse_sync(namespace)); + } drop(write_output(&[( Stream::Stderr, "Reading published SQL…\n", diff --git a/crates/sqlbuild-cli/src/native_commands/models.rs b/crates/sqlbuild-cli/src/native_commands/models.rs index a1bbf08388..99f7eb26df 100644 --- a/crates/sqlbuild-cli/src/native_commands/models.rs +++ b/crates/sqlbuild-cli/src/native_commands/models.rs @@ -204,9 +204,21 @@ pub(crate) struct LocationMetadataSync { pub(crate) statistics_limit: usize, pub(crate) pending: Option, pub(crate) next_at: tokio::time::Instant, + /// Between synchronizations; the warehouse is read again only after a run ends. + pub(crate) idle: bool, + /// Server time when the current or last synchronization began; `None` before the first. + pub(crate) cycle_started_at: Option>, } impl LocationMetadataSync { + /// A person asked for a sync: start one now when idle, or one more after the current one. + pub(crate) fn request_sync(&mut self) { + self.cycle_started_at = None; + if self.idle { + self.next_at = tokio::time::Instant::now(); + } + } + /// Failed samples revisit catalog before retrying; no frozen delivery is discarded. pub(crate) fn retry_read(&mut self) { self.statistics = None; diff --git a/crates/sqlbuild-cli/src/person_login/_helpers/transport.rs b/crates/sqlbuild-cli/src/person_login/_helpers/transport.rs index f44ef2f0ce..7703fe34ff 100644 --- a/crates/sqlbuild-cli/src/person_login/_helpers/transport.rs +++ b/crates/sqlbuild-cli/src/person_login/_helpers/transport.rs @@ -4,11 +4,14 @@ use std::path::Path; use std::time::Duration; use reqwest::header::{AUTHORIZATION, HeaderValue}; -use sqlbuild_orchestration::api::models::ApiErrorBody; +use sqlbuild_orchestration::api::main::parse_origin::parse_origin; +use sqlbuild_orchestration::api::models::{ApiErrorBody, MutationResponse}; use sqlbuild_orchestration::credentials::models::SessionIdentity; +use sqlbuild_orchestration::platform::models::PersonWorkRequest; use sqlbuild_orchestration::snapshots::models::NodeSql; use uuid::Uuid; +use crate::person_login::_helpers::credentials; use crate::person_login::constants::REQUEST_SECONDS; use crate::person_login::models::{PersonClient, PersonCredential}; @@ -38,6 +41,21 @@ pub(crate) fn connect(origin: &str, ca: Option<&Path>) -> Result( + (server, ca): (&str, Option<&Path>), + request: impl AsyncFnOnce(&PersonClient, &PersonCredential) -> Result, +) -> Result { + let origin = parse_origin(server, "CLI").map_err(std::io::Error::other)?; + let credential = credentials::load(&origin)? + .ok_or_else(|| std::io::Error::other("not signed in; run sqb login"))?; + let client = connect(&origin, ca)?; + tokio::runtime::Builder::new_multi_thread() + .enable_all() + .build()? + .block_on(request(&client, &credential)) +} + pub(crate) fn authorized( client: &PersonClient, method: reqwest::Method, @@ -104,6 +122,65 @@ pub(crate) async fn node_sql( credential: &PersonCredential, (location, target, node): (Uuid, &str, &str), ) -> Result, std::io::Error> { + let path = target_path(client, (location, target), &["nodes", node, "sql"])?; + send_json( + authorized(client, reqwest::Method::GET, &path, credential)?, + ( + "read published SQL from", + "SQL inspection", + "SQL inspection result", + ), + ) + .await +} + +/// Whether the server queued the request; `false` means this request ID was already queued. +pub(crate) async fn warehouse_sync( + client: &PersonClient, + credential: &PersonCredential, + (location, target): (Uuid, &str), +) -> Result { + let path = target_path(client, (location, target), &["warehouse-syncs"])?; + let request = PersonWorkRequest { + request_id: Uuid::now_v7(), + reason: "Requested from the CLI".into(), + }; + send_json::( + authorized(client, reqwest::Method::POST, &path, credential)?.json(&request), + ( + "request a warehouse sync from", + "Warehouse sync request", + "warehouse sync response", + ), + ) + .await + .map(|response| response.applied) +} + +/// Each failure names the action, e.g. "could not read published SQL from the server". +async fn send_json( + request: reqwest::RequestBuilder, + (unreachable, action, result): (&str, &str, &str), +) -> Result { + let response = request + .send() + .await + .map_err(|_| std::io::Error::other(format!("could not {unreachable} the server")))?; + if !response.status().is_success() { + return Err(rejection(response, action).await); + } + response + .json() + .await + .map_err(|_| std::io::Error::other(format!("server returned an invalid {result}"))) +} + +/// Each segment is percent-encoded, so a target or node key cannot change the route. +fn target_path( + client: &PersonClient, + (location, target): (Uuid, &str), + rest: &[&str], +) -> Result { let mut url = reqwest::Url::parse(&client.origin).map_err(std::io::Error::other)?; url.path_segments_mut() .map_err(|()| std::io::Error::other("invalid server origin"))? @@ -115,39 +192,26 @@ pub(crate) async fn node_sql( &location.to_string(), "targets", target, - "nodes", - node, - "sql", - ]); - let response = authorized(client, reqwest::Method::GET, url.path(), credential)? - .send() - .await - .map_err(|_| std::io::Error::other("could not read published SQL from the server"))?; + ]) + .extend(rest); + Ok(url.path().to_owned()) +} + +async fn rejection(response: reqwest::Response, action: &str) -> std::io::Error { let status = response.status(); - if !status.is_success() { - if status == reqwest::StatusCode::UNAUTHORIZED { - return Err(std::io::Error::other( - "CLI session is unavailable or revoked; run sqb login again", - )); - } - let decision = response - .json::() - .await - .map_err(|_| { - std::io::Error::other(format!( - "SQL inspection rejected (HTTP {}); invalid error response", - status.as_u16() - )) - })? - .decision_id; - return Err(std::io::Error::other(format!( - "SQL inspection rejected (HTTP {}{})", + if status == reqwest::StatusCode::UNAUTHORIZED { + return std::io::Error::other("CLI session is unavailable or revoked; run sqb login again"); + } + match response.json::().await { + Ok(body) => std::io::Error::other(format!( + "{action} rejected (HTTP {}{})", status.as_u16(), - decision.map_or_else(String::new, |id| format!("; decision {id}")) - ))); + body.decision_id + .map_or_else(String::new, |id| format!("; decision {id}")) + )), + Err(_) => std::io::Error::other(format!( + "{action} rejected (HTTP {}); invalid error response", + status.as_u16() + )), } - response - .json() - .await - .map_err(|_| std::io::Error::other("server returned an invalid SQL inspection result")) } diff --git a/crates/sqlbuild-cli/src/person_login/main/mod.rs b/crates/sqlbuild-cli/src/person_login/main/mod.rs index bf1d2af998..db72101ec9 100644 --- a/crates/sqlbuild-cli/src/person_login/main/mod.rs +++ b/crates/sqlbuild-cli/src/person_login/main/mod.rs @@ -1,2 +1,3 @@ pub(crate) mod read_node_sql; +pub(crate) mod request_warehouse_sync; pub(crate) mod run_login; diff --git a/crates/sqlbuild-cli/src/person_login/main/read_node_sql.rs b/crates/sqlbuild-cli/src/person_login/main/read_node_sql.rs index 99970998c7..b751802dc6 100644 --- a/crates/sqlbuild-cli/src/person_login/main/read_node_sql.rs +++ b/crates/sqlbuild-cli/src/person_login/main/read_node_sql.rs @@ -2,31 +2,21 @@ use std::path::Path; -use sqlbuild_orchestration::api::main::parse_origin::parse_origin; use sqlbuild_orchestration::snapshots::models::NodeSql; use uuid::Uuid; -use crate::person_login::_helpers::{credentials, transport}; +use crate::person_login::_helpers::transport; pub(crate) fn read_node_sql( - (server, ca): (&str, Option<&Path>), + server: (&str, Option<&Path>), (location, target, node): (Uuid, &str, &str), ) -> Result, std::io::Error> { - let origin = parse_origin(server, "CLI").map_err(std::io::Error::other)?; if target.trim().is_empty() || node.trim().is_empty() { return Err(std::io::Error::other( "target and exact node key must not be blank", )); } - let credential = credentials::load(&origin)? - .ok_or_else(|| std::io::Error::other("not signed in; run sqb login"))?; - let client = transport::connect(&origin, ca)?; - let runtime = tokio::runtime::Builder::new_multi_thread() - .enable_all() - .build()?; - runtime.block_on(transport::node_sql( - &client, - &credential, - (location, target, node), - )) + transport::with_person_session(server, async |client, credential| { + transport::node_sql(client, credential, (location, target, node)).await + }) } diff --git a/crates/sqlbuild-cli/src/person_login/main/request_warehouse_sync.rs b/crates/sqlbuild-cli/src/person_login/main/request_warehouse_sync.rs new file mode 100644 index 0000000000..28f04804a5 --- /dev/null +++ b/crates/sqlbuild-cli/src/person_login/main/request_warehouse_sync.rs @@ -0,0 +1,16 @@ +//! A saved person session asks a location to read the warehouse now; the server checks the target. + +use std::path::Path; + +use uuid::Uuid; + +use crate::person_login::_helpers::transport; + +pub(crate) fn request_warehouse_sync( + server: (&str, Option<&Path>), + (location, target): (Uuid, &str), +) -> Result { + transport::with_person_session(server, async |client, credential| { + transport::warehouse_sync(client, credential, (location, target)).await + }) +} diff --git a/crates/sqlbuild-orchestration/src/api/_helpers/authentication/password_sessions.rs b/crates/sqlbuild-orchestration/src/api/_helpers/authentication/password_sessions.rs index 7788726caf..f048bc2f1b 100644 --- a/crates/sqlbuild-orchestration/src/api/_helpers/authentication/password_sessions.rs +++ b/crates/sqlbuild-orchestration/src/api/_helpers/authentication/password_sessions.rs @@ -110,6 +110,10 @@ pub(crate) fn person_routes() -> Router { "/api/control/locations/{location_id}/targets/{target}/connection-tests", post(crate::api::_helpers::resources::person_location_work::request), ) + .route( + "/api/control/locations/{location_id}/targets/{target}/warehouse-syncs", + post(crate::api::_helpers::resources::person_location_work::request_sync), + ) .route( "/api/control/locations/{location_id}/targets/{target}/connection-tests/{request_id}", get(crate::api::_helpers::resources::person_location_work::result), diff --git a/crates/sqlbuild-orchestration/src/api/_helpers/resources/person_location_work.rs b/crates/sqlbuild-orchestration/src/api/_helpers/resources/person_location_work.rs index f3040a94db..142f7425d3 100644 --- a/crates/sqlbuild-orchestration/src/api/_helpers/resources/person_location_work.rs +++ b/crates/sqlbuild-orchestration/src/api/_helpers/resources/person_location_work.rs @@ -1,4 +1,4 @@ -//! Person connection tests reuse the location's named outbound work, not arbitrary commands. +//! Person connection tests and warehouse syncs reuse the location's named outbound work, not arbitrary commands. use axum::extract::{Path, State}; use axum::{Extension, Json}; @@ -11,14 +11,14 @@ use crate::credentials::models::AuthenticatedPerson; use crate::platform::main::read_person_connection_test::read_person_connection_test; use crate::platform::main::request_location_work::request_location_work; use crate::platform::models::{ - ConnectionTestRequest, LocationWork, LocationWorkRecord, LocationWorkRequest, + LocationWork, LocationWorkRecord, LocationWorkRequest, PersonWorkRequest, }; use crate::store::models::ControlStore; #[utoipa::path(post, path = "/api/control/locations/{location_id}/targets/{target}/connection-tests", operation_id = "requestConnectionTest", params(("location_id" = Uuid, Path), ("target" = String, Path), ("x-sqb-csrf" = String, Header)), - request_body = ConnectionTestRequest, + request_body = PersonWorkRequest, responses((status = 200, body = MutationResponse), (status = 400, body = crate::api::models::ApiErrorBody), (status = 401, description = "Session rejected"), (status = 403, body = crate::api::models::ApiErrorBody)), security(("session" = []), ("cliSession" = [])))] @@ -26,18 +26,56 @@ pub(crate) async fn request( (State(store), Extension(context)): (State, Extension), Path((location_id, target)): Path<(Uuid, String)>, Extension(AuthenticatedPerson(principal, _)): Extension, - Json(request): Json, + Json(request): Json, +) -> Result, ApiError> { + request_work( + (&store, &context, principal), + (location_id, target), + request, + LocationWork::CheckConnection {}, + ) + .await +} + +#[utoipa::path(post, path = "/api/control/locations/{location_id}/targets/{target}/warehouse-syncs", + operation_id = "requestWarehouseSync", + params(("location_id" = Uuid, Path), ("target" = String, Path), ("x-sqb-csrf" = String, Header)), + request_body = PersonWorkRequest, + responses((status = 200, body = MutationResponse), (status = 400, body = crate::api::models::ApiErrorBody), + (status = 401, description = "Session rejected"), (status = 403, body = crate::api::models::ApiErrorBody)), + security(("session" = []), ("cliSession" = [])))] +pub(crate) async fn request_sync( + (State(store), Extension(context)): (State, Extension), + Path((location_id, target)): Path<(Uuid, String)>, + Extension(AuthenticatedPerson(principal, _)): Extension, + Json(request): Json, +) -> Result, ApiError> { + request_work( + (&store, &context, principal), + (location_id, target), + request, + LocationWork::SyncWarehouse {}, + ) + .await +} + +/// Both requests need connection-test authority: each makes the location read the warehouse, never write it. +async fn request_work( + (store, context, principal): (&ControlStore, &AuditContext, Uuid), + (location_id, target): (Uuid, String), + request: PersonWorkRequest, + work: LocationWork, ) -> Result, ApiError> { Ok(Json(MutationResponse { applied: request_location_work( - &store, - (&context, Some(principal)), + store, + (context, Some(principal)), &LocationWorkRequest { request_id: request.request_id, location_id, target, reason: request.reason, - work: LocationWork::CheckConnection {}, + work, }, ) .await?, diff --git a/crates/sqlbuild-orchestration/src/api/_helpers/resources/router.rs b/crates/sqlbuild-orchestration/src/api/_helpers/resources/router.rs index 7613b7fada..5df6d1402d 100644 --- a/crates/sqlbuild-orchestration/src/api/_helpers/resources/router.rs +++ b/crates/sqlbuild-orchestration/src/api/_helpers/resources/router.rs @@ -39,6 +39,10 @@ pub(crate) fn router(store: &ControlStore, feed: ChangeFeed) -> Router, + Extension(identity): Extension, + Path((location_id, target)): Path<(Uuid, String)>, + Json(request): Json, +) -> Result, ApiError> { + require_location(&identity, location_id)?; + Ok(Json( + read_sync_due(&store, (location_id, &target), &request).await?, + )) +} diff --git a/crates/sqlbuild-orchestration/src/api/models.rs b/crates/sqlbuild-orchestration/src/api/models.rs index c52635f5df..54d45356d9 100644 --- a/crates/sqlbuild-orchestration/src/api/models.rs +++ b/crates/sqlbuild-orchestration/src/api/models.rs @@ -59,6 +59,7 @@ use crate::api::_helpers::resources::{ person_bootstrap::read, person_maintenance::coverage, person_location_work::request, + person_location_work::request_sync, person_location_work::result, person_plans::request, person_plans::read, @@ -96,6 +97,7 @@ use crate::api::_helpers::resources::{ trigger_reads::ticks, warehouse_metadata::report, warehouse_metadata::statistics_due, + warehouse_metadata::sync_due, warehouse_metadata::watermark ))] pub(crate) struct ControlApi; diff --git a/crates/sqlbuild-orchestration/src/platform/_helpers/recent_health.rs b/crates/sqlbuild-orchestration/src/platform/_helpers/recent_health.rs index 40e68baa26..7800fbbf9c 100644 --- a/crates/sqlbuild-orchestration/src/platform/_helpers/recent_health.rs +++ b/crates/sqlbuild-orchestration/src/platform/_helpers/recent_health.rs @@ -2,42 +2,53 @@ use uuid::Uuid; +use crate::platform::constants::LOCATION_RETENTION_DAYS; use crate::platform::models::LocationHealthObservation; use crate::store::errors::ControlError; use crate::store::main::decode_row::decode_row; use crate::store::main::query::query; use crate::store::models::{ControlTransaction, Parameter}; -/// At most 100, newest first, from today's and yesterday's reports plus each instance's last check-in, of one target when given. +/// At most 100, newest first, from the health retention window (a sync follows only runs) plus instance check-ins. pub(crate) async fn recent_health( transaction: ControlTransaction, location_id: Uuid, target: Option<&str>, ) -> Result<(ControlTransaction, Vec), ControlError> { let today = transaction.now.date_naive(); - let yesterday = today - chrono::TimeDelta::days(1); + let days = 0..=LOCATION_RETENTION_DAYS; + let target_placeholder = days.clone().count() + 1; let (event_target, instance_target) = match target { - Some(_) => (" AND payload->'report'->>'target'=$3", " AND target=$2"), - None => ("", ""), + Some(_) => ( + format!(" AND payload->'report'->>'target'=${target_placeholder}"), + " AND target=$2", + ), + None => (String::new(), ""), }; let target_parameter: Vec = target .map(|target| Parameter::Text(target.to_owned())) .into_iter() .collect(); + let daily = days + .clone() + .map(|day| { + format!("SELECT * FROM (SELECT occurred_at AS received_at, invocation_sequence, payload FROM recall_events \ + WHERE invocation_id=${} AND event_type='location_health'{event_target} ORDER BY invocation_sequence DESC LIMIT 100) day_{day}", day + 1) + }) + .collect::>() + .join(" UNION ALL "); let (transaction, rows) = query( transaction, - &format!("SELECT received_at, payload FROM (\ - SELECT * FROM (SELECT occurred_at AS received_at, invocation_sequence, payload FROM recall_events \ - WHERE invocation_id=$1 AND event_type='location_health'{event_target} ORDER BY invocation_sequence DESC LIMIT 100) current_day \ - UNION ALL \ - SELECT * FROM (SELECT occurred_at AS received_at, invocation_sequence, payload FROM recall_events \ - WHERE invocation_id=$2 AND event_type='location_health'{event_target} ORDER BY invocation_sequence DESC LIMIT 100) previous_day\ - ) observations ORDER BY received_at DESC, invocation_sequence DESC LIMIT 100"), - [ - Parameter::Text(format!("location/{location_id}/{today}")), - Parameter::Text(format!("location/{location_id}/{yesterday}")), - ] - .into_iter() + &format!( + "SELECT received_at, payload FROM ({daily}) observations \ + ORDER BY received_at DESC, invocation_sequence DESC LIMIT 100" + ), + days.map(|day| { + Parameter::Text(format!( + "location/{location_id}/{}", + today - chrono::TimeDelta::days(day) + )) + }) .chain(target_parameter.clone()) .collect(), ) diff --git a/crates/sqlbuild-orchestration/src/platform/_helpers/work_client.rs b/crates/sqlbuild-orchestration/src/platform/_helpers/work_client.rs index 6dee6aa9dd..eaf407d5aa 100644 --- a/crates/sqlbuild-orchestration/src/platform/_helpers/work_client.rs +++ b/crates/sqlbuild-orchestration/src/platform/_helpers/work_client.rs @@ -10,7 +10,8 @@ use crate::platform::models::{ LocationWorkResult, }; use crate::warehouse_metadata::models::{ - MetadataBatch, StatisticsDueRead, StatisticsRead, WarehouseSyncResponse, WarehouseWatermarkRead, + MetadataBatch, StatisticsDueRead, StatisticsRead, WarehouseSyncDue, WarehouseSyncDueRead, + WarehouseSyncResponse, WarehouseWatermarkRead, }; impl LocationWorkClient { @@ -44,6 +45,15 @@ impl LocationWorkClient { .await } + pub async fn metadata_sync_due( + &self, + target: &str, + request: &WarehouseSyncDueRead, + ) -> Result { + self.post_metadata((target, Some("sync-due")), request) + .await + } + /// Metadata shares the location's scoped pool, not the scheduler's execution authority. pub async fn sync_metadata( &self, diff --git a/crates/sqlbuild-orchestration/src/platform/main/claim_location_work.rs b/crates/sqlbuild-orchestration/src/platform/main/claim_location_work.rs index 0337aecc42..ffc4910264 100644 --- a/crates/sqlbuild-orchestration/src/platform/main/claim_location_work.rs +++ b/crates/sqlbuild-orchestration/src/platform/main/claim_location_work.rs @@ -28,7 +28,7 @@ pub async fn claim_location_work( "SELECT w.request_id FROM sqb_location_work w WHERE w.location_id=$1 AND w.target=$2 \ AND w.expires_at>$3 AND w.completed_at IS NULL \ AND (w.claimed_until IS NULL OR w.claimed_until<=$3 OR w.session_id=$4) \ - AND (w.request_json->'work'->>'kind' IN ('check_connection','plan') OR EXISTS ( \ + AND (w.request_json->'work'->>'kind' IN ('check_connection','sync_warehouse','plan') OR EXISTS ( \ SELECT 1 FROM sqb_sensor_evaluation_events e JOIN sqb_triggers t ON t.trigger_id=e.trigger_id \ JOIN sqb_trigger_state s ON s.trigger_id=t.trigger_id JOIN sqb_scheduler_lease l ON l.name='scheduler' \ WHERE e.evaluation_id=w.request_id AND e.phase='requested' AND s.status='running' \ diff --git a/crates/sqlbuild-orchestration/src/platform/main/complete_location_work.rs b/crates/sqlbuild-orchestration/src/platform/main/complete_location_work.rs index cc19fc50ad..8edbec28d0 100644 --- a/crates/sqlbuild-orchestration/src/platform/main/complete_location_work.rs +++ b/crates/sqlbuild-orchestration/src/platform/main/complete_location_work.rs @@ -93,7 +93,8 @@ fn redacted(result: &LocationWorkResult) -> LocationWorkResult { fn valid_outcome(work: &LocationWork, outcome: &LocationWorkOutcome) -> bool { match (work, outcome) { (_, LocationWorkOutcome::Superseded {}) - | (LocationWork::CheckConnection {}, LocationWorkOutcome::Connection { .. }) => true, + | (LocationWork::CheckConnection {}, LocationWorkOutcome::Connection { .. }) + | (LocationWork::SyncWarehouse {}, LocationWorkOutcome::SyncStarted {}) => true, (LocationWork::Plan { .. }, LocationWorkOutcome::Plan { document }) => { document .as_object() diff --git a/crates/sqlbuild-orchestration/src/platform/main/request_location_work.rs b/crates/sqlbuild-orchestration/src/platform/main/request_location_work.rs index 0e6ca1c7ed..fa1e4427f7 100644 --- a/crates/sqlbuild-orchestration/src/platform/main/request_location_work.rs +++ b/crates/sqlbuild-orchestration/src/platform/main/request_location_work.rs @@ -29,7 +29,7 @@ pub async fn request_location_work( "plans must be requested by a person on the current publication".into(), )); } - LocationWork::CheckConnection {} => {} + LocationWork::CheckConnection {} | LocationWork::SyncWarehouse {} => {} } if request.target.trim().is_empty() || request.reason.trim().is_empty() diff --git a/crates/sqlbuild-orchestration/src/platform/models.rs b/crates/sqlbuild-orchestration/src/platform/models.rs index 928081c82e..d1e91f47d8 100644 --- a/crates/sqlbuild-orchestration/src/platform/models.rs +++ b/crates/sqlbuild-orchestration/src/platform/models.rs @@ -69,6 +69,8 @@ pub struct LocationHealthObservation { #[serde(tag = "kind", rename_all = "snake_case", deny_unknown_fields)] pub enum LocationWork { CheckConnection {}, + /// Read SQLBuild's state and catalog back from the warehouse now, not only after a run ends. + SyncWarehouse {}, EvaluateSensor { #[schema(value_type = Object)] evaluation: Box, @@ -100,9 +102,10 @@ pub struct PlanRequest { pub command: Vec, } +/// A person asks a location for one named read-only operation, such as a connection test. #[derive(Clone, Debug, Serialize, Deserialize, utoipa::ToSchema)] #[serde(deny_unknown_fields)] -pub struct ConnectionTestRequest { +pub struct PersonWorkRequest { pub request_id: Uuid, pub reason: String, } @@ -138,6 +141,8 @@ pub enum LocationWorkOutcome { #[schema(value_type = Object)] document: serde_json::Value, }, + /// The location took the request; its next metadata sync reads the warehouse. + SyncStarted {}, Superseded {}, } diff --git a/crates/sqlbuild-orchestration/src/store/schema.sql b/crates/sqlbuild-orchestration/src/store/schema.sql index f12fd87027..215b451ebb 100644 --- a/crates/sqlbuild-orchestration/src/store/schema.sql +++ b/crates/sqlbuild-orchestration/src/store/schema.sql @@ -162,6 +162,7 @@ CREATE TABLE sqb_runs ( last_heartbeat_at {time} ); CREATE INDEX sqb_runs_location_time ON sqb_runs(location_id, target, queued_at); +CREATE INDEX sqb_runs_location_ended ON sqb_runs(location_id, target, ended_at); CREATE INDEX sqb_runs_queue ON sqb_runs(priority DESC, queued_at) WHERE status = 'queued'; CREATE INDEX sqb_runs_active ON sqb_runs(location_id, target, job_key) WHERE status IN ('queued', 'launched', 'running', 'unresponsive'); CREATE TABLE sqb_run_nodes ( diff --git a/crates/sqlbuild-orchestration/src/warehouse_metadata/main/location_reads/mod.rs b/crates/sqlbuild-orchestration/src/warehouse_metadata/main/location_reads/mod.rs index af7a2a5d7a..05bf3ab152 100644 --- a/crates/sqlbuild-orchestration/src/warehouse_metadata/main/location_reads/mod.rs +++ b/crates/sqlbuild-orchestration/src/warehouse_metadata/main/location_reads/mod.rs @@ -1,2 +1,3 @@ pub mod read_statistics_due; +pub mod read_sync_due; pub mod read_watermark; diff --git a/crates/sqlbuild-orchestration/src/warehouse_metadata/main/location_reads/read_sync_due.rs b/crates/sqlbuild-orchestration/src/warehouse_metadata/main/location_reads/read_sync_due.rs new file mode 100644 index 0000000000..573db4f6a5 --- /dev/null +++ b/crates/sqlbuild-orchestration/src/warehouse_metadata/main/location_reads/read_sync_due.rs @@ -0,0 +1,29 @@ +//! Runs that ended, not a stored request, decide when a location synchronizes again. + +use uuid::Uuid; + +use crate::store::errors::ControlError; +use crate::store::main::{commit::commit, query::query}; +use crate::store::models::{ControlStore, Parameter}; +use crate::warehouse_metadata::_helpers::ingestion::begin_ingestion; +use crate::warehouse_metadata::models::{WarehouseSyncDue, WarehouseSyncDueRead}; + +pub async fn read_sync_due( + store: &ControlStore, + scope: (Uuid, &str), + request: &WarehouseSyncDueRead, +) -> Result { + let transaction = + begin_ingestion(store, scope, (request.session_id, request.generation)).await?; + let now = transaction.now; + let (transaction, rows) = query( + transaction, + "SELECT MAX(ended_at) AS ended_at FROM sqb_runs WHERE location_id=$1 AND target=$2", + vec![Parameter::Id(scope.0), Parameter::Text(scope.1.into())], + ) + .await?; + let runs_ended_at = serde_json::from_value(rows[0]["ended_at"].clone()) + .map_err(|error| ControlError::Invalid(error.to_string()))?; + commit(transaction).await?; + Ok(WarehouseSyncDue { runs_ended_at, now }) +} diff --git a/crates/sqlbuild-orchestration/src/warehouse_metadata/models.rs b/crates/sqlbuild-orchestration/src/warehouse_metadata/models.rs index d678e6d29a..4c44e2d64e 100644 --- a/crates/sqlbuild-orchestration/src/warehouse_metadata/models.rs +++ b/crates/sqlbuild-orchestration/src/warehouse_metadata/models.rs @@ -390,6 +390,22 @@ pub struct WarehouseWatermarkRead { pub kind: WarehouseHistoryKind, } +/// A registered reader asks whether a run ended since its last synchronization began. +#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize, utoipa::ToSchema)] +#[serde(deny_unknown_fields)] +pub struct WarehouseSyncDueRead { + pub session_id: Uuid, + pub generation: i64, +} + +/// Both times come from the server clock, so the reader compares them without clock skew. +#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize, utoipa::ToSchema)] +#[serde(deny_unknown_fields)] +pub struct WarehouseSyncDue { + pub runs_ended_at: Option>, + pub now: DateTime, +} + /// Private interpreter input; cursor values are data and receive adapter literal rendering. #[derive(Clone, Debug, Serialize, Deserialize)] #[serde(deny_unknown_fields)] diff --git a/crates/sqlbuild-orchestration/tests/api/test_location_health.rs b/crates/sqlbuild-orchestration/tests/api/test_location_health.rs index 928b2759ca..14a932e7cc 100644 --- a/crates/sqlbuild-orchestration/tests/api/test_location_health.rs +++ b/crates/sqlbuild-orchestration/tests/api/test_location_health.rs @@ -6,8 +6,10 @@ use crate::platform::main::report_remote_location_health::report_remote_location use crate::platform::models::{LocationHealth, LocationHealthObservation, LocationHealthReport}; use crate::snapshots::main::publish_snapshot::publish_snapshot; use crate::store::main::migrate_store::migrate_store; +use crate::store::main::{begin::begin, commit::commit, query::query}; +use crate::store::models::{Parameter, TransactionMode}; use crate::store::tests::helpers::read_recall_events; -use chrono::Utc; +use chrono::{TimeDelta, Utc}; use reqwest::StatusCode; use serde_json::json; use uuid::Uuid; @@ -202,6 +204,28 @@ fn given_scoped_location_observations_when_retried_or_read_then_history_and_auth .is_empty(), "another target's read leaves this target's observations out" ); + // A location syncs only after runs, so its newest report can be days old. + for (age, kept) in [(7, true), (8, false), (0, true)] { + let transaction = begin(store, TransactionMode::Write).await.expect("age"); + let day = (transaction.now - TimeDelta::days(age)).date_naive(); + let (transaction, _) = query( + transaction, + "UPDATE recall_events SET invocation_id=$1 WHERE event_type='location_health'", + vec![Parameter::Text(format!("location/{location_id}/{day}"))], + ) + .await + .expect("aged report"); + commit(transaction).await.expect("aged report committed"); + let reports = person_read("preview #1").await.expect("aged health"); + assert_eq!( + reports + .iter() + .any(|observation| observation.report == connection), + kept, + "{}: a report {age} days old is read within location health retention", + test_case.description + ); + } let events = read_recall_events(store, 0, 100) .await .expect("read durable events"); diff --git a/crates/sqlbuild-orchestration/tests/api/test_location_work.rs b/crates/sqlbuild-orchestration/tests/api/test_location_work.rs index ac5bff2d86..5752d74e7f 100644 --- a/crates/sqlbuild-orchestration/tests/api/test_location_work.rs +++ b/crates/sqlbuild-orchestration/tests/api/test_location_work.rs @@ -290,6 +290,40 @@ fn given_outbound_work_when_claimed_retried_or_replaced_then_scope_and_immutable .expect("expired excluded") .is_none() ); + let sync = LocationWorkRequest { + request_id: Uuid::now_v7(), + work: LocationWork::SyncWarehouse {}, + ..next + }; + request_location_work(store, (&audit, None), &sync) + .await + .expect("sync request"); + let claim = worker.poll(&poll).await.expect("poll").expect("sync claim"); + assert_eq!(claim.request.work, LocationWork::SyncWarehouse {}); + let answer = |outcome| LocationWorkResult { + claim_id: claim.claim_id, + session_id: poll.session_id, + outcome, + }; + assert!( + worker + .complete( + sync.request_id, + &answer(LocationWorkOutcome::Connection { healthy: true }) + ) + .await + .is_err(), + "a sync request is not answered by a connection result" + ); + assert!( + worker + .complete( + sync.request_id, + &answer(LocationWorkOutcome::SyncStarted {}) + ) + .await + .expect("sync started") + ); stop_api(api).await; close_test_store(fixture).await; }); diff --git a/crates/sqlbuild-orchestration/tests/api/test_person_location_work.rs b/crates/sqlbuild-orchestration/tests/api/test_person_location_work.rs index 552171a323..0e5058430d 100644 --- a/crates/sqlbuild-orchestration/tests/api/test_person_location_work.rs +++ b/crates/sqlbuild-orchestration/tests/api/test_person_location_work.rs @@ -1,4 +1,4 @@ -//! Person connection tests use live built-ins and the same atomic work queue on both stores. +//! Person connection tests and warehouse syncs use live built-ins and the same atomic work queue on both stores. use uuid::Uuid; @@ -130,6 +130,47 @@ fn given_person_connection_tests_when_retried_and_revoked_then_live_scope_preced ); assert_eq!(result.requested_by, format!("principal:{person}")); assert!(result.result_json.is_none()); + let sync = LocationWorkRequest { + request_id: Uuid::now_v7(), + reason: "Sync orders state now".into(), + work: LocationWork::SyncWarehouse {}, + ..request.clone() + }; + assert!( + request_location_work(store, (&context, Some(credentials)), &sync) + .await + .expect("connection-test authority requests a sync") + ); + assert!( + read_person_connection_test( + store, + (&context, credentials), + (location, "development", sync.request_id) + ) + .await + .expect("authorized read") + .is_none(), + "a sync request never reads back as a connection test" + ); + for (other, target) in [(Uuid::now_v7(), "development"), (location, "preview")] { + assert!( + matches!( + request_location_work( + store, + (&context, Some(credentials)), + &LocationWorkRequest { + request_id: Uuid::now_v7(), + location_id: other, + target: target.into(), + ..sync.clone() + } + ) + .await, + Err(ControlError::Denied { .. }) + ), + "a sync outside the Developer's location and target is denied" + ); + } for (other, target) in [(Uuid::now_v7(), "development"), (location, "preview")] { assert!(matches!( read_person_connection_test( @@ -156,7 +197,7 @@ fn given_person_connection_tests_when_retried_and_revoked_then_live_scope_preced (&context, sessions[0].identity.principal_id), &RoleAssignmentRequest { revoke: true, - ..assignment + ..assignment.clone() }, ) .await @@ -165,6 +206,18 @@ fn given_person_connection_tests_when_retried_and_revoked_then_live_scope_preced request_location_work(store, (&context, Some(credentials)), &request).await, Err(ControlError::Denied { .. }) )); + assert!(matches!( + request_location_work( + store, + (&context, Some(credentials)), + &LocationWorkRequest { + request_id: Uuid::now_v7(), + ..sync.clone() + } + ) + .await, + Err(ControlError::Denied { .. }) + )); assert!(matches!( read_person_connection_test( store, @@ -174,6 +227,44 @@ fn given_person_connection_tests_when_retried_and_revoked_then_live_scope_preced .await, Err(ControlError::Denied { .. }) )); + for role in [BuiltinRole::Viewer, BuiltinRole::Operator] { + let lesser = RoleAssignmentRequest { + role: role.into(), + ..assignment.clone() + }; + assign_builtin_role( + store, + (&context, sessions[0].identity.principal_id), + &lesser, + ) + .await + .expect("assign lesser role"); + assert!( + matches!( + request_location_work( + store, + (&context, Some(credentials)), + &LocationWorkRequest { + request_id: Uuid::now_v7(), + ..sync.clone() + } + ) + .await, + Err(ControlError::Denied { .. }) + ), + "{role:?} cannot request a sync" + ); + assign_builtin_role( + store, + (&context, sessions[0].identity.principal_id), + &RoleAssignmentRequest { + revoke: true, + ..lesser + }, + ) + .await + .expect("revoke lesser role"); + } close_test_store(fixture).await; }); } diff --git a/crates/sqlbuild-orchestration/tests/api/test_warehouse_metadata.rs b/crates/sqlbuild-orchestration/tests/api/test_warehouse_metadata.rs index 9aae0418f6..56536e0d3f 100644 --- a/crates/sqlbuild-orchestration/tests/api/test_warehouse_metadata.rs +++ b/crates/sqlbuild-orchestration/tests/api/test_warehouse_metadata.rs @@ -349,6 +349,12 @@ fn given_warehouse_observations_when_pages_retry_then_scope_history_and_atomicit (&client, &url, &token.bearer), ) .await; + crate::api::tests::warehouse_metadata_helpers::sync_due_scope( + store, + (session_id, preview_session), + (&client, &url, &token.bearer), + ) + .await; crate::api::tests::warehouse_reconciliation_helpers::reconcile_history( store, (&client, &url, &token.bearer), diff --git a/crates/sqlbuild-orchestration/tests/api/warehouse_metadata/helpers.rs b/crates/sqlbuild-orchestration/tests/api/warehouse_metadata/helpers.rs index 3e9aa5a83f..103e9001b4 100644 --- a/crates/sqlbuild-orchestration/tests/api/warehouse_metadata/helpers.rs +++ b/crates/sqlbuild-orchestration/tests/api/warehouse_metadata/helpers.rs @@ -9,7 +9,7 @@ use crate::store::main::decompress::decompress; use crate::store::main::{ begin::begin, commit::commit, migrate_store::migrate_store, query::query, }; -use crate::store::models::{ControlStore, TransactionMode}; +use crate::store::models::{ControlStore, Parameter, TransactionMode}; use crate::store::tests::fault_helpers::{install_fault, remove_fault}; use crate::store::tests::fault_types::TransactionFault; use crate::warehouse_metadata::models::{MetadataBatch, WarehouseWatermarkRead}; @@ -515,6 +515,90 @@ pub(crate) async fn fingerprint_history( } } +/// Sync-due answers only the fenced reader's own location and target, with nothing but two times. +pub(crate) async fn sync_due_scope( + store: &ControlStore, + (session_id, preview_session_id): (Uuid, Uuid), + (client, url, bearer): (&Client, &str, &str), +) { + let ended_at = "2026-01-01T09:00:00Z"; + let run_id = Uuid::now_v7(); + let (transaction, _) = query(begin(store, TransactionMode::Write).await.expect("write"), + "INSERT INTO sqb_runs (run_id,location_id,target,snapshot_id,kind,origin,root_run_id,requested_by,\ + selection_json,request_json,write_set_digest,status,queued_at,ended_at,node_counts_json,last_event_storage_id) \ + SELECT $1,location_id,target,snapshot_id,'adhoc','manual',$2,'test-admin',$3,$4,'','success',$5,$6,$7,0 \ + FROM sqb_current_snapshots WHERE target='preview'", + vec![Parameter::Id(run_id), Parameter::Id(run_id), Parameter::Json(json!([])), Parameter::Json(json!({})), + Parameter::Time(ended_at.parse().expect("time")), Parameter::Time(ended_at.parse().expect("time")), + Parameter::Json(json!({}))]).await.expect("preview run"); + commit(transaction).await.expect("close write"); + let due_url = format!("{url}/sync-due"); + let preview_url = due_url.replace("/development/", "/preview/"); + let body = + |session: Uuid, generation: i64| json!({"session_id": session, "generation": generation}); + for (endpoint, session, expected) in [ + (&due_url, session_id, Value::Null), + (&preview_url, preview_session_id, json!(ended_at)), + ] { + let reply: Value = client + .post(endpoint) + .bearer_auth(bearer) + .json(&body(session, 1)) + .send() + .await + .expect("sync due") + .json() + .await + .expect("sync due reply"); + let fields = reply.as_object().expect("object"); + let mut keys: Vec<_> = fields.keys().collect(); + keys.sort(); + assert_eq!(keys, ["now", "runs_ended_at"]); + let ended = fields["runs_ended_at"] + .as_str() + .map(|time| time.parse::>().expect("time")); + let wanted = expected.as_str().map(|time| time.parse().expect("time")); + assert_eq!(ended, wanted, "another target's runs never count"); + } + let other_location = due_url.replace( + url.split('/').nth(5).expect("location id"), + &Uuid::now_v7().to_string(), + ); + for (endpoint, request, status) in [ + (&due_url, body(session_id, 0), StatusCode::CONFLICT), + ( + &due_url, + body(preview_session_id, 1), + StatusCode::BAD_REQUEST, + ), + (&preview_url, body(session_id, 1), StatusCode::BAD_REQUEST), + (&other_location, body(session_id, 1), StatusCode::FORBIDDEN), + ] { + assert_eq!( + client + .post(endpoint) + .bearer_auth(bearer) + .json(&request) + .send() + .await + .expect("refused sync due") + .status(), + status, + "{endpoint} {request}" + ); + } + assert_eq!( + client + .post(&due_url) + .json(&body(session_id, 1)) + .send() + .await + .expect("anonymous sync due") + .status(), + StatusCode::UNAUTHORIZED + ); +} + async fn mirrored_fingerprints(store: &ControlStore) -> Vec { let (transaction, rows) = query(begin(store, TransactionMode::Read).await.expect("read"), "SELECT f.event_id,f.source_event_id,f.definition_text_hash,f.record_json,t.body FROM sqb_warehouse_fingerprints f \ diff --git a/src/sqlbuild/.agents/skills/sqlbuild/references/docs/cli/server.md b/src/sqlbuild/.agents/skills/sqlbuild/references/docs/cli/server.md index a37f7ae52b..51d0f0119e 100644 --- a/src/sqlbuild/.agents/skills/sqlbuild/references/docs/cli/server.md +++ b/src/sqlbuild/.agents/skills/sqlbuild/references/docs/cli/server.md @@ -12,6 +12,7 @@ Online: https://sqlbuild.com/docs/cli/server/ - Set retention - Grant roles to OIDC groups (beta) - Inspect published SQL +- Sync from the warehouse now - Run ad-hoc commands - Plan builds - Define project roles @@ -159,6 +160,26 @@ data. Every successful inspection records an audit entry without copying SQL int audit cannot be saved, the request fails without returning SQL. Revocation takes effect on the next request, including sessions created by [`sqb login`](login.md). +## Sync from the warehouse now + +A code location reads what has been built, checks and freshness back from the warehouse when it +starts and after each run the server starts has ended. It does not see a `sqb build` you run on +your own machine until then. To sync now, select **Sync now** on the location's page, or run: + +```bash +sqb control sync \ + --server https://control.example.com \ + --location-id "$LOCATION_ID" \ + --target development +``` + +The command prints `{"applied": true}` when the server has queued the request. The location starts +the sync within a few seconds, and the top bar shows when it last synced. A sync reads the +warehouse and never writes to it. + +Requesting a sync needs the same access as a connection test: Developer or Admin on the location +and target. Viewer and Operator cannot request one. Every request is recorded in the audit log. + ## Run ad-hoc commands People with `run.adhoc` can run a `sqb` command on the current publication without declaring a diff --git a/tests/integration/src/sqlbuild/compiler/compiled_project/publication/catalog/helpers.py b/tests/integration/src/sqlbuild/compiler/compiled_project/publication/catalog/helpers.py index 3a2de107cd..65f57034c2 100644 --- a/tests/integration/src/sqlbuild/compiler/compiled_project/publication/catalog/helpers.py +++ b/tests/integration/src/sqlbuild/compiler/compiled_project/publication/catalog/helpers.py @@ -4,15 +4,57 @@ import sqlite3 import time from contextlib import closing +from datetime import UTC, datetime, timedelta from pathlib import Path from subprocess import Popen from typing import Any +from uuid import uuid4 + + +def record_finished_run(control: Path) -> None: + """Record a run that just ended for every published target, which makes its location resync.""" + ended_at: str = datetime.now(UTC).strftime("%Y-%m-%dT%H:%M:%S.%fZ") + run_id: str = str(uuid4()) + with closing(sqlite3.connect(control)) as database: + _ = database.execute( + "INSERT INTO sqb_runs (run_id,location_id,target,snapshot_id,kind,origin,root_run_id," + "requested_by,selection_json,request_json,write_set_digest,status,queued_at,ended_at," + "node_counts_json,last_event_storage_id) SELECT ?,location_id,target,snapshot_id," + "'adhoc','manual',?,'test-admin','[]','{}','','success',?,?,'{}',0 " + "FROM sqb_current_snapshots", + (run_id, run_id, ended_at, ended_at), + ) + database.commit() + + +def request_warehouse_sync(control: Path) -> str: + """Queue the location work a person's Sync now creates, for the one published target.""" + request_id: str = str(uuid4()) + now: datetime = datetime.now(UTC) + with closing(sqlite3.connect(control)) as database: + _ = database.execute( + "INSERT INTO sqb_location_work (request_id,location_id,target,snapshot_id,generation," + "request_json,requested_by,requested_at,expires_at) SELECT ?,c.location_id,c.target," + "c.snapshot_id,s.generation,json_object('request_id',?,'location_id',c.location_id," + "'target',c.target,'reason','Sync now','work',json_object('kind','sync_warehouse'))," + "'test-admin',?,? FROM sqb_current_snapshots c JOIN sqb_snapshots s " + "ON s.snapshot_id=c.snapshot_id", + ( + request_id, + request_id, + now.strftime("%Y-%m-%dT%H:%M:%S.%fZ"), + (now + timedelta(minutes=5)).strftime("%Y-%m-%dT%H:%M:%S.%fZ"), + ), + ) + database.commit() + return request_id def wait_for_catalog( control: Path, expected: list[dict[str, Any]], process: Popen[bytes], log: Path ) -> None: """Wait for the exact synthetic relation, not a count that unrelated rows can satisfy.""" + record_finished_run(control) deadline: float = time.monotonic() + 45 observed: list[dict[str, Any]] = [{}] while observed != expected: diff --git a/tests/integration/src/sqlbuild/compiler/compiled_project/publication/test_person_sql.py b/tests/integration/src/sqlbuild/compiler/compiled_project/publication/test_person_sql.py index d7e12791d8..d73478e099 100644 --- a/tests/integration/src/sqlbuild/compiler/compiled_project/publication/test_person_sql.py +++ b/tests/integration/src/sqlbuild/compiler/compiled_project/publication/test_person_sql.py @@ -1,8 +1,10 @@ """An installed CLI inspects actual published SQL through a bounded live person session.""" import json +import sqlite3 import subprocess import sys +from contextlib import closing from functools import partial from pathlib import Path from typing import Any @@ -103,6 +105,23 @@ def test_given_published_orders_when_cli_reads_sql_then_only_live_scoped_authori "--target", "development", ] + sync: list[str] = [ + binary, + "control", + "sync", + *common, + "--location-id", + worker.api.location_id, + "--target", + "development", + ] + for misused in ([*command[:3], *command[4:]], [*sync[:3], "model:orders", *sync[3:]]): + usage: subprocess.CompletedProcess[str] = invoke(misused) + assert (usage.returncode, usage.stdout, usage.stderr[:20]) == ( + 2, + "", + "sqb control: error: ", + ) absent: subprocess.CompletedProcess[str] = invoke(command) assert absent.returncode != 0 and not absent.stdout and "not signed in" in absent.stderr with pending_login( @@ -126,6 +145,12 @@ def test_given_published_orders_when_cli_reads_sql_then_only_live_scoped_authori assert ( denied.returncode != 0 and not denied.stdout and "HTTP 403; decision " in denied.stderr ) + refused: subprocess.CompletedProcess[str] = invoke(sync) + assert ( + refused.returncode != 0 + and not refused.stdout + and "Warehouse sync request rejected (HTTP 403; decision " in refused.stderr + ) administrator, _ = browser(origin, certificate) admin_csrf: str = sign_in( administrator, origin, "password", admin["name"], "orders-person-password-2026" @@ -140,6 +165,15 @@ def test_given_published_orders_when_cli_reads_sql_then_only_live_scoped_authori request(administrator, origin + "/api/admin/roles", assignment, csrf=admin_csrf)[0] == 200 ) + requested: subprocess.CompletedProcess[str] = invoke(sync) + assert requested.returncode == 0, requested.stderr + assert json.loads(requested.stdout) == {"applied": True} + with closing(sqlite3.connect(tmp_path / "control/.sqlbuild/control.sqlite3")) as store: + queued: list[tuple[str]] = store.execute( + "SELECT requested_by FROM sqb_location_work " + "WHERE json_extract(request_json,'$.work.kind')='sync_warehouse'" + ).fetchall() + assert queued == [("principal:" + person["principal_id"],)] inspected: subprocess.CompletedProcess[str] = invoke(command) assert inspected.returncode == 0 and "Published SQL read successfully" in inspected.stderr sql: dict[str, Any] = json.loads(inspected.stdout) @@ -182,6 +216,7 @@ def test_given_published_orders_when_cli_reads_sql_then_only_live_scoped_authori and not revoked.stdout and "HTTP 403; decision " in revoked.stderr ) + assert invoke(sync).returncode != 0 changed: subprocess.CompletedProcess[str] = administer( (Path(binary), tmp_path / "control"), ["set-password", person["principal_id"]], diff --git a/tests/integration/src/sqlbuild/compiler/compiled_project/publication/test_warehouse_metadata.py b/tests/integration/src/sqlbuild/compiler/compiled_project/publication/test_warehouse_metadata.py index 83f43139f4..30d62cfc26 100644 --- a/tests/integration/src/sqlbuild/compiler/compiled_project/publication/test_warehouse_metadata.py +++ b/tests/integration/src/sqlbuild/compiler/compiled_project/publication/test_warehouse_metadata.py @@ -43,6 +43,10 @@ StateEventsTestCase, WarehouseHistoryTestCase, ) +from tests.integration.src.sqlbuild.compiler.compiled_project.publication.catalog.helpers import ( + record_finished_run, + request_warehouse_sync, +) from tests.integration.src.sqlbuild.compiler.compiled_project.publication.helpers import ( append_audit_history, append_column_migration_history, @@ -60,6 +64,7 @@ snapshot_api, sync_warehouse_history, wait_for_generation, + wait_for_location_work, wait_for_warehouse_sync_health, warehouse_history_page, warehouse_history_pages, @@ -396,7 +401,8 @@ def test_given_a_served_location_when_state_event_history_is_rebuilt_and_retaine mirrored = database.execute( f"SELECT event_id,record_json FROM {mirror_table} ORDER BY event_id" ).fetchall() - time.sleep(0.05) + record_finished_run(control) + time.sleep(0.5) assert {json.loads(row)[source_key] for _, row in mirrored} == expected_ids for event_id, text in mirrored: record: dict[str, Any] = json.loads(text) @@ -415,6 +421,7 @@ def test_given_a_served_location_when_state_event_history_is_rebuilt_and_retaine ) _ = adapter.execute(connection=connection, sql=warehouse_sql) adapter.close(connection) + record_finished_run(control) @pytest.mark.parametrize( @@ -945,7 +952,8 @@ def test_given_a_served_location_when_warehouse_history_changes_then_periodic_re "SELECT (SELECT COUNT(*) FROM sqb_warehouse_fingerprints)," "(SELECT COUNT(*) FROM sqb_warehouse_freshness)" ).fetchone() - time.sleep(0.05) + record_finished_run(control) + time.sleep(0.5) with closing(sqlite3.connect(control)) as database: bodies: list[tuple[bytes]] = database.execute( "SELECT t.body FROM sqb_warehouse_fingerprints f " @@ -971,6 +979,7 @@ def test_given_a_served_location_when_warehouse_history_changes_then_periodic_re connection=connection, sql=warehouse_sql.format(table=table) ) adapter.close(connection) + record_finished_run(control) @pytest.mark.parametrize( @@ -1040,6 +1049,7 @@ def test_given_durable_mirror_when_late_rows_arrive_then_incremental_overlap_pre for table in ("sqb_warehouse_fingerprints", "sqb_warehouse_freshness"): _ = database.execute(stage["mirror_sql"].format(table=table)) database.commit() + record_finished_run(control) expected: set[str] = set(stage["expected_runs"]) | { f"middle-{index:03}" for index in range(stage.get("expected_overlap_count", stage["overlap_count"])) @@ -1080,8 +1090,26 @@ def test_given_durable_mirror_when_late_rows_arrive_then_incremental_overlap_pre observed_health_at = max( datetime.fromisoformat(report["received_at"]) for report in reports ) + time.sleep(6) + completed: int = log.read_text().count("Warehouse history synchronization completed.") + time.sleep(12) + assert ( + log.read_text().count("Warehouse history synchronization completed.") == completed + ), "an idle location read the warehouse without a run ending" + work: dict[str, Any] = wait_for_location_work( + api, request_warehouse_sync(control), location, log + ) + assert work["result_json"]["outcome"] == {"kind": "sync_started"} + deadline = time.monotonic() + 45 + while ( + log.read_text().count("Warehouse history synchronization completed.") == completed + ): + assert location.poll() is None, log.read_text() + assert time.monotonic() < deadline, log.read_text() + time.sleep(0.05) for availability in fixture["availability"]: _ = (project / availability["source"]).rename(project / availability["destination"]) + record_finished_run(control) reports = wait_for_warehouse_sync_health( api, availability, location, log, after=observed_health_at ) diff --git a/ui/src/lib/api/client.ts b/ui/src/lib/api/client.ts index 3bb3b5a18a..9a87552387 100644 --- a/ui/src/lib/api/client.ts +++ b/ui/src/lib/api/client.ts @@ -136,6 +136,8 @@ export interface ApiClient extends PlatformApi { /** Null when the location has reported no recent sync. */ getWarehouseSync(scope: Scope): Promise; + /** Asks the location to read SQLBuild's state back from the warehouse now. `connection.test` */ + requestWarehouseSync(scope: Scope): Promise; /** Live "changed" signals (SSE in the server client); consumers refetch. */ changes: ChangeSubscription; diff --git a/ui/src/lib/api/http-client.test.ts b/ui/src/lib/api/http-client.test.ts index b9979ba996..c9bb4078ae 100644 --- a/ui/src/lib/api/http-client.test.ts +++ b/ui/src/lib/api/http-client.test.ts @@ -551,6 +551,35 @@ describe('administration writes', () => { }); }); +describe('requestWarehouseSync', () => { + function server(status: number) { + const calls: Request[] = []; + vi.stubGlobal('document', { cookie: '__Host-sqb-csrf=token-1' }); + vi.stubGlobal( + 'fetch', + vi.fn(async (request: Request) => { + calls.push(request); + return status === 200 ? Response.json({ applied: true }) : Response.json({ error: 'You may not test this location', decision_id: 'd-1' }, { status }); + }) + ); + return { calls, client: createHttpClient({ baseUrl: 'http://ui.test', onUnauthorized: () => {} }) }; + } + + it('asks the location to sync with one CSRF-protected request and does not wait for the sync', async () => { + const { calls, client } = server(200); + await client.requestWarehouseSync(ORDERS); + expect(calls.map((c) => [c.method, new URL(c.url).pathname, c.headers.get('x-sqb-csrf')])).toEqual([ + ['POST', '/api/control/locations/0191f0aa-orders/targets/development/warehouse-syncs', 'token-1'] + ]); + expect(await calls[0].json()).toMatchObject({ reason: 'Requested from the web UI' }); + }); + + it('reports a refused request with its status', async () => { + const { client } = server(403); + await expect(client.requestWarehouseSync(ORDERS)).rejects.toMatchObject({ status: 403 }); + }); +}); + describe('testConnection', () => { const record = (completed: string | null, outcome: unknown = null) => ({ generation: 1, diff --git a/ui/src/lib/api/http-client.ts b/ui/src/lib/api/http-client.ts index 9f3fa8d2de..56cd7f6cac 100644 --- a/ui/src/lib/api/http-client.ts +++ b/ui/src/lib/api/http-client.ts @@ -80,6 +80,7 @@ const SUPPORTED: ReadonlySet = new Set([ 'listRoles', 'getRole', 'getWarehouseSync', + 'requestWarehouseSync', 'maintenanceHistory', 'testConnection', 'listWarehouseConnections', @@ -549,6 +550,16 @@ export function createHttpClient({ baseUrl = '', onUnauthorized }: HttpClientOpt if (!data) throw new ApiError(errorText(error, response), response.status); return warehouseSyncOf(data); }, + /** The location takes the request within seconds; the sync's result arrives as a `warehouse_sync` change. */ + async requestWarehouseSync(scope: Scope): Promise { + const { locationId, target } = parseLocationKey(scope.locationId); + const { error, response } = await http.POST('/api/control/locations/{location_id}/targets/{target}/warehouse-syncs', { + body: { request_id: crypto.randomUUID(), reason: 'Requested from the web UI' }, + params: { path: { location_id: locationId, target }, header: { 'x-sqb-csrf': csrfToken(document.cookie) } } + }); + if (response.status === 401) onUnauthorized(); + if (!response.ok) throw new ApiError(errorText(error, response), response.status); + }, /** Asks the location to connect, then waits for its answer until the request expires. */ async testConnection(scope: Scope): Promise { const { locationId, target } = parseLocationKey(scope.locationId); diff --git a/ui/src/lib/api/http-locations.test.ts b/ui/src/lib/api/http-locations.test.ts index 5521f5162f..50e6d53950 100644 --- a/ui/src/lib/api/http-locations.test.ts +++ b/ui/src/lib/api/http-locations.test.ts @@ -73,7 +73,7 @@ describe('warehouseSyncOf', () => { sync('2026-11-08T12:01:00Z', 'catalog', null), sync('2026-11-08T12:00:00Z', 'statistics', 'delivery') ]); - expect(result).toEqual({ lastSyncedAt: '2026-11-08T12:02:00Z', lastError: 'catalog: failed while reading the warehouse' }); + expect(result).toEqual({ lastSyncedAt: '2026-11-08T12:02:00Z', lastError: 'catalog: failed while reading the warehouse', lastReportedAt: '2026-11-08T12:03:00Z' }); }); it("gives each table its newest sync, the time it last synced cleanly, and where the state tables are read from", () => { diff --git a/ui/src/lib/api/http-locations.ts b/ui/src/lib/api/http-locations.ts index 32a21cd1c4..436cb4d6ad 100644 --- a/ui/src/lib/api/http-locations.ts +++ b/ui/src/lib/api/http-locations.ts @@ -167,6 +167,7 @@ export function warehouseSyncOf(observations: Wire['LocationHealthObservation'][ const failing = [...newest.values()].flatMap((s) => (s.report.health.failure ? [`${s.report.health.table}: failed while ${SYNC_STAGE[s.report.health.failure]}`] : [])); return { lastSyncedAt: syncs.find((s) => !s.report.health.failure)?.received_at ?? null, - lastError: failing.length ? failing.sort().join('\n') : null + lastError: failing.length ? failing.sort().join('\n') : null, + lastReportedAt: syncs[0].received_at }; } diff --git a/ui/src/lib/api/mock/mock-client.ts b/ui/src/lib/api/mock/mock-client.ts index b2f1f64d8f..5e3d7f4561 100644 --- a/ui/src/lib/api/mock/mock-client.ts +++ b/ui/src/lib/api/mock/mock-client.ts @@ -334,6 +334,17 @@ export class MockApiClient implements ApiClient { return this.#reply({ ...this.#sync(scope) }); } + async requestWarehouseSync(scope: Scope): Promise { + this.#require('connection.test'); + const sync = this.#sync(scope); + setTimeout(() => { + const now = new Date().toISOString(); + Object.assign(sync, { lastSyncedAt: now, lastError: null, lastReportedAt: now }); + this.changes.emit(`warehouse_sync:${scope.locationId}`); + }, 1500); + return this.#reply(undefined); + } + // ---- platform and admin: delegated to MockPlatformApi (mock/platform.ts) ---- platformHealth(...args: Parameters) { diff --git a/ui/src/lib/api/types.ts b/ui/src/lib/api/types.ts index 3349a87cc6..895d746c92 100644 --- a/ui/src/lib/api/types.ts +++ b/ui/src/lib/api/types.ts @@ -61,11 +61,13 @@ export type Snapshot = { sqbVersion: string; }; -/** The location reads SQLBuild's own tables back from the warehouse on its own schedule. */ +/** The location reads SQLBuild's own tables back from the warehouse after each run ends, and on request. */ export type WarehouseSync = { lastSyncedAt: Iso | null; /** Why the latest sync failed; null when it succeeded. */ lastError: string | null; + /** When the newest sync report arrived, failed or not; a new one means a requested sync has finished. */ + lastReportedAt?: Iso | null; }; /** The one target a location is deployed against, e.g. `prod` on Snowflake. */ diff --git a/ui/src/lib/components/platform/warehouse-sync-now.svelte b/ui/src/lib/components/platform/warehouse-sync-now.svelte new file mode 100644 index 0000000000..84872ef1c3 --- /dev/null +++ b/ui/src/lib/components/platform/warehouse-sync-now.svelte @@ -0,0 +1,75 @@ + + +
+ Warehouse sync + {#if pending} + + {:else if last?.lastError} + + {:else if last?.lastSyncedAt} + + {:else if last || sync.error} + + {/if} + {#if failure}{failure}{/if} + Sync now +
+ + diff --git a/ui/src/lib/components/shell/warehouse-sync-status.svelte b/ui/src/lib/components/shell/warehouse-sync-status.svelte index 859abdd183..8951d8d87c 100644 --- a/ui/src/lib/components/shell/warehouse-sync-status.svelte +++ b/ui/src/lib/components/shell/warehouse-sync-status.svelte @@ -2,7 +2,7 @@ import { formatAgo, formatStamp } from '$lib/format/time'; import { app } from '$lib/state/app.svelte'; - // The location syncs SQLBuild's own tables from the warehouse on its own schedule; + // The location syncs SQLBuild's own tables from the warehouse after each run ends and on request; // the shell only says when that last happened. const sync = $derived(app.warehouseSync); diff --git a/ui/src/routes/locations/[id]/+page.svelte b/ui/src/routes/locations/[id]/+page.svelte index 3060ab8f1e..7876dc3cdc 100644 --- a/ui/src/routes/locations/[id]/+page.svelte +++ b/ui/src/routes/locations/[id]/+page.svelte @@ -9,6 +9,7 @@ import SegmentedFilter from '$lib/components/list/segmented-filter.svelte'; import StatusLabel from '$lib/components/list/status-label.svelte'; import ConnectionTest from '$lib/components/platform/connection-test.svelte'; + import WarehouseSyncNow from '$lib/components/platform/warehouse-sync-now.svelte'; import KeyValues from '$lib/components/platform/key-values.svelte'; import LocationTokens from '$lib/components/platform/location-tokens.svelte'; import Panel from '$lib/components/platform/panel.svelte'; @@ -95,6 +96,7 @@ + diff --git a/website/src/content/docs/docs/cli/server.mdx b/website/src/content/docs/docs/cli/server.mdx index 0518129d53..a580028186 100644 --- a/website/src/content/docs/docs/cli/server.mdx +++ b/website/src/content/docs/docs/cli/server.mdx @@ -148,6 +148,26 @@ data. Every successful inspection records an audit entry without copying SQL int audit cannot be saved, the request fails without returning SQL. Revocation takes effect on the next request, including sessions created by [`sqb login`](/docs/cli/login/). +## Sync from the warehouse now + +A code location reads what has been built, checks and freshness back from the warehouse when it +starts and after each run the server starts has ended. It does not see a `sqb build` you run on +your own machine until then. To sync now, select **Sync now** on the location's page, or run: + +```bash +sqb control sync \ + --server https://control.example.com \ + --location-id "$LOCATION_ID" \ + --target development +``` + +The command prints `{"applied": true}` when the server has queued the request. The location starts +the sync within a few seconds, and the top bar shows when it last synced. A sync reads the +warehouse and never writes to it. + +Requesting a sync needs the same access as a connection test: Developer or Admin on the location +and target. Viewer and Operator cannot request one. Every request is recorded in the audit log. + ## Run ad-hoc commands People with `run.adhoc` can run a `sqb` command on the current publication without declaring a