From 7c1daaf8cc6ee38c5134f015366a275aa208de23 Mon Sep 17 00:00:00 2001 From: Kevin Longe <273747162+kvlonge@users.noreply.github.com> Date: Fri, 6 Nov 2026 03:15:16 +0000 Subject: [PATCH 1/4] feat: read check evidence through the control API People with view.evidence on a check's model can read the sample rows stored with one of its results. Columns are the rows' keys in sorted order, the order the warehouse stores them in, and a row without a key holds null there. Every read is audited, and a read whose audit cannot be written discloses nothing. The SQL and evidence reads now share one authorize, read and audit path. --- .../orchestration/check_identity.json | 59 +++++++++-- clients/typescript/src/schema.d.ts | 66 +++++++++++++ .../authentication/password_sessions.rs | 4 + .../api/_helpers/resources/person_checks.rs | 27 ++++- .../sqlbuild-orchestration/src/api/models.rs | 9 ++ .../src/warehouse_metadata/_helpers/checks.rs | 98 ++++++++++++++++++- .../src/warehouse_metadata/main/mod.rs | 1 + .../main/read_person_check_evidence.rs | 27 +++++ .../main/read_person_check_sql.rs | 68 +++++-------- .../src/warehouse_metadata/models.rs | 13 +++ .../tests/api/test_checks.rs | 85 ++++++++++++++++ fensu.toml | 1 - 12 files changed, 397 insertions(+), 61 deletions(-) create mode 100644 crates/sqlbuild-orchestration/src/warehouse_metadata/main/read_person_check_evidence.rs diff --git a/benchmarks/duckshop/edge_cases/orchestration/check_identity.json b/benchmarks/duckshop/edge_cases/orchestration/check_identity.json index 020286c84e..8ba0ca5e42 100644 --- a/benchmarks/duckshop/edge_cases/orchestration/check_identity.json +++ b/benchmarks/duckshop/edge_cases/orchestration/check_identity.json @@ -1,14 +1,59 @@ { "audit_name": "not_null", "checks": [ - {"binding_key": "6f1c0e2a-orders-order-id", "attached_target_name": "orders", "attached_column_name": "order_id"}, - {"binding_key": "0b9d4c71-customers-customer-id", "attached_target_name": "customers", "attached_column_name": "customer_id"}, - {"binding_key": "9a3e5d10-archived-orders-order-id", "attached_target_name": "archived_orders", "attached_column_name": "order_id"} + { + "binding_key": "6f1c0e2a-orders-order-id", + "attached_target_name": "orders", + "attached_column_name": "order_id" + }, + { + "binding_key": "0b9d4c71-customers-customer-id", + "attached_target_name": "customers", + "attached_column_name": "customer_id" + }, + { + "binding_key": "9a3e5d10-archived-orders-order-id", + "attached_target_name": "archived_orders", + "attached_column_name": "order_id" + } ], "results": [ - {"check": 0, "result_id": "orders-not-null-1", "occurred_at": "2026-02-03T04:05:06Z", "outcome": "fail", "violation_count": 3, "executed_sql": "SELECT order_id FROM orders WHERE order_id IS NULL /* first 'run' 🧾 */"}, - {"check": 0, "result_id": "orders-not-null-2", "occurred_at": "2026-02-04T04:05:06Z", "outcome": "pass", "violation_count": 0, "executed_sql": "SELECT order_id FROM orders WHERE order_id IS NULL /* second 'run' 🧾 */"}, - {"check": 1, "result_id": "customers-not-null-1", "occurred_at": "2026-02-03T05:05:06Z", "outcome": "error", "violation_count": null, "executed_sql": "SELECT customer_id FROM customers WHERE customer_id IS NULL", "execution_error": "Binder Error: Referenced column \"customer_id\" not found\n\nLINE 1: SELECT customer_id FROM customers WHERE email = 'private@example.com'"}, - {"check": 2, "result_id": "archived-orders-not-null-1", "occurred_at": "2026-02-02T04:05:06Z", "outcome": "pass", "violation_count": 0, "executed_sql": "SELECT order_id FROM archived_orders WHERE order_id IS NULL"} + { + "check": 0, + "result_id": "orders-not-null-1", + "occurred_at": "2026-02-03T04:05:06Z", + "outcome": "fail", + "violation_count": 3, + "executed_sql": "SELECT order_id FROM orders WHERE order_id IS NULL /* first 'run' 🧾 */", + "evidence_json": "[{\"amount\":1.5,\"order_id\":\"o-1\"},{\"is_late\":true,\"order_id\":\"o-2\"}]", + "evidence_count": 2, + "evidence_truncated": "false" + }, + { + "check": 0, + "result_id": "orders-not-null-2", + "occurred_at": "2026-02-04T04:05:06Z", + "outcome": "pass", + "violation_count": 0, + "executed_sql": "SELECT order_id FROM orders WHERE order_id IS NULL /* second 'run' 🧾 */" + }, + { + "check": 1, + "result_id": "customers-not-null-1", + "occurred_at": "2026-02-03T05:05:06Z", + "outcome": "error", + "violation_count": null, + "executed_sql": "SELECT customer_id FROM customers WHERE customer_id IS NULL", + "execution_error": "Binder Error: Referenced column \"customer_id\" not found\n\nLINE 1: SELECT customer_id FROM customers WHERE email = 'private@example.com'" + }, + { + "check": 2, + "result_id": "archived-orders-not-null-1", + "occurred_at": "2026-02-02T04:05:06Z", + "outcome": "pass", + "violation_count": 0, + "executed_sql": "SELECT order_id FROM archived_orders WHERE order_id IS NULL", + "evidence_json": "[\"o-9 malformed\"]" + } ] } diff --git a/clients/typescript/src/schema.d.ts b/clients/typescript/src/schema.d.ts index 1e5cde2100..c2f2ab1fdb 100644 --- a/clients/typescript/src/schema.d.ts +++ b/clients/typescript/src/schema.d.ts @@ -242,6 +242,22 @@ export interface paths { patch?: never; trace?: never; }; + "/api/control/locations/{location_id}/targets/{target}/checks/{binding_key}/results/{event_id}/evidence": { + parameters: { + query?: never; + header?: never; + path?: never; + cookie?: never; + }; + get: operations["readPersonCheckEvidence"]; + put?: never; + post?: never; + delete?: never; + options?: never; + head?: never; + patch?: never; + trace?: never; + }; "/api/control/locations/{location_id}/targets/{target}/checks/{binding_key}/sql": { parameters: { query?: never; @@ -817,6 +833,17 @@ export interface components { ChangeNotice: { topic: string; }; + /** @description One result's stored evidence rows; `view.evidence` reads them, audited. */ + CheckEvidence: { + binding_key: string; + /** @description Every row key, in sorted order; the warehouse stores rows with sorted keys. */ + columns: string[]; + /** Format: uuid */ + event_id: string; + /** @description Scalar cells aligned to `columns`; a row without a key holds null there. */ + rows: Record[][]; + truncated?: string | null; + }; /** @description One stored check result. Its SQL, error texts and evidence rows need their own permissions. */ CheckResult: { attached_column_name?: string | null; @@ -2301,6 +2328,45 @@ export interface operations { }; }; }; + readPersonCheckEvidence: { + parameters: { + query?: never; + header?: never; + path: { + location_id: string; + target: string; + binding_key: string; + event_id: string; + }; + cookie?: never; + }; + requestBody?: never; + responses: { + 200: { + headers: { + [name: string]: unknown; + }; + content: { + "application/json": null | components["schemas"]["CheckEvidence"]; + }; + }; + /** @description Session rejected */ + 401: { + headers: { + [name: string]: unknown; + }; + content?: never; + }; + 403: { + headers: { + [name: string]: unknown; + }; + content: { + "application/json": components["schemas"]["ApiErrorBody"]; + }; + }; + }; + }; readPersonCheckSql: { parameters: { query?: never; 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 f876dd1988..171a2401d1 100644 --- a/crates/sqlbuild-orchestration/src/api/_helpers/authentication/password_sessions.rs +++ b/crates/sqlbuild-orchestration/src/api/_helpers/authentication/password_sessions.rs @@ -121,6 +121,10 @@ pub(crate) fn person_routes() -> Router { "/api/control/locations/{location_id}/targets/{target}/checks/{binding_key}/sql", get(crate::api::_helpers::resources::person_checks::sql), ) + .route( + "/api/control/locations/{location_id}/targets/{target}/checks/{binding_key}/results/{event_id}/evidence", + get(crate::api::_helpers::resources::person_checks::evidence), + ) .layer(DefaultBodyLimit::max(PASSWORD_BODY_BYTES)) } diff --git a/crates/sqlbuild-orchestration/src/api/_helpers/resources/person_checks.rs b/crates/sqlbuild-orchestration/src/api/_helpers/resources/person_checks.rs index 14303e2b89..8f210a579c 100644 --- a/crates/sqlbuild-orchestration/src/api/_helpers/resources/person_checks.rs +++ b/crates/sqlbuild-orchestration/src/api/_helpers/resources/person_checks.rs @@ -1,17 +1,18 @@ -//! Check reads use live person grants; SQL needs its own permission and is audited. +//! Check reads use live person grants; SQL and evidence need their own permissions and are audited. use axum::extract::{Path, Query, State}; use axum::{Extension, Json}; use uuid::Uuid; use crate::api::errors::ApiError; -use crate::api::models::ReadQuery; +use crate::api::models::{CheckResultPath, ReadQuery}; use crate::credentials::models::{AuditContext, AuthenticatedPerson, DefaultRole}; use crate::store::models::ControlStore; use crate::warehouse_metadata::main::read_person_check::read_person_check; +use crate::warehouse_metadata::main::read_person_check_evidence::read_person_check_evidence; use crate::warehouse_metadata::main::read_person_check_sql::read_person_check_sql; use crate::warehouse_metadata::main::read_person_checks::read_person_checks; -use crate::warehouse_metadata::models::{CheckResult, CheckSql}; +use crate::warehouse_metadata::models::{CheckEvidence, CheckResult, CheckSql}; #[utoipa::path(get, path = "/api/control/locations/{location_id}/targets/{target}/checks", operation_id = "listPersonChecks", params(("location_id" = Uuid, Path), ("target" = String, Path)), @@ -79,3 +80,23 @@ pub(crate) async fn sql( .await?, )) } + +#[utoipa::path(get, path = "/api/control/locations/{location_id}/targets/{target}/checks/{binding_key}/results/{event_id}/evidence", operation_id = "readPersonCheckEvidence", + params(("location_id" = Uuid, Path), ("target" = String, Path), ("binding_key" = String, Path), ("event_id" = Uuid, Path)), + responses((status = 200, body = Option), (status = 401, description = "Session rejected"), + (status = 403, body = crate::api::models::ApiErrorBody)), security(("session" = []), ("cliSession" = [])))] +pub(crate) async fn evidence( + (State(store), Extension(context)): (State, Extension), + Path(path): Path, + Extension(AuthenticatedPerson(principal, _)): Extension, +) -> Result>, ApiError> { + Ok(Json( + read_person_check_evidence( + &store, + (&context, principal), + (path.location_id, &path.target, &path.binding_key), + path.event_id, + ) + .await?, + )) +} diff --git a/crates/sqlbuild-orchestration/src/api/models.rs b/crates/sqlbuild-orchestration/src/api/models.rs index 4b125b3979..16288de89a 100644 --- a/crates/sqlbuild-orchestration/src/api/models.rs +++ b/crates/sqlbuild-orchestration/src/api/models.rs @@ -27,6 +27,7 @@ use crate::api::_helpers::resources::{ person_checks::list, person_checks::history, person_checks::sql, + person_checks::evidence, person_location_work::request, person_location_work::result, person_node_sql::read, @@ -86,6 +87,14 @@ pub(crate) struct RunRecallQuery { pub(crate) limit: Option, } +#[derive(Debug, Deserialize)] +pub(crate) struct CheckResultPath { + pub(crate) location_id: uuid::Uuid, + pub(crate) target: String, + pub(crate) binding_key: String, + pub(crate) event_id: uuid::Uuid, +} + #[derive(Debug, Deserialize)] pub(crate) struct RunRecallPath { pub(crate) location_id: uuid::Uuid, diff --git a/crates/sqlbuild-orchestration/src/warehouse_metadata/_helpers/checks.rs b/crates/sqlbuild-orchestration/src/warehouse_metadata/_helpers/checks.rs index a15bd54010..34340e5fd7 100644 --- a/crates/sqlbuild-orchestration/src/warehouse_metadata/_helpers/checks.rs +++ b/crates/sqlbuild-orchestration/src/warehouse_metadata/_helpers/checks.rs @@ -3,13 +3,16 @@ use std::collections::BTreeSet; use serde::Deserialize; +use serde_json::json; use uuid::Uuid; -use crate::credentials::models::{Action, PermissionScope}; +use crate::authorization::main::authorize_person::authorize_person; +use crate::credentials::main::audit::append_audit::append_audit; +use crate::credentials::models::{Action, AuditContext, DefaultRole, PermissionScope}; use crate::store::errors::ControlError; -use crate::store::main::{query::query, query_records::query_records}; +use crate::store::main::{commit::commit, query::query, query_records::query_records}; use crate::store::models::{ControlTransaction, Parameter}; -use crate::warehouse_metadata::models::WarehouseAuditRecord; +use crate::warehouse_metadata::models::{CheckEvidence, WarehouseAuditRecord}; #[derive(Deserialize)] pub(crate) struct AuditRow { @@ -31,6 +34,62 @@ pub(crate) async fn latest_results( vec![Parameter::Id(location_id), Parameter::Text(target.into())]).await } +/// One result of one check. +pub(crate) async fn check_result( + transaction: ControlTransaction, + (location_id, target, binding_key): (Uuid, &str, &str), + event_id: Uuid, +) -> Result<(ControlTransaction, Option), ControlError> { + let (transaction, mut rows) = query_records( + transaction, + "SELECT event_id,record_json AS record FROM sqb_warehouse_audit_results \ + WHERE location_id=$1 AND target=$2 AND binding_key=$3 AND event_id=$4", + vec![ + Parameter::Id(location_id), + Parameter::Text(target.into()), + Parameter::Text(binding_key.into()), + Parameter::Id(event_id), + ], + ) + .await?; + Ok((transaction, rows.pop())) +} + +/// Stored evidence rows as columns in sorted key order and aligned cells. +pub(crate) fn evidence(row: AuditRow) -> Result { + let objects: Vec> = row + .record + .evidence_json + .as_deref() + .map(serde_json::from_str) + .transpose() + .map_err(|_| { + ControlError::Schema("stored check evidence is not an array of objects".into()) + })? + .unwrap_or_default(); + let columns: Vec = objects + .iter() + .flat_map(|object| object.keys().cloned()) + .collect::>() + .into_iter() + .collect(); + let mut rows: Vec> = Vec::with_capacity(objects.len()); + for object in &objects { + let mut cells: Vec = Vec::with_capacity(columns.len()); + for column in &columns { + cells.push(object.get(column).cloned().unwrap_or_default()); + } + rows.push(cells); + } + Ok(CheckEvidence { + event_id: row.event_id, + binding_key: row.record.binding_key, + columns, + rows, + truncated: row.record.evidence_truncated, + }) +} + /// One check's results, newest first. pub(crate) async fn check_results( transaction: ControlTransaction, @@ -44,8 +103,39 @@ pub(crate) async fn check_results( Parameter::Integer(i64::from(limit.clamp(1, 1000)))]).await } +/// Authorizes `action` on the row's check, projects the row, and audits the read before commit. +pub(crate) async fn sensitive_check_read( + (transaction, credentials): (ControlTransaction, (&AuditContext, Uuid)), + (location_id, target, binding_key): (Uuid, &str, &str), + (row, action, audit_action): (Option, Action, &str), + project: impl FnOnce(AuditRow) -> Result, +) -> Result, ControlError> { + let attached = row + .as_ref() + .and_then(|row| row.record.attached_target_name.as_deref()); + let (transaction, scope) = + check_scope(transaction, (location_id, target, attached), action).await?; + let (transaction, context) = + authorize_person(transaction, credentials, &scope, DefaultRole::None).await?; + let event_id = row.as_ref().map(|row| row.event_id); + let result = row.map(project).transpose()?; + let transaction = append_audit( + transaction, + &context, + (Uuid::now_v7(), audit_action), + ( + json!(scope), + json!(null), + json!({"binding_key": binding_key, "event_id": event_id, "found": result.is_some()}), + ), + ) + .await?; + commit(transaction).await?; + Ok(result) +} + /// The check's own node when the current snapshot has it, otherwise every current node. -pub(crate) async fn check_scope( +async fn check_scope( transaction: ControlTransaction, (location_id, target, attached): (Uuid, &str, Option<&str>), action: Action, diff --git a/crates/sqlbuild-orchestration/src/warehouse_metadata/main/mod.rs b/crates/sqlbuild-orchestration/src/warehouse_metadata/main/mod.rs index ab5d35815c..722d6939c3 100644 --- a/crates/sqlbuild-orchestration/src/warehouse_metadata/main/mod.rs +++ b/crates/sqlbuild-orchestration/src/warehouse_metadata/main/mod.rs @@ -10,6 +10,7 @@ pub mod ingest_statistics; pub(crate) mod prune_statistics; pub mod read_person_catalog; pub mod read_person_check; +pub mod read_person_check_evidence; pub mod read_person_check_sql; pub mod read_person_checks; pub mod read_person_fingerprint; diff --git a/crates/sqlbuild-orchestration/src/warehouse_metadata/main/read_person_check_evidence.rs b/crates/sqlbuild-orchestration/src/warehouse_metadata/main/read_person_check_evidence.rs new file mode 100644 index 0000000000..8132c3e6f5 --- /dev/null +++ b/crates/sqlbuild-orchestration/src/warehouse_metadata/main/read_person_check_evidence.rs @@ -0,0 +1,27 @@ +//! Live evidence permission on the check's node, one result's rows and the sensitive-read audit share one writer fence. + +use uuid::Uuid; + +use crate::credentials::models::{Action, AuditContext}; +use crate::store::errors::ControlError; +use crate::store::main::begin::begin; +use crate::store::models::{ControlStore, TransactionMode}; +use crate::warehouse_metadata::_helpers::checks::{check_result, evidence, sensitive_check_read}; +use crate::warehouse_metadata::models::CheckEvidence; + +pub async fn read_person_check_evidence( + store: &ControlStore, + credentials: (&AuditContext, Uuid), + check: (Uuid, &str, &str), + event_id: Uuid, +) -> Result, ControlError> { + let (transaction, row) = + check_result(begin(store, TransactionMode::Write).await?, check, event_id).await?; + sensitive_check_read( + (transaction, credentials), + check, + (row, Action::ViewEvidence, "evidence.read"), + evidence, + ) + .await +} diff --git a/crates/sqlbuild-orchestration/src/warehouse_metadata/main/read_person_check_sql.rs b/crates/sqlbuild-orchestration/src/warehouse_metadata/main/read_person_check_sql.rs index 399e66df51..6e615cd52b 100644 --- a/crates/sqlbuild-orchestration/src/warehouse_metadata/main/read_person_check_sql.rs +++ b/crates/sqlbuild-orchestration/src/warehouse_metadata/main/read_person_check_sql.rs @@ -1,60 +1,36 @@ //! Live SQL permission on the check's node, its latest SQL and errors, and the sensitive-read audit share one writer fence. -use serde_json::json; use uuid::Uuid; -use crate::authorization::main::authorize_person::authorize_person; -use crate::credentials::main::audit::append_audit::append_audit; -use crate::credentials::models::{Action, AuditContext, DefaultRole}; +use crate::credentials::models::{Action, AuditContext}; use crate::store::errors::ControlError; -use crate::store::main::{begin::begin, commit::commit}; +use crate::store::main::begin::begin; use crate::store::models::{ControlStore, TransactionMode}; -use crate::warehouse_metadata::_helpers::checks::{check_results, check_scope}; +use crate::warehouse_metadata::_helpers::checks::{check_results, sensitive_check_read}; use crate::warehouse_metadata::models::CheckSql; pub async fn read_person_check_sql( store: &ControlStore, credentials: (&AuditContext, Uuid), - (location_id, target, binding_key): (Uuid, &str, &str), + check: (Uuid, &str, &str), ) -> Result, ControlError> { - let (transaction, mut rows) = check_results( - begin(store, TransactionMode::Write).await?, - (location_id, target, binding_key), - 1, + let (transaction, mut rows) = + check_results(begin(store, TransactionMode::Write).await?, check, 1).await?; + sensitive_check_read( + (transaction, credentials), + check, + (rows.pop(), Action::ViewSql, "sql.read"), + |row| { + Ok(CheckSql { + event_id: row.event_id, + binding_key: row.record.binding_key, + executed_sql: row.record.executed_sql, + measurement_sql: row.record.measurement_sql, + evidence_sql: row.record.evidence_sql, + execution_error: row.record.execution_error, + evidence_error: row.record.evidence_error, + }) + }, ) - .await?; - let latest = rows.pop(); - let attached = latest - .as_ref() - .and_then(|row| row.record.attached_target_name.as_deref()); - let (transaction, scope) = check_scope( - transaction, - (location_id, target, attached), - Action::ViewSql, - ) - .await?; - let (transaction, context) = - authorize_person(transaction, credentials, &scope, DefaultRole::None).await?; - let result = latest.map(|row| CheckSql { - event_id: row.event_id, - binding_key: row.record.binding_key, - executed_sql: row.record.executed_sql, - measurement_sql: row.record.measurement_sql, - evidence_sql: row.record.evidence_sql, - execution_error: row.record.execution_error, - evidence_error: row.record.evidence_error, - }); - let transaction = append_audit( - transaction, - &context, - (Uuid::now_v7(), "sql.read"), - ( - json!(scope), - json!(null), - json!({"binding_key": binding_key, "event_id": result.as_ref().map(|sql| sql.event_id), "found": result.is_some()}), - ), - ) - .await?; - commit(transaction).await?; - Ok(result) + .await } diff --git a/crates/sqlbuild-orchestration/src/warehouse_metadata/models.rs b/crates/sqlbuild-orchestration/src/warehouse_metadata/models.rs index 35ae53abcb..31c94dcb87 100644 --- a/crates/sqlbuild-orchestration/src/warehouse_metadata/models.rs +++ b/crates/sqlbuild-orchestration/src/warehouse_metadata/models.rs @@ -149,6 +149,19 @@ pub struct CheckSql { pub evidence_error: Option, } +/// One result's stored evidence rows; `view.evidence` reads them, audited. +#[derive(Clone, Debug, PartialEq, Serialize, Deserialize, utoipa::ToSchema)] +pub struct CheckEvidence { + pub event_id: Uuid, + pub binding_key: String, + /// Every row key, in sorted order; the warehouse stores rows with sorted keys. + pub columns: Vec, + /// Scalar cells aligned to `columns`; a row without a key holds null there. + #[schema(value_type = Vec>)] + pub rows: Vec>, + pub truncated: Option, +} + /// Warehouse cleanup facts do not authorize control-plane deletion or worker recovery. #[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize, utoipa::ToSchema)] #[serde(deny_unknown_fields)] diff --git a/crates/sqlbuild-orchestration/tests/api/test_checks.rs b/crates/sqlbuild-orchestration/tests/api/test_checks.rs index 3fb234a575..24e1fc9acc 100644 --- a/crates/sqlbuild-orchestration/tests/api/test_checks.rs +++ b/crates/sqlbuild-orchestration/tests/api/test_checks.rs @@ -24,6 +24,8 @@ use crate::platform::models::{LocationHealth, LocationHealthReport}; use crate::runs::tests::helpers::snapshot; use crate::snapshots::main::publish_snapshot::publish_snapshot; use crate::store::main::migrate_store::migrate_store; +use crate::store::tests::fault_helpers::{install_fault, remove_fault}; +use crate::store::tests::fault_types::TransactionFault; use crate::store::tests::store_helpers::{close_test_store, open_test_store, run}; use crate::store::tests::store_test_types::Backend; use crate::warehouse_metadata::main::ingest_audit_results::ingest_audit_results; @@ -164,6 +166,7 @@ fn given_checks_sharing_an_audit_name_when_people_read_them_then_binding_keys_an let client = Client::new(); let viewer = login(&client, &api.url, "orders-viewer").await; let developer = login(&client, &api.url, "orders-developer").await; + let administrator = login(&client, &api.url, "orders-admin").await; let base_path = format!("/api/control/locations/{location}/targets/development/checks"); let orders_key = data["checks"][0]["binding_key"].as_str().expect("key"); let customers_key = data["checks"][1]["binding_key"].as_str().expect("key"); @@ -299,6 +302,88 @@ fn given_checks_sharing_an_audit_name_when_people_read_them_then_binding_keys_an StatusCode::FORBIDDEN ); } + let failed_id = history[1]["event_id"].as_str().expect("event"); + let evidence_path = format!("{base_path}/{orders_key}/results/{failed_id}/evidence"); + let (status, _) = get(&client, (&api.url, &developer), &evidence_path).await; + assert_eq!( + status, + StatusCode::FORBIDDEN, + "developers lack view.evidence" + ); + let denial = latest_audit(store).await; + assert_eq!(denial["scope_json"]["action"], "view.evidence"); + assert_eq!(denial["scope_json"]["nodes"], json!(["orders"])); + let (status, evidence) = get(&client, (&api.url, &administrator), &evidence_path).await; + assert_eq!(status, StatusCode::OK); + assert_eq!( + evidence, + json!({"event_id": failed_id, "binding_key": orders_key, + "columns": ["amount", "is_late", "order_id"], + "rows": [[1.5, null, "o-1"], [null, true, "o-2"]], "truncated": "false"}), + "sorted keys become columns; missing keys become null cells" + ); + let read = latest_audit(store).await; + assert_eq!(read["action"], "evidence.read"); + assert_eq!( + read["after_json"], + json!({"binding_key": orders_key, "event_id": failed_id, "found": true}) + ); + assert!( + !read.to_string().contains("o-1"), + "the audit never holds evidence" + ); + let (status, mismatched) = get( + &client, + (&api.url, &administrator), + &evidence_path.replace(orders_key, customers_key), + ) + .await; + assert_eq!( + (status, mismatched), + (StatusCode::OK, Value::Null), + "a result is read only under its own check" + ); + let (_, archived) = get( + &client, + (&api.url, &administrator), + &format!("{base_path}/{archived_key}"), + ) + .await; + let malformed_path = format!( + "{base_path}/{archived_key}/results/{}/evidence", + archived[0]["event_id"].as_str().expect("event") + ); + let (status, refusal) = get(&client, (&api.url, &administrator), &malformed_path).await; + assert_eq!( + (status, refusal), + ( + StatusCode::INTERNAL_SERVER_ERROR, + json!({"error": "control-store operation failed"}) + ), + "malformed stored evidence is never echoed" + ); + let fault = TransactionFault { + description: "evidence read requires durable audit", + table: "sqb_security_audit", + operation: "INSERT", + condition: "WHEN NEW.action='evidence.read'", + sqlite_body: "SELECT RAISE(ABORT, 'synthetic evidence audit failure')", + postgres_body: "RAISE EXCEPTION 'synthetic evidence audit failure'", + }; + install_fault(store, &fault).await; + let (status, refusal) = get(&client, (&api.url, &administrator), &evidence_path).await; + assert_eq!( + status, + StatusCode::INTERNAL_SERVER_ERROR, + "{}", + fault.description + ); + assert!( + !refusal.to_string().contains("o-1"), + "no evidence without its audit" + ); + remove_fault(store, &fault).await; + stop_api(api).await; close_test_store(fixture_store).await; }); diff --git a/fensu.toml b/fensu.toml index e67c674104..146aa9b37e 100644 --- a/fensu.toml +++ b/fensu.toml @@ -1641,7 +1641,6 @@ construction_required = [ "Action=A permission is live only when a route requires it; role tables and classifiers do not count.", ] exceptions = [ - "Action::ViewEvidence=Kept under D129; reader comes with the UI epic.", "InaccessibleReason=Wire enum that serde builds from Python scope records.", "ScopeDiagnosticCode=Wire enum that serde builds from Python scope diagnostics.", "TickOutcome::Covered=The scheduler writes it as SQL text and serde reads it back.", From 81a503ea08dc7c7c5b481118f4b51e95b43cb3b3 Mon Sep 17 00:00:00 2001 From: Kevin Longe <273747162+kvlonge@users.noreply.github.com> Date: Fri, 6 Nov 2026 03:34:30 +0000 Subject: [PATCH 2/4] feat: read maintenance history through the control API People with view on a target can read its recorded model moves, every step of their old-name views, and janitor cleanups, newest moves first. Filtering by model returns only moves into that model and their views, using new indexes on the destination model and the move's event. Check and maintenance reads now live in their own warehouse-history domain, apart from metadata ingestion. --- .../orchestration/maintenance_history.json | 81 +++++++ clients/typescript/src/schema.d.ts | 135 +++++++++++ .../authentication/password_sessions.rs | 4 + .../src/api/_helpers/resources/mod.rs | 1 + .../api/_helpers/resources/person_checks.rs | 10 +- .../_helpers/resources/person_maintenance.rs | 34 +++ .../sqlbuild-orchestration/src/api/models.rs | 5 +- crates/sqlbuild-orchestration/src/lib.rs | 1 + .../src/store/schema.sql | 2 + .../_helpers/checks.rs | 3 +- .../warehouse_history/_helpers/maintenance.rs | 86 +++++++ .../src/warehouse_history/_helpers/mod.rs | 2 + .../src/warehouse_history/main/mod.rs | 5 + .../main/read_person_check.rs | 4 +- .../main/read_person_check_evidence.rs | 4 +- .../main/read_person_check_sql.rs | 4 +- .../main/read_person_checks.rs | 4 +- .../main/read_person_maintenance.rs | 34 +++ .../src/warehouse_history/mod.rs | 5 + .../src/warehouse_history/models.rs | 114 ++++++++++ .../src/warehouse_metadata/_helpers/mod.rs | 1 - .../src/warehouse_metadata/main/mod.rs | 4 - .../src/warehouse_metadata/models.rs | 88 ------- .../tests/api/checks/helpers.rs | 68 ------ .../tests/api/test_checks.rs | 122 ++-------- .../tests/api/test_contracts.rs | 6 +- .../tests/api/test_maintenance.rs | 162 +++++++++++++ .../tests/api/warehouse_history/helpers.rs | 214 ++++++++++++++++++ fensu.toml | 6 +- 29 files changed, 918 insertions(+), 291 deletions(-) create mode 100644 benchmarks/duckshop/edge_cases/orchestration/maintenance_history.json create mode 100644 crates/sqlbuild-orchestration/src/api/_helpers/resources/person_maintenance.rs rename crates/sqlbuild-orchestration/src/{warehouse_metadata => warehouse_history}/_helpers/checks.rs (98%) create mode 100644 crates/sqlbuild-orchestration/src/warehouse_history/_helpers/maintenance.rs create mode 100644 crates/sqlbuild-orchestration/src/warehouse_history/_helpers/mod.rs create mode 100644 crates/sqlbuild-orchestration/src/warehouse_history/main/mod.rs rename crates/sqlbuild-orchestration/src/{warehouse_metadata => warehouse_history}/main/read_person_check.rs (90%) rename crates/sqlbuild-orchestration/src/{warehouse_metadata => warehouse_history}/main/read_person_check_evidence.rs (84%) rename crates/sqlbuild-orchestration/src/{warehouse_metadata => warehouse_history}/main/read_person_check_sql.rs (90%) rename crates/sqlbuild-orchestration/src/{warehouse_metadata => warehouse_history}/main/read_person_checks.rs (89%) create mode 100644 crates/sqlbuild-orchestration/src/warehouse_history/main/read_person_maintenance.rs create mode 100644 crates/sqlbuild-orchestration/src/warehouse_history/mod.rs create mode 100644 crates/sqlbuild-orchestration/src/warehouse_history/models.rs delete mode 100644 crates/sqlbuild-orchestration/tests/api/checks/helpers.rs create mode 100644 crates/sqlbuild-orchestration/tests/api/test_maintenance.rs create mode 100644 crates/sqlbuild-orchestration/tests/api/warehouse_history/helpers.rs diff --git a/benchmarks/duckshop/edge_cases/orchestration/maintenance_history.json b/benchmarks/duckshop/edge_cases/orchestration/maintenance_history.json new file mode 100644 index 0000000000..860b31bec6 --- /dev/null +++ b/benchmarks/duckshop/edge_cases/orchestration/maintenance_history.json @@ -0,0 +1,81 @@ +{ + "migrations": [ + { + "event_id": "move-orders-1", + "destination_model": "orders", + "origin_model": "orders_v1", + "origin_name": "orders_v1", + "decision": "migrate", + "created_at": "2026-02-01T04:05:06Z" + }, + { + "event_id": "move-orders-2", + "destination_model": "orders", + "origin_model": null, + "origin_name": "orders_🧾", + "decision": "renamed", + "discovery": "automatic", + "created_at": "2026-02-03T04:05:06Z" + }, + { + "event_id": "move-customers-1", + "destination_model": "customers", + "origin_model": "clients", + "origin_name": "clients", + "decision": "superseded_replace", + "created_at": "2026-02-02T04:05:06Z" + } + ], + "old_name_views": [ + { + "event_id": "view-orders-1", + "migration_event_id": "move-orders-2", + "destination_model": "orders", + "event_type": "required", + "created_at": "2026-02-03T04:05:06Z" + }, + { + "event_id": "view-orders-2", + "migration_event_id": "move-orders-2", + "destination_model": "orders", + "event_type": "view_created", + "created_at": "2026-02-03T04:05:07Z", + "column_aliases": "{\"order_id\":\"legacy_order_id\"}" + }, + { + "event_id": "view-customers-1", + "migration_event_id": "move-customers-1", + "destination_model": "customers", + "event_type": "required", + "created_at": "2026-02-02T04:05:06Z" + }, + { + "event_id": "view-orders-old-1", + "migration_event_id": "move-orders-1", + "destination_model": "orders", + "event_type": "required", + "created_at": "2026-02-01T04:05:06Z" + } + ], + "janitor_events": [ + { + "event_id": "cleanup-1", + "event_type": "archive", + "occurred_at": "2026-02-01T05:05:06Z" + }, + { + "event_id": "cleanup-2", + "event_type": "delete", + "occurred_at": "2026-02-04T05:05:06Z" + } + ], + "preview_old_name_views": [ + { + "event_id": "view-preview-1", + "migration_event_id": "move-orders-2", + "destination_model": "orders", + "event_type": "view_dropped", + "created_at": "2026-02-05T04:05:06Z" + } + ] +} diff --git a/clients/typescript/src/schema.d.ts b/clients/typescript/src/schema.d.ts index c2f2ab1fdb..ae6e30de48 100644 --- a/clients/typescript/src/schema.d.ts +++ b/clients/typescript/src/schema.d.ts @@ -322,6 +322,22 @@ export interface paths { patch?: never; trace?: never; }; + "/api/control/locations/{location_id}/targets/{target}/maintenance": { + parameters: { + query?: never; + header?: never; + path?: never; + cookie?: never; + }; + get: operations["readPersonMaintenance"]; + put?: never; + post?: never; + delete?: never; + options?: never; + head?: never; + patch?: never; + trace?: never; + }; "/api/control/locations/{location_id}/targets/{target}/node-status": { parameters: { query?: never; @@ -1121,6 +1137,14 @@ export interface components { /** Format: uuid */ session_id: string; }; + /** @description Recorded moves, their old-name view steps and janitor cleanups, newest moves first. */ + MaintenanceHistory: { + /** @description Empty when the history is for one model; cleanups are not tied to models. */ + janitor_events: components["schemas"]["WarehouseJanitorRecord"][]; + migrations: components["schemas"]["WarehouseMigrationRecord"][]; + /** @description Every step recorded for the listed moves, oldest first; same-time steps are not in step order. */ + old_name_views: components["schemas"]["WarehouseOldNameRecord"][]; + }; /** @description One bounded page of one state table from the registered reader of a published generation. */ MetadataBatch: { /** Format: int64 */ @@ -1731,12 +1755,81 @@ export interface components { * @enum {string} */ WarehouseHistoryKind: "fingerprints" | "source_freshness" | "audit_results" | "janitor_events" | "migrations" | "old_name_views" | "microbatches" | "catalog" | "statistics"; + /** @description Warehouse cleanup facts do not authorize control-plane deletion or worker recovery. */ + WarehouseJanitorRecord: { + archive_name: string; + archive_qualified_name: string; + /** Format: date-time */ + archived_at: string; + event_id: string; + event_type: string; + /** Format: date-time */ + occurred_at: string; + original_name?: string | null; + original_qualified_name?: string | null; + relation_database?: string | null; + relation_schema: string; + relation_type?: string | null; + run_id: string; + /** Format: int64 */ + schema_version: number; + state_database?: string | null; + state_schema: string; + }; + /** @description Model-move history is evidence, not authority to migrate a warehouse relation. */ + WarehouseMigrationRecord: { + /** Format: date-time */ + created_at: string; + decision: string; + destination_database?: string | null; + destination_model: string; + destination_name: string; + destination_schema: string; + discovery: string; + event_id: string; + origin_database?: string | null; + origin_model?: string | null; + origin_name: string; + origin_schema?: string | null; + origin_version_hash: string; + run_id: string; + state_database?: string | null; + state_schema: string; + target_name?: string | null; + }; WarehouseMirrorCursor: { /** Format: uuid */ event_id: string; /** Format: date-time */ observed_at: string; }; + /** @description Preserve nullable compatibility-view facts without projecting live warehouse existence. */ + WarehouseOldNameRecord: { + archive_name?: string | null; + column_aliases?: string | null; + /** Format: date-time */ + created_at: string; + destination_model: string; + drop_reason?: string | null; + event_id: string; + event_type: string; + /** Format: date-time */ + expires_at?: string | null; + grants_copied?: string | null; + migration_event_id: string; + new_database?: string | null; + new_name: string; + new_schema: string; + old_database?: string | null; + old_name: string; + old_schema?: string | null; + run_id: string; + state_database?: string | null; + state_schema: string; + target_name?: string | null; + view_retention?: string | null; + view_sql?: string | null; + }; /** @description Physical namespace of SQLBuild's source-freshness state table, never a source query. */ WarehouseStateLocation: { database?: string | null; @@ -2547,6 +2640,48 @@ export interface operations { }; }; }; + readPersonMaintenance: { + parameters: { + query?: { + /** @description Only moves into this model, and no janitor events */ + node?: string; + /** @description Maximum moves and janitor events each; defaults to 100, capped at 1000 */ + limit?: number; + }; + header?: never; + path: { + location_id: string; + target: string; + }; + cookie?: never; + }; + requestBody?: never; + responses: { + 200: { + headers: { + [name: string]: unknown; + }; + content: { + "application/json": components["schemas"]["MaintenanceHistory"]; + }; + }; + /** @description Session rejected */ + 401: { + headers: { + [name: string]: unknown; + }; + content?: never; + }; + 403: { + headers: { + [name: string]: unknown; + }; + content: { + "application/json": components["schemas"]["ApiErrorBody"]; + }; + }; + }; + }; readPersonNodeStatus: { parameters: { query: { 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 171a2401d1..0a955d6235 100644 --- a/crates/sqlbuild-orchestration/src/api/_helpers/authentication/password_sessions.rs +++ b/crates/sqlbuild-orchestration/src/api/_helpers/authentication/password_sessions.rs @@ -125,6 +125,10 @@ pub(crate) fn person_routes() -> Router { "/api/control/locations/{location_id}/targets/{target}/checks/{binding_key}/results/{event_id}/evidence", get(crate::api::_helpers::resources::person_checks::evidence), ) + .route( + "/api/control/locations/{location_id}/targets/{target}/maintenance", + get(crate::api::_helpers::resources::person_maintenance::read), + ) .layer(DefaultBodyLimit::max(PASSWORD_BODY_BYTES)) } diff --git a/crates/sqlbuild-orchestration/src/api/_helpers/resources/mod.rs b/crates/sqlbuild-orchestration/src/api/_helpers/resources/mod.rs index 3b6efd68ac..be262332c1 100644 --- a/crates/sqlbuild-orchestration/src/api/_helpers/resources/mod.rs +++ b/crates/sqlbuild-orchestration/src/api/_helpers/resources/mod.rs @@ -7,6 +7,7 @@ pub(crate) mod location_health; pub(crate) mod location_work; pub(crate) mod person_checks; pub(crate) mod person_location_work; +pub(crate) mod person_maintenance; pub(crate) mod person_node_sql; pub(crate) mod person_operations; pub(crate) mod router; diff --git a/crates/sqlbuild-orchestration/src/api/_helpers/resources/person_checks.rs b/crates/sqlbuild-orchestration/src/api/_helpers/resources/person_checks.rs index 8f210a579c..8fbb6f4fb2 100644 --- a/crates/sqlbuild-orchestration/src/api/_helpers/resources/person_checks.rs +++ b/crates/sqlbuild-orchestration/src/api/_helpers/resources/person_checks.rs @@ -8,11 +8,11 @@ use crate::api::errors::ApiError; use crate::api::models::{CheckResultPath, ReadQuery}; use crate::credentials::models::{AuditContext, AuthenticatedPerson, DefaultRole}; use crate::store::models::ControlStore; -use crate::warehouse_metadata::main::read_person_check::read_person_check; -use crate::warehouse_metadata::main::read_person_check_evidence::read_person_check_evidence; -use crate::warehouse_metadata::main::read_person_check_sql::read_person_check_sql; -use crate::warehouse_metadata::main::read_person_checks::read_person_checks; -use crate::warehouse_metadata::models::{CheckEvidence, CheckResult, CheckSql}; +use crate::warehouse_history::main::read_person_check::read_person_check; +use crate::warehouse_history::main::read_person_check_evidence::read_person_check_evidence; +use crate::warehouse_history::main::read_person_check_sql::read_person_check_sql; +use crate::warehouse_history::main::read_person_checks::read_person_checks; +use crate::warehouse_history::models::{CheckEvidence, CheckResult, CheckSql}; #[utoipa::path(get, path = "/api/control/locations/{location_id}/targets/{target}/checks", operation_id = "listPersonChecks", params(("location_id" = Uuid, Path), ("target" = String, Path)), diff --git a/crates/sqlbuild-orchestration/src/api/_helpers/resources/person_maintenance.rs b/crates/sqlbuild-orchestration/src/api/_helpers/resources/person_maintenance.rs new file mode 100644 index 0000000000..1c704c0b52 --- /dev/null +++ b/crates/sqlbuild-orchestration/src/api/_helpers/resources/person_maintenance.rs @@ -0,0 +1,34 @@ +//! Maintenance history reads use live person grants over the whole target. + +use axum::extract::{Path, Query, State}; +use axum::{Extension, Json}; +use uuid::Uuid; + +use crate::api::errors::ApiError; +use crate::credentials::models::{AuditContext, AuthenticatedPerson, DefaultRole}; +use crate::store::models::ControlStore; +use crate::warehouse_history::main::read_person_maintenance::read_person_maintenance; +use crate::warehouse_history::models::{MaintenanceHistory, MaintenanceQuery}; + +#[utoipa::path(get, path = "/api/control/locations/{location_id}/targets/{target}/maintenance", operation_id = "readPersonMaintenance", + params(("location_id" = Uuid, Path), ("target" = String, Path), + ("node" = Option, Query, description = "Only moves into this model, and no janitor events"), + ("limit" = Option, Query, description = "Maximum moves and janitor events each; defaults to 100, capped at 1000")), + responses((status = 200, body = MaintenanceHistory), (status = 401, description = "Session rejected"), + (status = 403, body = crate::api::models::ApiErrorBody)), security(("session" = []), ("cliSession" = [])))] +pub(crate) async fn read( + (State(store), Extension(context)): (State, Extension), + Extension(default_role): Extension, + (Path((location_id, target)), Query(options)): (Path<(Uuid, String)>, Query), + Extension(AuthenticatedPerson(principal, _)): Extension, +) -> Result, ApiError> { + Ok(Json( + read_person_maintenance( + &store, + (&context, principal), + (location_id, &target, &options), + default_role, + ) + .await?, + )) +} diff --git a/crates/sqlbuild-orchestration/src/api/models.rs b/crates/sqlbuild-orchestration/src/api/models.rs index 16288de89a..0174bb33c6 100644 --- a/crates/sqlbuild-orchestration/src/api/models.rs +++ b/crates/sqlbuild-orchestration/src/api/models.rs @@ -6,8 +6,8 @@ use crate::api::_helpers::authentication::password_sessions; use crate::api::_helpers::authentication::{cli_login, oidc}; use crate::api::_helpers::resources::{ administrators, audit, changes, location_health, location_work, person_checks, - person_location_work, person_node_sql, person_operations, run_reads, runs, snapshots, - trigger_reads, warehouse_metadata, + person_location_work, person_maintenance, person_node_sql, person_operations, run_reads, runs, + snapshots, trigger_reads, warehouse_metadata, }; #[derive(Debug, utoipa::OpenApi)] @@ -28,6 +28,7 @@ use crate::api::_helpers::resources::{ person_checks::history, person_checks::sql, person_checks::evidence, + person_maintenance::read, person_location_work::request, person_location_work::result, person_node_sql::read, diff --git a/crates/sqlbuild-orchestration/src/lib.rs b/crates/sqlbuild-orchestration/src/lib.rs index 87aebeb15e..213b03ccff 100644 --- a/crates/sqlbuild-orchestration/src/lib.rs +++ b/crates/sqlbuild-orchestration/src/lib.rs @@ -12,6 +12,7 @@ pub mod scheduler; pub mod sensors; pub mod snapshots; pub mod store; +pub mod warehouse_history; pub mod warehouse_metadata; pub mod webhooks; pub mod worker; diff --git a/crates/sqlbuild-orchestration/src/store/schema.sql b/crates/sqlbuild-orchestration/src/store/schema.sql index 39a45c68a7..537f4ccdd4 100644 --- a/crates/sqlbuild-orchestration/src/store/schema.sql +++ b/crates/sqlbuild-orchestration/src/store/schema.sql @@ -429,6 +429,7 @@ CREATE TABLE sqb_warehouse_migrations ( scan_id {uuid} ); CREATE INDEX sqb_migrations_namespace ON sqb_warehouse_migrations (location_id,target,state_database,state_schema,observed_at,event_id); +CREATE INDEX sqb_migrations_destination ON sqb_warehouse_migrations (location_id,target,destination_model,observed_at,event_id); CREATE TABLE sqb_warehouse_old_name_views ( event_id {uuid} PRIMARY KEY, location_id {uuid} NOT NULL REFERENCES sqb_locations(location_id), @@ -444,6 +445,7 @@ CREATE TABLE sqb_warehouse_old_name_views ( scan_id {uuid} ); CREATE INDEX sqb_old_name_views_namespace ON sqb_warehouse_old_name_views (location_id,target,state_database,state_schema,observed_at,event_id); +CREATE INDEX sqb_old_name_views_migration ON sqb_warehouse_old_name_views (location_id,target,migration_event_id,observed_at,event_id); CREATE TABLE sqb_warehouse_microbatches ( event_id {uuid} PRIMARY KEY, location_id {uuid} NOT NULL REFERENCES sqb_locations(location_id), diff --git a/crates/sqlbuild-orchestration/src/warehouse_metadata/_helpers/checks.rs b/crates/sqlbuild-orchestration/src/warehouse_history/_helpers/checks.rs similarity index 98% rename from crates/sqlbuild-orchestration/src/warehouse_metadata/_helpers/checks.rs rename to crates/sqlbuild-orchestration/src/warehouse_history/_helpers/checks.rs index 34340e5fd7..d4a78fbad3 100644 --- a/crates/sqlbuild-orchestration/src/warehouse_metadata/_helpers/checks.rs +++ b/crates/sqlbuild-orchestration/src/warehouse_history/_helpers/checks.rs @@ -12,7 +12,8 @@ use crate::credentials::models::{Action, AuditContext, DefaultRole, PermissionSc use crate::store::errors::ControlError; use crate::store::main::{commit::commit, query::query, query_records::query_records}; use crate::store::models::{ControlTransaction, Parameter}; -use crate::warehouse_metadata::models::{CheckEvidence, WarehouseAuditRecord}; +use crate::warehouse_history::models::CheckEvidence; +use crate::warehouse_metadata::models::WarehouseAuditRecord; #[derive(Deserialize)] pub(crate) struct AuditRow { diff --git a/crates/sqlbuild-orchestration/src/warehouse_history/_helpers/maintenance.rs b/crates/sqlbuild-orchestration/src/warehouse_history/_helpers/maintenance.rs new file mode 100644 index 0000000000..279f7a16dc --- /dev/null +++ b/crates/sqlbuild-orchestration/src/warehouse_history/_helpers/maintenance.rs @@ -0,0 +1,86 @@ +//! Maintenance reads over the migration, old-name view and janitor mirrors. + +use serde::Deserialize; +use serde::de::DeserializeOwned; +use uuid::Uuid; + +use crate::store::errors::ControlError; +use crate::store::main::query_records::query_records; +use crate::store::models::{ControlTransaction, Parameter}; +use crate::warehouse_history::models::MaintenanceHistory; +use crate::warehouse_metadata::models::{ + WarehouseJanitorRecord, WarehouseMigrationRecord, WarehouseOldNameRecord, +}; + +#[derive(Deserialize)] +struct Recorded { + record: T, +} + +async fn records( + transaction: ControlTransaction, + sql: &str, + parameters: Vec, +) -> Result<(ControlTransaction, Vec), ControlError> { + let (transaction, rows): (ControlTransaction, Vec>) = + query_records(transaction, sql, parameters).await?; + Ok(( + transaction, + rows.into_iter().map(|row| row.record).collect(), + )) +} + +/// The newest moves, optionally into one model, with every step of their old-name views. +pub(crate) async fn maintenance_history( + transaction: ControlTransaction, + (location_id, target): (Uuid, &str), + (node, limit): (Option<&str>, u32), +) -> Result<(ControlTransaction, MaintenanceHistory), ControlError> { + let limit = Parameter::Integer(i64::from(limit.clamp(1, 1000))); + let scope = vec![Parameter::Id(location_id), Parameter::Text(target.into())]; + let (transaction, migrations): (ControlTransaction, Vec) = + match node { + Some(node) => { + records(transaction, + "SELECT record_json AS record FROM sqb_warehouse_migrations WHERE location_id=$1 AND target=$2 \ + AND destination_model=$3 ORDER BY observed_at DESC,event_id DESC LIMIT $4", + [scope.clone(), vec![Parameter::Text(node.into()), limit.clone()]].concat()).await? + } + None => { + records(transaction, + "SELECT record_json AS record FROM sqb_warehouse_migrations WHERE location_id=$1 AND target=$2 \ + ORDER BY observed_at DESC,event_id DESC LIMIT $3", + [scope.clone(), vec![limit.clone()]].concat()).await? + } + }; + let mut transaction = transaction; + let mut old_name_views: Vec = Vec::new(); + if !migrations.is_empty() { + let mut parameters = scope.clone(); + parameters.extend( + migrations + .iter() + .map(|migration| Parameter::Text(migration.event_id.clone())), + ); + let placeholders = Parameter::placeholders(3..=parameters.len()); + (transaction, old_name_views) = records(transaction, + &format!("SELECT record_json AS record FROM sqb_warehouse_old_name_views WHERE location_id=$1 AND target=$2 \ + AND migration_event_id IN ({placeholders}) ORDER BY observed_at,event_id"), + parameters).await?; + } + let mut janitor_events: Vec = Vec::new(); + if node.is_none() { + (transaction, janitor_events) = records(transaction, + "SELECT record_json AS record FROM sqb_warehouse_janitor_events WHERE location_id=$1 AND target=$2 \ + ORDER BY observed_at DESC,event_id DESC LIMIT $3", + [scope, vec![limit]].concat()).await?; + } + Ok(( + transaction, + MaintenanceHistory { + migrations, + old_name_views, + janitor_events, + }, + )) +} diff --git a/crates/sqlbuild-orchestration/src/warehouse_history/_helpers/mod.rs b/crates/sqlbuild-orchestration/src/warehouse_history/_helpers/mod.rs new file mode 100644 index 0000000000..e0260321e5 --- /dev/null +++ b/crates/sqlbuild-orchestration/src/warehouse_history/_helpers/mod.rs @@ -0,0 +1,2 @@ +pub(crate) mod checks; +pub(crate) mod maintenance; diff --git a/crates/sqlbuild-orchestration/src/warehouse_history/main/mod.rs b/crates/sqlbuild-orchestration/src/warehouse_history/main/mod.rs new file mode 100644 index 0000000000..5a314e91ca --- /dev/null +++ b/crates/sqlbuild-orchestration/src/warehouse_history/main/mod.rs @@ -0,0 +1,5 @@ +pub mod read_person_check; +pub mod read_person_check_evidence; +pub mod read_person_check_sql; +pub mod read_person_checks; +pub mod read_person_maintenance; diff --git a/crates/sqlbuild-orchestration/src/warehouse_metadata/main/read_person_check.rs b/crates/sqlbuild-orchestration/src/warehouse_history/main/read_person_check.rs similarity index 90% rename from crates/sqlbuild-orchestration/src/warehouse_metadata/main/read_person_check.rs rename to crates/sqlbuild-orchestration/src/warehouse_history/main/read_person_check.rs index c3f9a121c1..e48000c366 100644 --- a/crates/sqlbuild-orchestration/src/warehouse_metadata/main/read_person_check.rs +++ b/crates/sqlbuild-orchestration/src/warehouse_history/main/read_person_check.rs @@ -7,8 +7,8 @@ use crate::credentials::models::{AuditContext, DefaultRole}; use crate::store::errors::ControlError; use crate::store::main::{begin::begin, commit::commit}; use crate::store::models::{ControlStore, TransactionMode}; -use crate::warehouse_metadata::_helpers::checks::check_results; -use crate::warehouse_metadata::models::CheckResult; +use crate::warehouse_history::_helpers::checks::check_results; +use crate::warehouse_history::models::CheckResult; pub async fn read_person_check( store: &ControlStore, diff --git a/crates/sqlbuild-orchestration/src/warehouse_metadata/main/read_person_check_evidence.rs b/crates/sqlbuild-orchestration/src/warehouse_history/main/read_person_check_evidence.rs similarity index 84% rename from crates/sqlbuild-orchestration/src/warehouse_metadata/main/read_person_check_evidence.rs rename to crates/sqlbuild-orchestration/src/warehouse_history/main/read_person_check_evidence.rs index 8132c3e6f5..2efdbabf0a 100644 --- a/crates/sqlbuild-orchestration/src/warehouse_metadata/main/read_person_check_evidence.rs +++ b/crates/sqlbuild-orchestration/src/warehouse_history/main/read_person_check_evidence.rs @@ -6,8 +6,8 @@ use crate::credentials::models::{Action, AuditContext}; use crate::store::errors::ControlError; use crate::store::main::begin::begin; use crate::store::models::{ControlStore, TransactionMode}; -use crate::warehouse_metadata::_helpers::checks::{check_result, evidence, sensitive_check_read}; -use crate::warehouse_metadata::models::CheckEvidence; +use crate::warehouse_history::_helpers::checks::{check_result, evidence, sensitive_check_read}; +use crate::warehouse_history::models::CheckEvidence; pub async fn read_person_check_evidence( store: &ControlStore, diff --git a/crates/sqlbuild-orchestration/src/warehouse_metadata/main/read_person_check_sql.rs b/crates/sqlbuild-orchestration/src/warehouse_history/main/read_person_check_sql.rs similarity index 90% rename from crates/sqlbuild-orchestration/src/warehouse_metadata/main/read_person_check_sql.rs rename to crates/sqlbuild-orchestration/src/warehouse_history/main/read_person_check_sql.rs index 6e615cd52b..32f4f0c1e4 100644 --- a/crates/sqlbuild-orchestration/src/warehouse_metadata/main/read_person_check_sql.rs +++ b/crates/sqlbuild-orchestration/src/warehouse_history/main/read_person_check_sql.rs @@ -6,8 +6,8 @@ use crate::credentials::models::{Action, AuditContext}; use crate::store::errors::ControlError; use crate::store::main::begin::begin; use crate::store::models::{ControlStore, TransactionMode}; -use crate::warehouse_metadata::_helpers::checks::{check_results, sensitive_check_read}; -use crate::warehouse_metadata::models::CheckSql; +use crate::warehouse_history::_helpers::checks::{check_results, sensitive_check_read}; +use crate::warehouse_history::models::CheckSql; pub async fn read_person_check_sql( store: &ControlStore, diff --git a/crates/sqlbuild-orchestration/src/warehouse_metadata/main/read_person_checks.rs b/crates/sqlbuild-orchestration/src/warehouse_history/main/read_person_checks.rs similarity index 89% rename from crates/sqlbuild-orchestration/src/warehouse_metadata/main/read_person_checks.rs rename to crates/sqlbuild-orchestration/src/warehouse_history/main/read_person_checks.rs index 13632263e9..4a31d398bd 100644 --- a/crates/sqlbuild-orchestration/src/warehouse_metadata/main/read_person_checks.rs +++ b/crates/sqlbuild-orchestration/src/warehouse_history/main/read_person_checks.rs @@ -7,8 +7,8 @@ use crate::credentials::models::{AuditContext, DefaultRole}; use crate::store::errors::ControlError; use crate::store::main::{begin::begin, commit::commit}; use crate::store::models::{ControlStore, TransactionMode}; -use crate::warehouse_metadata::_helpers::checks::latest_results; -use crate::warehouse_metadata::models::CheckResult; +use crate::warehouse_history::_helpers::checks::latest_results; +use crate::warehouse_history::models::CheckResult; pub async fn read_person_checks( store: &ControlStore, diff --git a/crates/sqlbuild-orchestration/src/warehouse_history/main/read_person_maintenance.rs b/crates/sqlbuild-orchestration/src/warehouse_history/main/read_person_maintenance.rs new file mode 100644 index 0000000000..a74a92e6e8 --- /dev/null +++ b/crates/sqlbuild-orchestration/src/warehouse_history/main/read_person_maintenance.rs @@ -0,0 +1,34 @@ +//! Live view permission over the whole target and its recorded maintenance history. + +use uuid::Uuid; + +use crate::authorization::main::authorize_listing::authorize_listing; +use crate::credentials::models::{AuditContext, DefaultRole}; +use crate::store::errors::ControlError; +use crate::store::main::{begin::begin, commit::commit}; +use crate::store::models::{ControlStore, TransactionMode}; +use crate::warehouse_history::_helpers::maintenance::maintenance_history; +use crate::warehouse_history::models::{MaintenanceHistory, MaintenanceQuery}; + +pub async fn read_person_maintenance( + store: &ControlStore, + credentials: (&AuditContext, Uuid), + (location_id, target, options): (Uuid, &str, &MaintenanceQuery), + default_role: DefaultRole, +) -> Result { + let (transaction, _) = authorize_listing( + begin(store, TransactionMode::Write).await?, + credentials, + (location_id, target), + default_role, + ) + .await?; + let (transaction, history) = maintenance_history( + transaction, + (location_id, target), + (options.node.as_deref(), options.limit.unwrap_or(100)), + ) + .await?; + commit(transaction).await?; + Ok(history) +} diff --git a/crates/sqlbuild-orchestration/src/warehouse_history/mod.rs b/crates/sqlbuild-orchestration/src/warehouse_history/mod.rs new file mode 100644 index 0000000000..7488923fa9 --- /dev/null +++ b/crates/sqlbuild-orchestration/src/warehouse_history/mod.rs @@ -0,0 +1,5 @@ +//! Person reads of warehouse-owned history mirrored into the control store: checks and maintenance. + +pub(crate) mod _helpers; +pub mod main; +pub mod models; diff --git a/crates/sqlbuild-orchestration/src/warehouse_history/models.rs b/crates/sqlbuild-orchestration/src/warehouse_history/models.rs new file mode 100644 index 0000000000..cca287650b --- /dev/null +++ b/crates/sqlbuild-orchestration/src/warehouse_history/models.rs @@ -0,0 +1,114 @@ +//! Check results, their SQL and evidence, and maintenance history as people read them. + +use chrono::{DateTime, Utc}; +use serde::{Deserialize, Serialize}; +use uuid::Uuid; + +use crate::warehouse_metadata::models::{ + WarehouseAuditRecord, WarehouseJanitorRecord, WarehouseMigrationRecord, WarehouseOldNameRecord, +}; + +/// One stored check result. Its SQL, error texts and evidence rows need their own permissions. +#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize, utoipa::ToSchema)] +pub struct CheckResult { + pub event_id: Uuid, + /// Identity of the check; a renamed model or changed severity or thresholds starts a new one. + pub binding_key: String, + pub audit_name: String, + pub audit_definition_name: String, + pub audit_description: Option, + pub attachment_kind: String, + pub attached_target_kind: Option, + pub attached_target_name: Option, + pub attached_column_name: Option, + pub severity: String, + pub evaluation_mode: String, + pub run_scope_phase: String, + pub thresholds_json: Option, + pub occurred_at: DateTime, + pub run_id: String, + pub invocation_id: String, + pub outcome: String, + pub violation_count: Option, + pub measured_value: Option, + pub sample_count: Option, + pub sample_unit: Option, + pub minimum_samples: Option, + pub evidence_count: Option, + pub evidence_truncated: Option, + pub reused: String, +} + +impl CheckResult { + pub(crate) fn new(event_id: Uuid, record: WarehouseAuditRecord) -> Self { + Self { + event_id, + binding_key: record.binding_key, + audit_name: record.audit_name, + audit_definition_name: record.audit_definition_name, + audit_description: record.audit_description, + attachment_kind: record.attachment_kind, + attached_target_kind: record.attached_target_kind, + attached_target_name: record.attached_target_name, + attached_column_name: record.attached_column_name, + severity: record.severity, + evaluation_mode: record.evaluation_mode, + run_scope_phase: record.run_scope_phase, + thresholds_json: record.thresholds_json, + occurred_at: record.occurred_at, + run_id: record.run_id, + invocation_id: record.invocation_id, + outcome: record.outcome, + violation_count: record.violation_count, + measured_value: record.measured_value, + sample_count: record.sample_count, + sample_unit: record.sample_unit, + minimum_samples: record.minimum_samples, + evidence_count: record.evidence_count, + evidence_truncated: record.evidence_truncated, + reused: record.reused, + } + } +} + +/// A check's latest SQL and error texts, which quote SQL; `view.sql` reads them, audited. +#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize, utoipa::ToSchema)] +pub struct CheckSql { + pub event_id: Uuid, + pub binding_key: String, + pub executed_sql: Option, + pub measurement_sql: Option, + pub evidence_sql: Option, + pub execution_error: Option, + pub evidence_error: Option, +} + +/// One result's stored evidence rows; `view.evidence` reads them, audited. +#[derive(Clone, Debug, PartialEq, Serialize, Deserialize, utoipa::ToSchema)] +pub struct CheckEvidence { + pub event_id: Uuid, + pub binding_key: String, + /// Every row key, in sorted order; the warehouse stores rows with sorted keys. + pub columns: Vec, + /// Scalar cells aligned to `columns`; a row without a key holds null there. + #[schema(value_type = Vec>)] + pub rows: Vec>, + pub truncated: Option, +} + +/// Recorded moves, their old-name view steps and janitor cleanups, newest moves first. +#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize, utoipa::ToSchema)] +pub struct MaintenanceHistory { + pub migrations: Vec, + /// Every step recorded for the listed moves, oldest first; same-time steps are not in step order. + pub old_name_views: Vec, + /// Empty when the history is for one model; cleanups are not tied to models. + pub janitor_events: Vec, +} + +#[derive(Clone, Debug, Deserialize)] +#[serde(deny_unknown_fields)] +pub struct MaintenanceQuery { + pub node: Option, + pub limit: Option, +} diff --git a/crates/sqlbuild-orchestration/src/warehouse_metadata/_helpers/mod.rs b/crates/sqlbuild-orchestration/src/warehouse_metadata/_helpers/mod.rs index 6df3ded4b0..35e0fdc180 100644 --- a/crates/sqlbuild-orchestration/src/warehouse_metadata/_helpers/mod.rs +++ b/crates/sqlbuild-orchestration/src/warehouse_metadata/_helpers/mod.rs @@ -1,5 +1,4 @@ pub(crate) mod catalog; -pub(crate) mod checks; pub(crate) mod events; pub(crate) mod fingerprint_reads; pub(crate) mod fingerprints; diff --git a/crates/sqlbuild-orchestration/src/warehouse_metadata/main/mod.rs b/crates/sqlbuild-orchestration/src/warehouse_metadata/main/mod.rs index 722d6939c3..4ebb52f84f 100644 --- a/crates/sqlbuild-orchestration/src/warehouse_metadata/main/mod.rs +++ b/crates/sqlbuild-orchestration/src/warehouse_metadata/main/mod.rs @@ -9,10 +9,6 @@ pub mod ingest_source_freshness; pub mod ingest_statistics; pub(crate) mod prune_statistics; pub mod read_person_catalog; -pub mod read_person_check; -pub mod read_person_check_evidence; -pub mod read_person_check_sql; -pub mod read_person_checks; pub mod read_person_fingerprint; pub mod read_person_node_status; pub mod read_person_statistics; diff --git a/crates/sqlbuild-orchestration/src/warehouse_metadata/models.rs b/crates/sqlbuild-orchestration/src/warehouse_metadata/models.rs index 31c94dcb87..61e87d6a6c 100644 --- a/crates/sqlbuild-orchestration/src/warehouse_metadata/models.rs +++ b/crates/sqlbuild-orchestration/src/warehouse_metadata/models.rs @@ -74,94 +74,6 @@ pub struct WarehouseAuditRecord { pub audit_meta_json: Option, } -/// One stored check result. Its SQL, error texts and evidence rows need their own permissions. -#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize, utoipa::ToSchema)] -pub struct CheckResult { - pub event_id: Uuid, - /// Identity of the check; a renamed model or changed severity or thresholds starts a new one. - pub binding_key: String, - pub audit_name: String, - pub audit_definition_name: String, - pub audit_description: Option, - pub attachment_kind: String, - pub attached_target_kind: Option, - pub attached_target_name: Option, - pub attached_column_name: Option, - pub severity: String, - pub evaluation_mode: String, - pub run_scope_phase: String, - pub thresholds_json: Option, - pub occurred_at: DateTime, - pub run_id: String, - pub invocation_id: String, - pub outcome: String, - pub violation_count: Option, - pub measured_value: Option, - pub sample_count: Option, - pub sample_unit: Option, - pub minimum_samples: Option, - pub evidence_count: Option, - pub evidence_truncated: Option, - pub reused: String, -} - -impl CheckResult { - pub(crate) fn new(event_id: Uuid, record: WarehouseAuditRecord) -> Self { - Self { - event_id, - binding_key: record.binding_key, - audit_name: record.audit_name, - audit_definition_name: record.audit_definition_name, - audit_description: record.audit_description, - attachment_kind: record.attachment_kind, - attached_target_kind: record.attached_target_kind, - attached_target_name: record.attached_target_name, - attached_column_name: record.attached_column_name, - severity: record.severity, - evaluation_mode: record.evaluation_mode, - run_scope_phase: record.run_scope_phase, - thresholds_json: record.thresholds_json, - occurred_at: record.occurred_at, - run_id: record.run_id, - invocation_id: record.invocation_id, - outcome: record.outcome, - violation_count: record.violation_count, - measured_value: record.measured_value, - sample_count: record.sample_count, - sample_unit: record.sample_unit, - minimum_samples: record.minimum_samples, - evidence_count: record.evidence_count, - evidence_truncated: record.evidence_truncated, - reused: record.reused, - } - } -} - -/// A check's latest SQL and error texts, which quote SQL; `view.sql` reads them, audited. -#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize, utoipa::ToSchema)] -pub struct CheckSql { - pub event_id: Uuid, - pub binding_key: String, - pub executed_sql: Option, - pub measurement_sql: Option, - pub evidence_sql: Option, - pub execution_error: Option, - pub evidence_error: Option, -} - -/// One result's stored evidence rows; `view.evidence` reads them, audited. -#[derive(Clone, Debug, PartialEq, Serialize, Deserialize, utoipa::ToSchema)] -pub struct CheckEvidence { - pub event_id: Uuid, - pub binding_key: String, - /// Every row key, in sorted order; the warehouse stores rows with sorted keys. - pub columns: Vec, - /// Scalar cells aligned to `columns`; a row without a key holds null there. - #[schema(value_type = Vec>)] - pub rows: Vec>, - pub truncated: Option, -} - /// Warehouse cleanup facts do not authorize control-plane deletion or worker recovery. #[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize, utoipa::ToSchema)] #[serde(deny_unknown_fields)] diff --git a/crates/sqlbuild-orchestration/tests/api/checks/helpers.rs b/crates/sqlbuild-orchestration/tests/api/checks/helpers.rs deleted file mode 100644 index ca8d39c2d2..0000000000 --- a/crates/sqlbuild-orchestration/tests/api/checks/helpers.rs +++ /dev/null @@ -1,68 +0,0 @@ -//! Password sessions, fixture loading and audit reads for the check API tests. - -use reqwest::{Client, StatusCode, header}; -use serde_json::{Value, json}; - -use crate::store::main::{begin::begin, commit::commit, query::query}; -use crate::store::models::{ControlStore, TransactionMode}; - -pub(crate) const ORIGIN: &str = "https://control.example.com"; -pub(crate) const PASSWORD: &str = "orders-password-2026"; - -pub(crate) fn fixture(name: &str) -> Value { - let path = std::path::Path::new(env!("CARGO_MANIFEST_DIR")) - .join("../../benchmarks/duckshop/edge_cases/orchestration") - .join(name); - serde_json::from_str(&std::fs::read_to_string(path).expect("fixture")).expect("fixture JSON") -} - -pub(crate) async fn get( - client: &Client, - (url, cookie): (&str, &str), - path: &str, -) -> (StatusCode, Value) { - let response = client - .get(format!("{url}{path}")) - .header(header::HOST, "control.example.com") - .header(header::ORIGIN, ORIGIN) - .header("cookie", cookie) - .send() - .await - .expect("check read"); - let status = response.status(); - (status, response.json().await.unwrap_or(Value::Null)) -} - -pub(crate) async fn login(client: &Client, url: &str, username: &str) -> String { - let response = client - .post(format!("{url}/api/auth/password/login")) - .header(header::HOST, "control.example.com") - .header(header::ORIGIN, ORIGIN) - .json(&json!({"username": username, "password": PASSWORD})) - .send() - .await - .expect("password login"); - assert_eq!(response.status(), StatusCode::OK); - response - .headers() - .get_all(header::SET_COOKIE) - .iter() - .map(|value| value.to_str().expect("cookie")) - .find(|value| value.starts_with("__Host-sqb-session=")) - .expect("browser session") - .split(';') - .next() - .expect("cookie pair") - .to_owned() -} - -pub(crate) async fn latest_audit(store: &ControlStore) -> Value { - let transaction = begin(store, TransactionMode::Read) - .await - .expect("audit cut"); - let (transaction, rows) = query(transaction, - "SELECT action,granting_role,scope_json,after_json FROM sqb_security_audit ORDER BY storage_id DESC LIMIT 1", - vec![]).await.expect("audit"); - commit(transaction).await.expect("close audit cut"); - json!(rows[0]) -} diff --git a/crates/sqlbuild-orchestration/tests/api/test_checks.rs b/crates/sqlbuild-orchestration/tests/api/test_checks.rs index 24e1fc9acc..46e3c3bf7a 100644 --- a/crates/sqlbuild-orchestration/tests/api/test_checks.rs +++ b/crates/sqlbuild-orchestration/tests/api/test_checks.rs @@ -1,35 +1,19 @@ //! Check reads through the public API: keyed by binding key, view-gated, with audited SQL. -use reqwest::{Client, StatusCode}; +use reqwest::StatusCode; use serde_json::{Value, json}; use uuid::Uuid; -use crate::api::models::{ApiAuthentication, HumanLogin}; -use crate::api::tests::checks_helpers::{ORIGIN, PASSWORD, fixture, get, latest_audit, login}; -use crate::api::tests::helpers::{start_api_with_authentication, stop_api}; +use crate::api::tests::helpers::stop_api; use crate::api::tests::test_types::WarehouseMetadataTestCase; -use crate::authorization::main::{ - assign_builtin_role::assign_builtin_role, bootstrap_user::bootstrap_user, +use crate::api::tests::warehouse_history_helpers::{ + batch, fixture, get, latest_audit, login, seed_target, start_password_api, }; -use crate::credentials::main::{ - identities::create_principal::create_principal, sessions::login_password::login_password, - sessions::set_password::set_password, -}; -use crate::credentials::models::{ - AuditContext, BuiltinRole, PasswordAuthority, PasswordLogin, RoleAssignmentRequest, -}; -use crate::credentials::tests::location_helpers::create_location; -use crate::platform::main::report_location_health::report_location_health; -use crate::platform::models::{LocationHealth, LocationHealthReport}; -use crate::runs::tests::helpers::snapshot; -use crate::snapshots::main::publish_snapshot::publish_snapshot; -use crate::store::main::migrate_store::migrate_store; use crate::store::tests::fault_helpers::{install_fault, remove_fault}; use crate::store::tests::fault_types::TransactionFault; use crate::store::tests::store_helpers::{close_test_store, open_test_store, run}; use crate::store::tests::store_test_types::Backend; use crate::warehouse_metadata::main::ingest_audit_results::ingest_audit_results; -use crate::warehouse_metadata::models::MetadataBatch; #[test] fn given_checks_sharing_an_audit_name_when_people_read_them_then_binding_keys_and_sql_permissions_hold() @@ -50,79 +34,7 @@ fn given_checks_sharing_an_audit_name_when_people_read_them_then_binding_keys_an run(async { let fixture_store = open_test_store(test_case.backend).await.expect("store"); let store = &fixture_store.store; - migrate_store(store).await.expect("migrate"); - let context = AuditContext { - actor: "control-store-owner".into(), - granting_role: "control-store-owner".into(), - request_id: Uuid::now_v7(), - ip_address: Some("127.0.0.1".parse().expect("local peer")), - }; - let admin = bootstrap_user(store, &context, "orders-admin") - .await - .expect("bootstrap"); - let PasswordLogin::Authenticated(admin_session) = ({ - set_password( - store, - &context, - PasswordAuthority::Operator(admin), - PASSWORD, - ) - .await - .expect("admin password"); - login_password(store, &context, ("orders-admin", PASSWORD)) - .await - .expect("admin login") - }) else { - panic!("authenticated administrator"); - }; - let location = create_location(store, "orders", "test-admin") - .await - .expect("location"); - for (name, role) in [ - ("orders-viewer", BuiltinRole::Viewer), - ("orders-developer", BuiltinRole::Developer), - ] { - let person = create_principal(store, &context, name) - .await - .expect("person"); - set_password( - store, - &context, - PasswordAuthority::Operator(person), - PASSWORD, - ) - .await - .expect("password"); - assign_builtin_role( - store, - (&context, admin_session.identity.principal_id), - &RoleAssignmentRequest { - principal_id: person, - role: role.into(), - location_id: Some(location), - target: Some("development".into()), - revoke: false, - }, - ) - .await - .expect("scoped role"); - } - publish_snapshot(store, &snapshot(location)) - .await - .expect("snapshot"); - let session_id = Uuid::now_v7(); - report_location_health( - store, - location, - &LocationHealthReport { - event_id: Uuid::now_v7(), - session_id, - target: "development".into(), - health: LocationHealth::Heartbeat {}, - }, - ) - .await - .expect("reader"); + let (location, session_id) = seed_target(store).await; let data = fixture("check_identity.json"); let base = fixture("audit_result_history.json")["record"].clone(); let records: Vec = data["results"] @@ -146,24 +58,18 @@ fn given_checks_sharing_an_audit_name_when_people_read_them_then_binding_keys_an }) .collect(); assert_eq!(records.len(), test_case.expected_observations); - let batch: MetadataBatch = serde_json::from_value(json!({"kind":"audit_results","session_id":session_id,"generation":1,"records":records, - "scan":{"scan_id":Uuid::now_v7(),"location":{"database":null,"schema":"main"},"complete":false}})).expect("batch"); assert!( - ingest_audit_results(store, (location, "development"), &batch) - .await - .expect("real audit ingestion") - .applied + ingest_audit_results( + store, + (location, "development"), + &batch(session_id, "audit_results", records) + ) + .await + .expect("real audit ingestion") + .applied ); - let api = start_api_with_authentication( - store, - ApiAuthentication { - login: HumanLogin::Password(ORIGIN.into()), - ..Default::default() - }, - ) - .await; - let client = Client::new(); + let (api, client) = start_password_api(store).await; let viewer = login(&client, &api.url, "orders-viewer").await; let developer = login(&client, &api.url, "orders-developer").await; let administrator = login(&client, &api.url, "orders-admin").await; diff --git a/crates/sqlbuild-orchestration/tests/api/test_contracts.rs b/crates/sqlbuild-orchestration/tests/api/test_contracts.rs index e9878e27e2..25527b6ad0 100644 --- a/crates/sqlbuild-orchestration/tests/api/test_contracts.rs +++ b/crates/sqlbuild-orchestration/tests/api/test_contracts.rs @@ -2,8 +2,6 @@ #[path = "changes/helpers.rs"] mod changes_helpers; -#[path = "checks/helpers.rs"] -pub(crate) mod checks_helpers; #[path = "fingerprint_reads/helpers.rs"] mod fingerprint_reads_helpers; #[path = "helpers.rs"] @@ -50,6 +48,8 @@ mod test_external_sessions; mod test_location_health; #[path = "test_location_work.rs"] mod test_location_work; +#[path = "test_maintenance.rs"] +mod test_maintenance; #[path = "test_oidc.rs"] mod test_oidc; #[path = "test_oidc_groups.rs"] @@ -76,6 +76,8 @@ mod test_worker; mod warehouse_catalog_helpers; #[path = "warehouse_metadata/events/helpers.rs"] mod warehouse_events_helpers; +#[path = "warehouse_history/helpers.rs"] +pub(crate) mod warehouse_history_helpers; #[path = "warehouse_metadata/helpers.rs"] mod warehouse_metadata_helpers; #[path = "warehouse_metadata/reconciliation/helpers.rs"] diff --git a/crates/sqlbuild-orchestration/tests/api/test_maintenance.rs b/crates/sqlbuild-orchestration/tests/api/test_maintenance.rs new file mode 100644 index 0000000000..c19db6b760 --- /dev/null +++ b/crates/sqlbuild-orchestration/tests/api/test_maintenance.rs @@ -0,0 +1,162 @@ +//! Maintenance history through the public API: moves, their old-name view steps and cleanups. + +use reqwest::StatusCode; +use serde_json::{Value, json}; +use uuid::Uuid; + +use crate::api::tests::helpers::stop_api; +use crate::api::tests::test_types::WarehouseMetadataTestCase; +use crate::api::tests::warehouse_history_helpers::{ + batch, fixture, get, ids, login, records, seed_target, start_password_api, +}; +use crate::platform::main::report_location_health::report_location_health; +use crate::platform::models::{LocationHealth, LocationHealthReport}; +use crate::runs::tests::helpers::snapshot; +use crate::snapshots::main::publish_snapshot::publish_snapshot; +use crate::store::tests::store_helpers::{close_test_store, open_test_store, run}; +use crate::store::tests::store_test_types::Backend; +use crate::warehouse_metadata::main::{ + ingest_janitor_events::ingest_janitor_events, ingest_migrations::ingest_migrations, + ingest_old_name_views::ingest_old_name_views, +}; + +#[test] +fn given_moves_views_and_cleanups_when_people_read_maintenance_then_each_model_sees_its_own_history() + { + let test_cases = [ + WarehouseMetadataTestCase { + description: "SQLite maintenance reads", + backend: Backend::Sqlite, + expected_observations: 9, + }, + WarehouseMetadataTestCase { + description: "PostgreSQL maintenance reads", + backend: Backend::Postgres, + expected_observations: 9, + }, + ]; + for test_case in test_cases { + run(async { + let fixture_store = open_test_store(test_case.backend).await.expect("store"); + let store = &fixture_store.store; + let (location, session_id) = seed_target(store).await; + let data = fixture("maintenance_history.json"); + let scope = (location, "development"); + let moves = records("migration_history.json", &data["migrations"]); + let views = records("old_name_history.json", &data["old_name_views"]); + let cleanups = records("janitor_event_history.json", &data["janitor_events"]); + assert_eq!( + moves.len() + views.len() + cleanups.len(), + test_case.expected_observations + ); + ingest_migrations(store, scope, &batch(session_id, "migrations", moves)) + .await + .expect("real migration ingestion"); + ingest_old_name_views(store, scope, &batch(session_id, "old_name_views", views)) + .await + .expect("real old-name ingestion"); + ingest_janitor_events(store, scope, &batch(session_id, "janitor_events", cleanups)) + .await + .expect("real janitor ingestion"); + + let mut preview = snapshot(location); + preview.target = "preview".into(); + publish_snapshot(store, &preview) + .await + .expect("preview snapshot"); + let preview_session = Uuid::now_v7(); + report_location_health( + store, + location, + &LocationHealthReport { + event_id: Uuid::now_v7(), + session_id: preview_session, + target: "preview".into(), + health: LocationHealth::Heartbeat {}, + }, + ) + .await + .expect("preview reader"); + ingest_old_name_views( + store, + (location, "preview"), + &batch( + preview_session, + "old_name_views", + records("old_name_history.json", &data["preview_old_name_views"]), + ), + ) + .await + .expect("another target's view"); + + let (api, client) = start_password_api(store).await; + let viewer = login(&client, &api.url, "orders-viewer").await; + let path = format!("/api/control/locations/{location}/targets/development/maintenance"); + + let (status, all) = get(&client, (&api.url, &viewer), &path).await; + assert_eq!(status, StatusCode::OK, "{}", test_case.description); + assert_eq!( + ids(&all, "migrations"), + ["move-orders-2", "move-customers-1", "move-orders-1"] + ); + assert_eq!( + ids(&all, "old_name_views"), + [ + "view-orders-old-1", + "view-customers-1", + "view-orders-1", + "view-orders-2" + ], + "another target's steps never appear" + ); + assert_eq!(ids(&all, "janitor_events"), ["cleanup-2", "cleanup-1"]); + assert_eq!(all["migrations"][0]["origin_model"], Value::Null); + assert_eq!(all["migrations"][0]["discovery"], "automatic"); + + let (_, orders) = + get(&client, (&api.url, &viewer), &format!("{path}?node=orders")).await; + assert_eq!( + ids(&orders, "migrations"), + ["move-orders-2", "move-orders-1"] + ); + assert_eq!( + ids(&orders, "old_name_views"), + ["view-orders-old-1", "view-orders-1", "view-orders-2"], + "another model's views never appear" + ); + assert_eq!(orders["janitor_events"], json!([])); + + let (_, newest) = get(&client, (&api.url, &viewer), &format!("{path}?limit=1")).await; + assert_eq!(ids(&newest, "migrations"), ["move-orders-2"]); + assert_eq!( + ids(&newest, "old_name_views"), + ["view-orders-1", "view-orders-2"], + "only the listed moves' steps, not every step of their models" + ); + assert_eq!(ids(&newest, "janitor_events"), ["cleanup-2"]); + + let (_, unknown) = get( + &client, + (&api.url, &viewer), + &format!("{path}?node=unknown"), + ) + .await; + assert_eq!( + unknown, + json!({"migrations": [], "old_name_views": [], "janitor_events": []}) + ); + assert_eq!( + get( + &client, + (&api.url, &viewer), + &path.replace("development", "preview") + ) + .await + .0, + StatusCode::FORBIDDEN + ); + stop_api(api).await; + close_test_store(fixture_store).await; + }); + } +} diff --git a/crates/sqlbuild-orchestration/tests/api/warehouse_history/helpers.rs b/crates/sqlbuild-orchestration/tests/api/warehouse_history/helpers.rs new file mode 100644 index 0000000000..b7679b1b7d --- /dev/null +++ b/crates/sqlbuild-orchestration/tests/api/warehouse_history/helpers.rs @@ -0,0 +1,214 @@ +//! A seeded target with scoped people, password sessions, fixtures and audit reads for warehouse history API tests. + +use reqwest::{Client, StatusCode, header}; +use serde_json::{Value, json}; +use uuid::Uuid; + +use crate::api::models::{ApiAuthentication, HumanLogin}; +use crate::api::tests::helpers::start_api_with_authentication; +use crate::api::tests::test_types::TestApi; +use crate::authorization::main::{ + assign_builtin_role::assign_builtin_role, bootstrap_user::bootstrap_user, +}; +use crate::credentials::main::{ + identities::create_principal::create_principal, sessions::login_password::login_password, + sessions::set_password::set_password, +}; +use crate::credentials::models::{ + AuditContext, BuiltinRole, PasswordAuthority, PasswordLogin, RoleAssignmentRequest, +}; +use crate::credentials::tests::location_helpers::create_location; +use crate::platform::main::report_location_health::report_location_health; +use crate::platform::models::{LocationHealth, LocationHealthReport}; +use crate::runs::tests::helpers::snapshot; +use crate::snapshots::main::publish_snapshot::publish_snapshot; +use crate::store::main::{ + begin::begin, commit::commit, migrate_store::migrate_store, query::query, +}; +use crate::store::models::{ControlStore, TransactionMode}; +use crate::warehouse_metadata::models::MetadataBatch; + +pub(crate) const ORIGIN: &str = "https://control.example.com"; +pub(crate) const PASSWORD: &str = "orders-password-2026"; + +pub(crate) fn fixture(name: &str) -> Value { + let path = std::path::Path::new(env!("CARGO_MANIFEST_DIR")) + .join("../../benchmarks/duckshop/edge_cases/orchestration") + .join(name); + serde_json::from_str(&std::fs::read_to_string(path).expect("fixture")).expect("fixture JSON") +} + +pub(crate) async fn get( + client: &Client, + (url, cookie): (&str, &str), + path: &str, +) -> (StatusCode, Value) { + let response = client + .get(format!("{url}{path}")) + .header(header::HOST, "control.example.com") + .header(header::ORIGIN, ORIGIN) + .header("cookie", cookie) + .send() + .await + .expect("check read"); + let status = response.status(); + (status, response.json().await.unwrap_or(Value::Null)) +} + +pub(crate) async fn login(client: &Client, url: &str, username: &str) -> String { + let response = client + .post(format!("{url}/api/auth/password/login")) + .header(header::HOST, "control.example.com") + .header(header::ORIGIN, ORIGIN) + .json(&json!({"username": username, "password": PASSWORD})) + .send() + .await + .expect("password login"); + assert_eq!(response.status(), StatusCode::OK); + response + .headers() + .get_all(header::SET_COOKIE) + .iter() + .map(|value| value.to_str().expect("cookie")) + .find(|value| value.starts_with("__Host-sqb-session=")) + .expect("browser session") + .split(';') + .next() + .expect("cookie pair") + .to_owned() +} + +pub(crate) async fn latest_audit(store: &ControlStore) -> Value { + let transaction = begin(store, TransactionMode::Read) + .await + .expect("audit cut"); + let (transaction, rows) = query(transaction, + "SELECT action,granting_role,scope_json,after_json FROM sqb_security_audit ORDER BY storage_id DESC LIMIT 1", + vec![]).await.expect("audit"); + commit(transaction).await.expect("close audit cut"); + json!(rows[0]) +} + +/// A migrated store with scoped people, the orders snapshot and a live metadata reader session. +pub(crate) async fn seed_target(store: &ControlStore) -> (Uuid, Uuid) { + migrate_store(store).await.expect("migrate"); + let context = AuditContext { + actor: "control-store-owner".into(), + granting_role: "control-store-owner".into(), + request_id: Uuid::now_v7(), + ip_address: Some("127.0.0.1".parse().expect("local peer")), + }; + let admin = bootstrap_user(store, &context, "orders-admin") + .await + .expect("bootstrap"); + let PasswordLogin::Authenticated(admin_session) = ({ + set_password( + store, + &context, + PasswordAuthority::Operator(admin), + PASSWORD, + ) + .await + .expect("admin password"); + login_password(store, &context, ("orders-admin", PASSWORD)) + .await + .expect("admin login") + }) else { + panic!("authenticated administrator"); + }; + let location = create_location(store, "orders", "test-admin") + .await + .expect("location"); + for (name, role) in [ + ("orders-viewer", BuiltinRole::Viewer), + ("orders-developer", BuiltinRole::Developer), + ] { + let person = create_principal(store, &context, name) + .await + .expect("person"); + set_password( + store, + &context, + PasswordAuthority::Operator(person), + PASSWORD, + ) + .await + .expect("password"); + assign_builtin_role( + store, + (&context, admin_session.identity.principal_id), + &RoleAssignmentRequest { + principal_id: person, + role: role.into(), + location_id: Some(location), + target: Some("development".into()), + revoke: false, + }, + ) + .await + .expect("scoped role"); + } + publish_snapshot(store, &snapshot(location)) + .await + .expect("snapshot"); + let session_id = Uuid::now_v7(); + report_location_health( + store, + location, + &LocationHealthReport { + event_id: Uuid::now_v7(), + session_id, + target: "development".into(), + health: LocationHealth::Heartbeat {}, + }, + ) + .await + .expect("reader"); + (location, session_id) +} + +/// One metadata batch of `kind` from the reader session, scanning the `main` state schema. +pub(crate) fn batch(session_id: Uuid, kind: &str, records: Vec) -> MetadataBatch { + serde_json::from_value(json!({"kind":kind,"session_id":session_id,"generation":1,"records":records, + "scan":{"scan_id":Uuid::now_v7(),"location":{"database":null,"schema":"main"},"complete":false}})).expect("batch") +} + +/// An API with password sign-in and a client for it. +pub(crate) async fn start_password_api(store: &ControlStore) -> (TestApi, Client) { + let api = start_api_with_authentication( + store, + ApiAuthentication { + login: HumanLogin::Password(ORIGIN.into()), + ..Default::default() + }, + ) + .await; + (api, Client::new()) +} + +/// Fixture records of `base` with each override applied. +pub(crate) fn records(base: &str, overrides: &Value) -> Vec { + let base = fixture(base)["record"].clone(); + overrides + .as_array() + .expect("overrides") + .iter() + .map(|values| { + let mut record = base.clone(); + for (key, value) in values.as_object().expect("override") { + record[key] = value.clone(); + } + record + }) + .collect() +} + +/// The event IDs of one kind of record in a history response. +pub(crate) fn ids(history: &Value, kind: &str) -> Vec { + history[kind] + .as_array() + .expect("records") + .iter() + .map(|record| record["event_id"].as_str().expect("event").to_owned()) + .collect() +} diff --git a/fensu.toml b/fensu.toml index 146aa9b37e..b36a40292e 100644 --- a/fensu.toml +++ b/fensu.toml @@ -1367,11 +1367,11 @@ path = "src/sqlbuild/adapters/sqlserver/classes/sqlserver_adapter.py" reason = "The SQL Server adapter intentionally keeps overrideable backend behavior on one class." [[dupes.allowlist]] -paths = ["crates/sqlbuild-orchestration/src/api/_helpers/resources/administrators.rs", "crates/sqlbuild-orchestration/src/api/_helpers/resources/audit.rs", "crates/sqlbuild-orchestration/src/api/_helpers/resources/location_health.rs", "crates/sqlbuild-orchestration/src/api/_helpers/resources/location_work.rs", "crates/sqlbuild-orchestration/src/api/_helpers/resources/person_checks.rs", "crates/sqlbuild-orchestration/src/api/_helpers/resources/person_location_work.rs", "crates/sqlbuild-orchestration/src/api/_helpers/resources/person_node_sql.rs", "crates/sqlbuild-orchestration/src/api/_helpers/resources/person_operations.rs", "crates/sqlbuild-orchestration/src/api/_helpers/resources/run_reads.rs", "crates/sqlbuild-orchestration/src/api/_helpers/resources/runs.rs", "crates/sqlbuild-orchestration/src/api/_helpers/resources/snapshots.rs", "crates/sqlbuild-orchestration/src/api/_helpers/resources/trigger_reads.rs", "crates/sqlbuild-orchestration/src/api/_helpers/resources/warehouse_metadata.rs"] +paths = ["crates/sqlbuild-orchestration/src/api/_helpers/resources/administrators.rs", "crates/sqlbuild-orchestration/src/api/_helpers/resources/audit.rs", "crates/sqlbuild-orchestration/src/api/_helpers/resources/location_health.rs", "crates/sqlbuild-orchestration/src/api/_helpers/resources/location_work.rs", "crates/sqlbuild-orchestration/src/api/_helpers/resources/person_checks.rs", "crates/sqlbuild-orchestration/src/api/_helpers/resources/person_location_work.rs", "crates/sqlbuild-orchestration/src/api/_helpers/resources/person_maintenance.rs", "crates/sqlbuild-orchestration/src/api/_helpers/resources/person_node_sql.rs", "crates/sqlbuild-orchestration/src/api/_helpers/resources/person_operations.rs", "crates/sqlbuild-orchestration/src/api/_helpers/resources/run_reads.rs", "crates/sqlbuild-orchestration/src/api/_helpers/resources/runs.rs", "crates/sqlbuild-orchestration/src/api/_helpers/resources/snapshots.rs", "crates/sqlbuild-orchestration/src/api/_helpers/resources/trigger_reads.rs", "crates/sqlbuild-orchestration/src/api/_helpers/resources/warehouse_metadata.rs"] reason = "Each function is one utoipa-annotated axum route: OpenAPI needs one annotated handler per operation, and each body only checks the caller's scope and calls one store operation. Scope checks and store work live in their shared owners." [[dupes.allowlist]] -paths = ["crates/sqlbuild-orchestration/src/authorization/main/read_security_audit.rs", "crates/sqlbuild-orchestration/src/credentials/_helpers/active_session.rs", "crates/sqlbuild-orchestration/src/declarations/main/list_triggers.rs", "crates/sqlbuild-orchestration/src/platform/main/read_location_work.rs", "crates/sqlbuild-orchestration/src/runs/main/list_person_runs.rs", "crates/sqlbuild-orchestration/src/runs/main/list_runs.rs", "crates/sqlbuild-orchestration/src/runs/_helpers/lifecycle/reads.rs", "crates/sqlbuild-orchestration/src/warehouse_metadata/_helpers/checks.rs", "crates/sqlbuild-orchestration/src/scheduler/main/list_ticks.rs", "crates/sqlbuild-orchestration/src/sensors/_helpers/attempts.rs", "crates/sqlbuild-orchestration/src/sensors/_helpers/facts.rs", "crates/sqlbuild-orchestration/src/snapshots/main/read_current_record.rs", "crates/sqlbuild-orchestration/src/snapshots/main/read_snapshot_documents.rs", "crates/sqlbuild-orchestration/src/snapshots/main/read_snapshot_nodes.rs", "crates/sqlbuild-orchestration/src/warehouse_metadata/main/read_person_catalog.rs", "crates/sqlbuild-orchestration/src/warehouse_metadata/main/read_person_check.rs", "crates/sqlbuild-orchestration/src/warehouse_metadata/main/read_person_checks.rs", "crates/sqlbuild-orchestration/src/warehouse_metadata/main/read_person_statistics.rs"] +paths = ["crates/sqlbuild-orchestration/src/authorization/main/read_security_audit.rs", "crates/sqlbuild-orchestration/src/credentials/_helpers/active_session.rs", "crates/sqlbuild-orchestration/src/declarations/main/list_triggers.rs", "crates/sqlbuild-orchestration/src/platform/main/read_location_work.rs", "crates/sqlbuild-orchestration/src/runs/main/list_person_runs.rs", "crates/sqlbuild-orchestration/src/runs/main/list_runs.rs", "crates/sqlbuild-orchestration/src/runs/_helpers/lifecycle/reads.rs", "crates/sqlbuild-orchestration/src/warehouse_history/_helpers/checks.rs", "crates/sqlbuild-orchestration/src/scheduler/main/list_ticks.rs", "crates/sqlbuild-orchestration/src/sensors/_helpers/attempts.rs", "crates/sqlbuild-orchestration/src/sensors/_helpers/facts.rs", "crates/sqlbuild-orchestration/src/snapshots/main/read_current_record.rs", "crates/sqlbuild-orchestration/src/snapshots/main/read_snapshot_documents.rs", "crates/sqlbuild-orchestration/src/snapshots/main/read_snapshot_nodes.rs", "crates/sqlbuild-orchestration/src/warehouse_metadata/main/read_person_catalog.rs", "crates/sqlbuild-orchestration/src/warehouse_history/main/read_person_check.rs", "crates/sqlbuild-orchestration/src/warehouse_history/main/read_person_checks.rs", "crates/sqlbuild-orchestration/src/warehouse_history/main/read_person_maintenance.rs", "crates/sqlbuild-orchestration/src/warehouse_metadata/main/read_person_statistics.rs"] reason = "Each is one control-store read with its own SQL and record type. Transactions, person authorization and decoding go through the shared begin_authorized, query_records, read_records and query_bodies helpers; what remains alike is the call shape." [[dupes.allowlist]] @@ -1632,8 +1632,6 @@ allowed = [ "sqb_warehouse_microbatches.execution_run_id=Reader comes with the UI epic (D132), which filters on it: the run that executed the batch.", "sqb_warehouse_microbatches.record_type=Reader comes with the UI epic (D132), which filters on it: the batch state.", "sqb_warehouse_migrations.decision=Reader comes with the UI epic (D132), which filters on it.", - "sqb_warehouse_migrations.destination_model=Reader comes with the UI epic (D132), which filters on it.", - "sqb_warehouse_old_name_views.migration_event_id=Reader comes with the UI epic (D132), which filters on it.", ] [targets.rust.rule_options.XSB077] From c00870adae68038ac4ed11f0a190a566ddb88a04 Mon Sep 17 00:00:00 2001 From: Kevin Longe <273747162+kvlonge@users.noreply.github.com> Date: Fri, 6 Nov 2026 04:01:25 +0000 Subject: [PATCH 3/4] feat: read microbatch coverage through the control API People with view on a target can read one model's recorded microbatch coverage: its completed segments, the holes between completed partitions, and segments a newer replay requirement still needs rebuilt at its version. Coverage is projected from the mirrored completion and replay-requirement facts of the model's newest relation and generation, with the same projection the executor uses; cursor bounds sort by value, so offset timestamps order by instant rather than text. Coverage is tracked only for concurrent microbatch models, so other models read as null. --- Cargo.lock | 2 + Cargo.toml | 2 +- .../orchestration/microbatch_coverage.json | 368 ++++++++++++++++++ clients/typescript/src/schema.d.ts | 90 +++++ crates/sqlbuild-orchestration/Cargo.toml | 2 + .../authentication/password_sessions.rs | 4 + .../_helpers/resources/person_maintenance.rs | 33 +- .../sqlbuild-orchestration/src/api/models.rs | 1 + .../src/store/schema.sql | 1 + .../warehouse_history/_helpers/coverage.rs | 331 ++++++++++++++++ .../warehouse_history/_helpers/maintenance.rs | 3 +- .../src/warehouse_history/_helpers/mod.rs | 1 + .../src/warehouse_history/main/mod.rs | 1 + .../main/read_person_microbatch_coverage.rs | 34 ++ .../src/warehouse_history/mod.rs | 2 +- .../src/warehouse_history/models.rs | 48 +++ .../tests/api/test_contracts.rs | 2 + .../tests/api/test_microbatch_coverage.rs | 169 ++++++++ .../tests/api/warehouse_history/helpers.rs | 5 + fensu.toml | 2 +- 20 files changed, 1095 insertions(+), 6 deletions(-) create mode 100644 benchmarks/duckshop/edge_cases/orchestration/microbatch_coverage.json create mode 100644 crates/sqlbuild-orchestration/src/warehouse_history/_helpers/coverage.rs create mode 100644 crates/sqlbuild-orchestration/src/warehouse_history/main/read_person_microbatch_coverage.rs create mode 100644 crates/sqlbuild-orchestration/tests/api/test_microbatch_coverage.rs diff --git a/Cargo.lock b/Cargo.lock index 6b1b21abed..26490db6e8 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -4290,6 +4290,8 @@ dependencies = [ "serde", "serde_json", "sqlbuild-core", + "sqlbuild-cursor", + "sqlbuild-model-config", "sqlbuild-recall", "sqlbuild-state", "sqlx-core", diff --git a/Cargo.toml b/Cargo.toml index d2927259d1..bb3bb388e2 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -42,8 +42,8 @@ order = [ "sqlbuild-config", "sqlbuild-host", "sqlbuild-recall", - "sqlbuild-orchestration", "sqlbuild-model-config", + "sqlbuild-orchestration", "sqlbuild-discovery", "sqlbuild-render", "sqlbuild-scopes", diff --git a/benchmarks/duckshop/edge_cases/orchestration/microbatch_coverage.json b/benchmarks/duckshop/edge_cases/orchestration/microbatch_coverage.json new file mode 100644 index 0000000000..7b2af242fa --- /dev/null +++ b/benchmarks/duckshop/edge_cases/orchestration/microbatch_coverage.json @@ -0,0 +1,368 @@ +{ + "microbatches": [ + { + "cursor_column": "ordered_at", + "cursor_type": "timestamp", + "batch_size": "1 hour", + "run_start": "2026-01-01 00:00:00+00", + "run_end": "2026-01-01T08:00:00+03:00:00", + "event_id": "orders-old-generation", + "created_at": "2026-01-01T00:30:00Z", + "observed_at": "2026-01-01T00:30:00Z", + "record_type": "partition_completion", + "partition_start": "2026-01-01T10:00:00+00:00", + "partition_end": "2026-01-01T11:00:00+00:00", + "execution_run_id": "run-0", + "model_version_hash": "orders-v1", + "physical_generation_id": "orders-generation-0" + }, + { + "cursor_column": "ordered_at", + "cursor_type": "timestamp", + "batch_size": "1 hour", + "run_start": "2026-01-01 00:00:00+00", + "run_end": "2026-01-01T08:00:00+03:00:00", + "event_id": "orders-other-scope", + "created_at": "2026-01-01T00:35:00Z", + "observed_at": "2026-01-01T00:35:00Z", + "record_type": "partition_completion", + "partition_start": "2026-01-01T12:00:00+00:00", + "partition_end": "2026-01-01T13:00:00+00:00", + "execution_run_id": "run-0", + "model_version_hash": "orders-v1", + "scope_key": "orders:main.orders_previous" + }, + { + "cursor_column": "ordered_at", + "cursor_type": "timestamp", + "batch_size": "1 hour", + "run_start": "2026-01-01 00:00:00+00", + "run_end": "2026-01-01T08:00:00+03:00:00", + "event_id": "orders-older-requirement", + "created_at": "2026-01-01T00:45:00Z", + "observed_at": "2026-01-01T00:45:00Z", + "record_type": "replay_requirement", + "partition_start": null, + "partition_end": null, + "completion_type": null, + "execution_run_id": "run-replay", + "model_version_hash": "orders-v9", + "replay_requirement_id": "orders-older-requirement", + "required_model_version_hash": "orders-v9", + "previous_model_version_hash": "orders-v8", + "replay_policy": "full" + }, + { + "cursor_column": "ordered_at", + "cursor_type": "timestamp", + "batch_size": "1 hour", + "run_start": "2026-01-01 00:00:00+00", + "run_end": "2026-01-01T08:00:00+03:00:00", + "event_id": "orders-batch-1", + "created_at": "2026-01-01T01:30:00Z", + "observed_at": "2026-01-01T01:30:00Z", + "record_type": "partition_completion", + "partition_start": "2026-01-01 00:00:00+00", + "partition_end": "2026-01-01T09:00:00+08:00", + "execution_run_id": "run-1", + "model_version_hash": "orders-v1" + }, + { + "cursor_column": "ordered_at", + "cursor_type": "timestamp", + "batch_size": "1 hour", + "run_start": "2026-01-01 00:00:00+00", + "run_end": "2026-01-01T08:00:00+03:00:00", + "event_id": "orders-batch-2-older", + "created_at": "2026-01-01T01:00:00Z", + "observed_at": "2026-01-01T01:00:00Z", + "record_type": "partition_completion", + "partition_start": "2026-01-01T01:00:00+00:00", + "partition_end": "2026-01-01T02:00:00+00:00", + "execution_run_id": "run-old", + "model_version_hash": "orders-v1" + }, + { + "cursor_column": "ordered_at", + "cursor_type": "timestamp", + "batch_size": "1 hour", + "run_start": "2026-01-01 00:00:00+00", + "run_end": "2026-01-01T08:00:00+03:00:00", + "event_id": "orders-batch-2", + "created_at": "2026-01-01T02:30:00Z", + "observed_at": "2026-01-01T02:30:00Z", + "record_type": "partition_completion", + "partition_start": "2026-01-01T09:00:00+08:00", + "partition_end": "2026-01-01T02:00:00+00:00", + "execution_run_id": "run-2", + "model_version_hash": "orders-v1" + }, + { + "cursor_column": "ordered_at", + "cursor_type": "timestamp", + "batch_size": "1 hour", + "run_start": "2026-01-01 00:00:00+00", + "run_end": "2026-01-01T08:00:00+03:00:00", + "event_id": "orders-batch-4", + "created_at": "2026-01-01T04:30:00Z", + "observed_at": "2026-01-01T04:30:00Z", + "record_type": "partition_completion", + "partition_start": "2026-01-01T03:00:00+00:00", + "partition_end": "2026-01-01T04:00:00+00:00", + "execution_run_id": "run-3", + "model_version_hash": "orders-v2" + }, + { + "cursor_column": "ordered_at", + "cursor_type": "timestamp", + "batch_size": "1 hour", + "run_start": "2026-01-01 00:00:00+00", + "run_end": "2026-01-01T08:00:00+03:00:00", + "event_id": "orders-batch-5", + "created_at": "2026-01-01T05:30:00Z", + "observed_at": "2026-01-01T05:30:00Z", + "record_type": "synthetic_completion", + "partition_start": "2026-01-01T04:00:00+00:00", + "partition_end": "2026-01-01T05:00:00+00:00", + "execution_run_id": "", + "fingerprint_status": "unknown", + "model_version_hash": null, + "definition_hash": null, + "synthetic_reason": "Retained rows", + "unaccounted_policy": "synthesize", + "observed_row_count": 12, + "completion_type": "initial" + }, + { + "cursor_column": "ordered_at", + "cursor_type": "timestamp", + "batch_size": "1 hour", + "run_start": "2026-01-01 00:00:00+00", + "run_end": "2026-01-01T08:00:00+03:00:00", + "event_id": "orders-batch-4-rerun", + "created_at": "2026-01-01T05:45:00Z", + "observed_at": "2026-01-01T05:45:00Z", + "record_type": "partition_completion", + "partition_start": "2026-01-01T03:00:00+00:00", + "partition_end": "2026-01-01T04:00:00+00:00", + "execution_run_id": "run-5", + "model_version_hash": "orders-v2" + }, + { + "cursor_column": "ordered_at", + "cursor_type": "timestamp", + "batch_size": "1 hour", + "run_start": "2026-01-01 00:00:00+00", + "run_end": "2026-01-01T08:00:00+03:00:00", + "event_id": "orders-batch-7", + "created_at": "2026-01-01T05:50:00Z", + "observed_at": "2026-01-01T05:50:00Z", + "record_type": "synthetic_completion", + "partition_start": "2026-01-01T06:00:00+00:00", + "partition_end": "2026-01-01T07:00:00+00:00", + "execution_run_id": "", + "fingerprint_status": "unknown", + "model_version_hash": null, + "definition_hash": null, + "synthetic_reason": "Retained rows", + "unaccounted_policy": "synthesize", + "observed_row_count": 12, + "completion_type": "initial" + }, + { + "cursor_column": "ordered_at", + "cursor_type": "timestamp", + "batch_size": "1 hour", + "run_start": "2026-01-01 00:00:00+00", + "run_end": "2026-01-01T08:00:00+03:00:00", + "event_id": "orders-requirement", + "created_at": "2026-01-01T06:30:00Z", + "observed_at": "2026-01-01T06:30:00Z", + "record_type": "replay_requirement", + "partition_start": null, + "partition_end": null, + "completion_type": null, + "execution_run_id": "run-replay", + "model_version_hash": "orders-v2", + "replay_requirement_id": "orders-requirement", + "required_model_version_hash": "orders-v2", + "previous_model_version_hash": "orders-v1", + "replay_policy": "full" + }, + { + "cursor_column": "ordered_at", + "cursor_type": "timestamp", + "batch_size": "1 hour", + "run_start": "2026-01-01 00:00:00+00", + "run_end": "2026-01-01T08:00:00+03:00:00", + "event_id": "customers-batch-1", + "created_at": "2026-01-01T01:30:00Z", + "observed_at": "2026-01-01T01:30:00Z", + "record_type": "partition_completion", + "partition_start": "2026-01-01T00:00:00+00:00", + "partition_end": "2026-01-01T01:00:00+00:00", + "execution_run_id": "run-1", + "model_version_hash": "customers-v1", + "model_name": "customers", + "scope_key": "customers:main.customers", + "target_name": "customers" + }, + { + "cursor_column": "event_id", + "cursor_type": "integer", + "batch_size": "1", + "run_start": "2", + "run_end": "11", + "event_id": "events-batch-2-older", + "created_at": "2026-01-01T00:30:00Z", + "observed_at": "2026-01-01T00:30:00Z", + "record_type": "partition_completion", + "partition_start": "2", + "partition_end": "5", + "execution_run_id": "run-old", + "model_version_hash": "events-v1", + "model_name": "events", + "scope_key": "events:main.events", + "target_name": "events", + "physical_generation_id": "events-generation-1" + }, + { + "cursor_column": "event_id", + "cursor_type": "integer", + "batch_size": "1", + "run_start": "2", + "run_end": "11", + "event_id": "events-batch-2", + "created_at": "2026-01-01T01:00:00Z", + "observed_at": "2026-01-01T01:00:00Z", + "record_type": "partition_completion", + "partition_start": "2", + "partition_end": "9", + "execution_run_id": "run-a", + "model_version_hash": "events-v1", + "model_name": "events", + "scope_key": "events:main.events", + "target_name": "events", + "physical_generation_id": "events-generation-1" + }, + { + "cursor_column": "event_id", + "cursor_type": "integer", + "batch_size": "1", + "run_start": "2", + "run_end": "11", + "event_id": "events-batch-9", + "created_at": "2026-01-01T02:00:00Z", + "observed_at": "2026-01-01T02:00:00Z", + "record_type": "partition_completion", + "partition_start": "9", + "partition_end": "10", + "execution_run_id": "run-b", + "model_version_hash": "events-v1", + "model_name": "events", + "scope_key": "events:main.events", + "target_name": "events", + "physical_generation_id": "events-generation-1" + }, + { + "cursor_column": "event_id", + "cursor_type": "integer", + "batch_size": "1", + "run_start": "2", + "run_end": "11", + "event_id": "events-batch-10", + "created_at": "2026-01-01T03:00:00Z", + "observed_at": "2026-01-01T03:00:00Z", + "record_type": "partition_completion", + "partition_start": "10", + "partition_end": "11", + "execution_run_id": "run-c", + "model_version_hash": "events-v1", + "model_name": "events", + "scope_key": "events:main.events", + "target_name": "events", + "physical_generation_id": "events-generation-1" + }, + { + "cursor_column": "ordered_at", + "cursor_type": "timestamp", + "batch_size": "1 hour", + "run_start": "2026-01-01T00:00:00+00:00", + "run_end": "2026-01-01T02:00:00+00:00", + "event_id": "payments-batch-0", + "created_at": "2026-01-01T00:30:00Z", + "observed_at": "2026-01-01T00:30:00Z", + "record_type": "partition_completion", + "partition_start": "2026-01-01T00:00:00+00:00", + "partition_end": "2026-01-01T01:00:00+00:00", + "execution_run_id": "run-p1", + "model_version_hash": "payments-v1", + "model_name": "payments", + "scope_key": "payments:main.payments", + "target_name": "payments", + "physical_generation_id": "payments-generation-1" + }, + { + "cursor_column": "ordered_at", + "cursor_type": "timestamp", + "batch_size": "1 hour", + "run_start": "2026-01-01T00:00:00+00:00", + "run_end": "2026-01-01T02:00:00+00:00", + "event_id": "payments-batch-1", + "created_at": "2026-01-01T00:40:00Z", + "observed_at": "2026-01-01T00:40:00Z", + "record_type": "partition_completion", + "partition_start": "2026-01-01T01:00:00+00:00", + "partition_end": "2026-01-01T02:00:00+00:00", + "execution_run_id": "run-p2", + "model_version_hash": "payments-v1", + "model_name": "payments", + "scope_key": "payments:main.payments", + "target_name": "payments", + "physical_generation_id": "payments-generation-1" + }, + { + "cursor_column": "ordered_at", + "cursor_type": "timestamp", + "batch_size": "1 hour", + "run_start": "2026-01-01T00:00:00+00:00", + "run_end": "2026-01-01T02:00:00+00:00", + "event_id": "payments-requirement", + "created_at": "2026-01-01T01:00:00Z", + "observed_at": "2026-01-01T01:00:00Z", + "record_type": "replay_requirement", + "partition_start": null, + "partition_end": null, + "completion_type": null, + "execution_run_id": "run-replay", + "model_version_hash": "payments-v2", + "replay_requirement_id": "payments-requirement", + "required_model_version_hash": "payments-v2", + "previous_model_version_hash": "payments-v1", + "replay_policy": "full", + "model_name": "payments", + "scope_key": "payments:main.payments", + "target_name": "payments", + "physical_generation_id": "payments-generation-1" + }, + { + "cursor_column": "ordered_at", + "cursor_type": "timestamp", + "batch_size": "1 hour", + "run_start": "2026-01-01T00:00:00+00:00", + "run_end": "2026-01-01T02:00:00+00:00", + "event_id": "payments-batch-0-v3", + "created_at": "2026-01-01T02:00:00Z", + "observed_at": "2026-01-01T02:00:00Z", + "record_type": "partition_completion", + "partition_start": "2026-01-01T00:00:00+00:00", + "partition_end": "2026-01-01T01:00:00+00:00", + "execution_run_id": "run-p3", + "model_version_hash": "payments-v3", + "model_name": "payments", + "scope_key": "payments:main.payments", + "target_name": "payments", + "physical_generation_id": "payments-generation-1" + } + ] +} diff --git a/clients/typescript/src/schema.d.ts b/clients/typescript/src/schema.d.ts index ae6e30de48..de824cbd63 100644 --- a/clients/typescript/src/schema.d.ts +++ b/clients/typescript/src/schema.d.ts @@ -338,6 +338,22 @@ export interface paths { patch?: never; trace?: never; }; + "/api/control/locations/{location_id}/targets/{target}/microbatches/{model}": { + parameters: { + query?: never; + header?: never; + path?: never; + cookie?: never; + }; + get: operations["readPersonMicrobatchCoverage"]; + put?: never; + post?: never; + delete?: never; + options?: never; + head?: never; + patch?: never; + trace?: never; + }; "/api/control/locations/{location_id}/targets/{target}/node-status": { parameters: { query?: never; @@ -824,6 +840,11 @@ export interface components { /** Format: uuid */ request_id: string; }; + /** + * @description How one recorded stretch of a microbatch model's cursor stands. + * @enum {string} + */ + BatchState: "complete" | "gap" | "replay_required"; /** @enum {string} */ BuiltinRole: "viewer" | "operator" | "developer" | "admin"; /** @description Browser cancellation cannot supply an actor, claim or lifecycle transition. */ @@ -934,6 +955,14 @@ export interface components { /** Format: uuid */ request_id: string; }; + /** @description One stretch of the cursor, as raw bound texts. */ + CoverageBatch: { + end: string; + /** @description The run that executed the latest completion here; absent for gaps and external completions. */ + run_id?: string | null; + start: string; + state: components["schemas"]["BatchState"]; + }; /** @description The current successful snapshot, read with a consistent generation and change set. */ CurrentSnapshot: { changes: components["schemas"]["SnapshotChange"][]; @@ -1162,6 +1191,25 @@ export interface components { /** Format: uuid */ scan_id: string; }; + /** @description Recorded coverage of one model's current relation; tracked only for concurrent microbatch models. */ + MicrobatchCoverage: { + batch_size: string; + /** @description The newest batches, oldest first. */ + batches: components["schemas"]["CoverageBatch"][]; + /** @description Whether older batches were left out of `batches`. */ + batches_truncated: boolean; + /** @description End of the completed range that starts at `earliest` without a hole. */ + contiguous_frontier?: string | null; + cursor_column: string; + cursor_type: string; + /** @description Start of the earliest completed partition. */ + earliest: string; + /** @description Holes between completed partitions, oldest first. */ + known_gaps: components["schemas"]["CoverageBatch"][]; + model: string; + physical_generation_id: string; + target_name: string; + }; MutationResponse: { applied: boolean; }; @@ -2682,6 +2730,48 @@ export interface operations { }; }; }; + readPersonMicrobatchCoverage: { + parameters: { + query?: { + /** @description Maximum newest batches; defaults to 200, capped at 1000 */ + limit?: number; + }; + header?: never; + path: { + location_id: string; + target: string; + model: string; + }; + cookie?: never; + }; + requestBody?: never; + responses: { + /** @description Null when the model has no tracked coverage */ + 200: { + headers: { + [name: string]: unknown; + }; + content: { + "application/json": null | components["schemas"]["MicrobatchCoverage"]; + }; + }; + /** @description Session rejected */ + 401: { + headers: { + [name: string]: unknown; + }; + content?: never; + }; + 403: { + headers: { + [name: string]: unknown; + }; + content: { + "application/json": components["schemas"]["ApiErrorBody"]; + }; + }; + }; + }; readPersonNodeStatus: { parameters: { query: { diff --git a/crates/sqlbuild-orchestration/Cargo.toml b/crates/sqlbuild-orchestration/Cargo.toml index a2cbd1322e..f680f7a1ea 100644 --- a/crates/sqlbuild-orchestration/Cargo.toml +++ b/crates/sqlbuild-orchestration/Cargo.toml @@ -22,6 +22,8 @@ rusqlite.workspace = true serde.workspace = true serde_json.workspace = true sqlbuild-core.workspace = true +sqlbuild-cursor.workspace = true +sqlbuild-model-config.workspace = true sqlbuild-recall.workspace = true sqlbuild-state.workspace = true sqlx-core.workspace = true 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 0a955d6235..0e14f0579e 100644 --- a/crates/sqlbuild-orchestration/src/api/_helpers/authentication/password_sessions.rs +++ b/crates/sqlbuild-orchestration/src/api/_helpers/authentication/password_sessions.rs @@ -129,6 +129,10 @@ pub(crate) fn person_routes() -> Router { "/api/control/locations/{location_id}/targets/{target}/maintenance", get(crate::api::_helpers::resources::person_maintenance::read), ) + .route( + "/api/control/locations/{location_id}/targets/{target}/microbatches/{model}", + get(crate::api::_helpers::resources::person_maintenance::coverage), + ) .layer(DefaultBodyLimit::max(PASSWORD_BODY_BYTES)) } diff --git a/crates/sqlbuild-orchestration/src/api/_helpers/resources/person_maintenance.rs b/crates/sqlbuild-orchestration/src/api/_helpers/resources/person_maintenance.rs index 1c704c0b52..2793482bce 100644 --- a/crates/sqlbuild-orchestration/src/api/_helpers/resources/person_maintenance.rs +++ b/crates/sqlbuild-orchestration/src/api/_helpers/resources/person_maintenance.rs @@ -1,4 +1,4 @@ -//! Maintenance history reads use live person grants over the whole target. +//! Maintenance history and microbatch coverage reads use live person grants over the whole target. use axum::extract::{Path, Query, State}; use axum::{Extension, Json}; @@ -8,7 +8,10 @@ use crate::api::errors::ApiError; use crate::credentials::models::{AuditContext, AuthenticatedPerson, DefaultRole}; use crate::store::models::ControlStore; use crate::warehouse_history::main::read_person_maintenance::read_person_maintenance; -use crate::warehouse_history::models::{MaintenanceHistory, MaintenanceQuery}; +use crate::warehouse_history::main::read_person_microbatch_coverage::read_person_microbatch_coverage; +use crate::warehouse_history::models::{ + CoverageQuery, MaintenanceHistory, MaintenanceQuery, MicrobatchCoverage, +}; #[utoipa::path(get, path = "/api/control/locations/{location_id}/targets/{target}/maintenance", operation_id = "readPersonMaintenance", params(("location_id" = Uuid, Path), ("target" = String, Path), @@ -32,3 +35,29 @@ pub(crate) async fn read( .await?, )) } + +#[utoipa::path(get, path = "/api/control/locations/{location_id}/targets/{target}/microbatches/{model}", operation_id = "readPersonMicrobatchCoverage", + params(("location_id" = Uuid, Path), ("target" = String, Path), ("model" = String, Path), + ("limit" = Option, Query, description = "Maximum newest batches; defaults to 200, capped at 1000")), + responses((status = 200, body = Option, description = "Null when the model has no tracked coverage"), + (status = 401, description = "Session rejected"), + (status = 403, body = crate::api::models::ApiErrorBody)), security(("session" = []), ("cliSession" = [])))] +pub(crate) async fn coverage( + (State(store), Extension(context)): (State, Extension), + Extension(default_role): Extension, + (Path((location_id, target, model)), Query(options)): ( + Path<(Uuid, String, String)>, + Query, + ), + Extension(AuthenticatedPerson(principal, _)): Extension, +) -> Result>, ApiError> { + Ok(Json( + read_person_microbatch_coverage( + &store, + (&context, principal), + ((location_id, &target), (&model, &options)), + default_role, + ) + .await?, + )) +} diff --git a/crates/sqlbuild-orchestration/src/api/models.rs b/crates/sqlbuild-orchestration/src/api/models.rs index 0174bb33c6..0b13d5a793 100644 --- a/crates/sqlbuild-orchestration/src/api/models.rs +++ b/crates/sqlbuild-orchestration/src/api/models.rs @@ -29,6 +29,7 @@ use crate::api::_helpers::resources::{ person_checks::sql, person_checks::evidence, person_maintenance::read, + person_maintenance::coverage, person_location_work::request, person_location_work::result, person_node_sql::read, diff --git a/crates/sqlbuild-orchestration/src/store/schema.sql b/crates/sqlbuild-orchestration/src/store/schema.sql index 537f4ccdd4..a7e4973c6d 100644 --- a/crates/sqlbuild-orchestration/src/store/schema.sql +++ b/crates/sqlbuild-orchestration/src/store/schema.sql @@ -464,6 +464,7 @@ CREATE TABLE sqb_warehouse_microbatches ( scan_id {uuid} ); CREATE INDEX sqb_microbatches_namespace ON sqb_warehouse_microbatches (location_id,target,state_database,state_schema,observed_at,event_id); +CREATE INDEX sqb_microbatches_model ON sqb_warehouse_microbatches (location_id,target,model_name); CREATE INDEX sqb_freshness_source_latest ON sqb_warehouse_freshness ( location_id, target, (record_json ->> 'source_name'), (record_json ->> 'target_database'), (record_json ->> 'target_schema'), diff --git a/crates/sqlbuild-orchestration/src/warehouse_history/_helpers/coverage.rs b/crates/sqlbuild-orchestration/src/warehouse_history/_helpers/coverage.rs new file mode 100644 index 0000000000..83ca28dac9 --- /dev/null +++ b/crates/sqlbuild-orchestration/src/warehouse_history/_helpers/coverage.rs @@ -0,0 +1,331 @@ +//! Microbatch coverage projected from the mirrored completion and replay-requirement facts. + +use std::cmp::Ordering; +use std::collections::HashMap; + +use chrono::{DateTime, Datelike, NaiveDateTime, Timelike}; +use sqlbuild_cursor::algebra::main::normalize_timestamp_text::normalize_timestamp_text; +use sqlbuild_cursor::algebra::main::parse_integer_text::parse_integer_text; +use sqlbuild_cursor::algebra::models::{CivilDate, LocalDateTime, SortKey}; +use sqlbuild_model_config::model_validation::main::parse_iso_timestamp::parse_iso_timestamp; +use sqlbuild_state::microbatches::main::project_coverage::project_coverage; +use sqlbuild_state::microbatches::models::{ + BoundInterval, CompletionFact, CoverageProjection, CursorBound, TextInterval, +}; +use uuid::Uuid; + +use crate::store::errors::ControlError; +use crate::store::models::{ControlTransaction, Parameter}; +use crate::warehouse_history::_helpers::maintenance::records; +use crate::warehouse_history::models::{BatchState, CoverageBatch, MicrobatchCoverage}; +use crate::warehouse_metadata::models::WarehouseMicrobatchRecord; + +const ORDINARY_COMPLETION: &str = "partition_completion"; +const SYNTHETIC_COMPLETION: &str = "synthetic_completion"; +const REPLAY_REQUIREMENT: &str = "replay_requirement"; +const UNKNOWN_FINGERPRINT: &str = "unknown"; +const INTEGER_CURSOR: &str = "integer"; +/// The `datetime.fromisoformat` rules cursor bounds are validated with. +const ISOFORMAT_PYTHON_MINOR: u8 = 12; + +/// The key a cursor bound sorts by: the integer, or the wall time of a naive value and the UTC instant of an aware one. +fn sort_key(text: &str, cursor_type: &str) -> Option { + if cursor_type == INTEGER_CURSOR { + return parse_integer_text(text).map(SortKey::Integer); + } + let (micros, _) = parse_iso_timestamp(&normalize_timestamp_text(text), ISOFORMAT_PYTHON_MINOR)?; + let wall: NaiveDateTime = DateTime::from_timestamp_micros(micros)?.naive_utc(); + Some(SortKey::Moment(LocalDateTime { + date: CivilDate { + year: i64::from(wall.year()), + month: wall.month(), + day: wall.day(), + }, + hour: wall.hour(), + minute: wall.minute(), + second: wall.second(), + microsecond: wall.nanosecond() / 1000, + })) +} + +fn order(left: &CursorBound, right: &CursorBound) -> Ordering { + match (&left.key, &right.key) { + (SortKey::Integer(left), SortKey::Integer(right)) => left.cmp(right), + (SortKey::Moment(left), SortKey::Moment(right)) => left.cmp(right), + _ => Ordering::Equal, + } +} + +/// Bound texts of one cursor type, each parsed once. +struct Bounds<'a> { + cursor_type: &'a str, + keys: HashMap, +} + +impl Bounds<'_> { + fn bound(&mut self, text: &str) -> Result { + if let Some(key) = self.keys.get(text) { + return Ok(CursorBound { + text: text.into(), + key: *key, + }); + } + let key: SortKey = sort_key(text, self.cursor_type).ok_or_else(|| { + ControlError::Schema(format!( + "stored microbatch bound is not a {} cursor value", + self.cursor_type + )) + })?; + self.keys.insert(text.into(), key); + Ok(CursorBound { + text: text.into(), + key, + }) + } + + fn interval(&mut self, interval: &TextInterval) -> Result { + Ok(BoundInterval { + start: self.bound(&interval.start)?, + end: self.bound(&interval.end)?, + }) + } +} + +fn newest<'a>( + records: impl Iterator, +) -> Option<&'a WarehouseMicrobatchRecord> { + records.max_by(|left, right| { + (left.created_at, &left.event_id).cmp(&(right.created_at, &right.event_id)) + }) +} + +/// Completion facts of the current relation and the records they came from, in the same order. +fn completions<'a>( + scope: &[&'a WarehouseMicrobatchRecord], + bounds: &mut Bounds<'_>, +) -> Result<(Vec, Vec<&'a WarehouseMicrobatchRecord>), ControlError> { + let mut facts: Vec = Vec::new(); + let mut sources: Vec<&WarehouseMicrobatchRecord> = Vec::new(); + for record in scope { + let (Some(start), Some(end)) = (&record.partition_start, &record.partition_end) else { + continue; + }; + if ![ORDINARY_COMPLETION, SYNTHETIC_COMPLETION].contains(&record.record_type.as_str()) { + continue; + } + facts.push(CompletionFact { + interval: BoundInterval { + start: bounds.bound(start)?, + end: bounds.bound(end)?, + }, + ordinary: record.record_type == ORDINARY_COMPLETION, + unknown_fingerprint: record.fingerprint_status == UNKNOWN_FINGERPRINT, + created: (record.created_at - DateTime::UNIX_EPOCH).as_seconds_f64(), + event_id: record.event_id.clone(), + }); + sources.push(record); + } + Ok((facts, sources)) +} + +/// The newest replay requirement's range and required version, unless a newer completion at a third version ended it, as the executor decides. +fn replay_range( + scope: &[&WarehouseMicrobatchRecord], + bounds: &mut Bounds<'_>, +) -> Result)>, ControlError> { + let Some(requirement) = newest( + scope + .iter() + .copied() + .filter(|record| record.record_type == REPLAY_REQUIREMENT), + ) else { + return Ok(None); + }; + let superseded: bool = scope.iter().any(|record| { + record.record_type == ORDINARY_COMPLETION + && (record.created_at, &record.event_id) + > (requirement.created_at, &requirement.event_id) + && record.model_version_hash.as_ref().is_some_and(|version| { + Some(version) != requirement.required_model_version_hash.as_ref() + }) + }); + if superseded { + return Ok(None); + } + Ok(Some(( + bounds.interval(&TextInterval { + start: requirement.run_start.clone(), + end: requirement.run_end.clone(), + })?, + requirement.required_model_version_hash.clone(), + ))) +} + +fn overlaps(left: &BoundInterval, right: &BoundInterval) -> bool { + order(&left.start, &right.end).is_lt() && order(&right.start, &left.end).is_lt() +} + +/// One stretch of the cursor and the completion that accounts for it, if any. +struct Piece { + interval: BoundInterval, + state: BatchState, + completion: Option, +} + +/// Adjacent stretches of one completion, or of one hole, read as one batch; empty stretches are dropped. +fn merged(mut pieces: Vec) -> Vec { + pieces.sort_by(|left, right| order(&left.interval.start, &right.interval.start)); + let mut kept: Vec = Vec::with_capacity(pieces.len()); + for piece in pieces { + if order(&piece.interval.start, &piece.interval.end).is_ge() { + continue; + } + match kept.last_mut() { + Some(last) + if last.completion == piece.completion + && last.state == piece.state + && order(&last.interval.end, &piece.interval.start).is_eq() => + { + last.interval.end = piece.interval.end; + } + _ => kept.push(piece), + } + } + kept +} + +fn batch(piece: &Piece, sources: &[&WarehouseMicrobatchRecord]) -> CoverageBatch { + CoverageBatch { + start: piece.interval.start.text.clone(), + end: piece.interval.end.text.clone(), + state: piece.state, + run_id: piece + .completion + .map(|completion| sources[completion].execution_run_id.clone()) + .filter(|run| !run.is_empty()), + } +} + +/// Completed segments, spelled as their completion recorded them where they share its bounds, and the holes between them, oldest first; and the holes alone. +fn batches( + projection: &CoverageProjection, + sources: &[&WarehouseMicrobatchRecord], + (bounds, replay): (&mut Bounds<'_>, Option<&(BoundInterval, Option)>), +) -> Result<(Vec, Vec), ControlError> { + let mut completed: Vec = Vec::new(); + for segment in &projection.segments { + let mut interval: BoundInterval = bounds.interval(&segment.interval)?; + let source: &WarehouseMicrobatchRecord = sources[segment.completion]; + for (recorded, bound) in [ + (&source.partition_start, &mut interval.start), + (&source.partition_end, &mut interval.end), + ] { + let Some(recorded) = recorded else { continue }; + let recorded: CursorBound = bounds.bound(recorded)?; + if order(&recorded, bound).is_eq() { + *bound = recorded; + } + } + let stale: bool = replay.is_some_and(|(range, required)| { + overlaps(&interval, range) + && source.fingerprint_status != UNKNOWN_FINGERPRINT + && source.model_version_hash != *required + }); + completed.push(Piece { + interval, + state: if stale { + BatchState::ReplayRequired + } else { + BatchState::Complete + }, + completion: Some(segment.completion), + }); + } + let mut holes: Vec = Vec::new(); + for hole in projection + .known_missing + .iter() + .chain(&projection.unaccounted) + { + holes.push(Piece { + interval: bounds.interval(hole)?, + state: BatchState::Gap, + completion: None, + }); + } + let holes: Vec = merged(holes); + let gaps: Vec = holes.iter().map(|piece| batch(piece, sources)).collect(); + let mut every: Vec = merged(completed); + every.extend(holes); + every.sort_by(|left, right| order(&left.interval.start, &right.interval.start)); + Ok(( + every.iter().map(|piece| batch(piece, sources)).collect(), + gaps, + )) +} + +/// Coverage of the model's newest recorded relation and generation, or `None` when none is tracked. +pub(crate) async fn microbatch_coverage( + transaction: ControlTransaction, + (location_id, target): (Uuid, &str), + (model, limit): (&str, u32), +) -> Result<(ControlTransaction, Option), ControlError> { + let (transaction, history): (ControlTransaction, Vec) = records(transaction, + "SELECT record_json AS record FROM sqb_warehouse_microbatches WHERE location_id=$1 AND target=$2 AND model_name=$3", + vec![Parameter::Id(location_id), Parameter::Text(target.into()), Parameter::Text(model.into())]).await?; + let Some(current) = newest(history.iter()) else { + return Ok((transaction, None)); + }; + let scope: Vec<&WarehouseMicrobatchRecord> = history + .iter() + .filter(|record| { + record.scope_key == current.scope_key + && record.physical_generation_id == current.physical_generation_id + }) + .collect(); + let mut bounds = Bounds { + cursor_type: ¤t.cursor_type, + keys: HashMap::new(), + }; + let (facts, sources) = completions(&scope, &mut bounds)?; + let (Some(earliest), Some(latest)) = ( + facts + .iter() + .map(|fact| &fact.interval.start) + .min_by(|left, right| order(left, right)), + facts + .iter() + .map(|fact| &fact.interval.end) + .max_by(|left, right| order(left, right)), + ) else { + return Ok((transaction, None)); + }; + let projection: CoverageProjection = project_coverage( + &facts, + &[BoundInterval { + start: earliest.clone(), + end: latest.clone(), + }], + ); + let replay: Option<(BoundInterval, Option)> = replay_range(&scope, &mut bounds)?; + let (mut recent, known_gaps) = batches(&projection, &sources, (&mut bounds, replay.as_ref()))?; + let keep: usize = usize::try_from(limit.clamp(1, 1000)).unwrap_or(usize::MAX); + let dropped: usize = recent.len().saturating_sub(keep); + recent.drain(..dropped); + Ok(( + transaction, + Some(MicrobatchCoverage { + model: model.into(), + target_name: current.target_name.clone(), + physical_generation_id: current.physical_generation_id.clone(), + cursor_column: current.cursor_column.clone(), + cursor_type: current.cursor_type.clone(), + batch_size: current.batch_size.clone(), + earliest: earliest.text.clone(), + contiguous_frontier: projection.contiguous_frontier.clone(), + known_gaps, + batches: recent, + batches_truncated: dropped > 0, + }), + )) +} diff --git a/crates/sqlbuild-orchestration/src/warehouse_history/_helpers/maintenance.rs b/crates/sqlbuild-orchestration/src/warehouse_history/_helpers/maintenance.rs index 279f7a16dc..6c52ab9094 100644 --- a/crates/sqlbuild-orchestration/src/warehouse_history/_helpers/maintenance.rs +++ b/crates/sqlbuild-orchestration/src/warehouse_history/_helpers/maintenance.rs @@ -17,7 +17,8 @@ struct Recorded { record: T, } -async fn records( +/// The `record` column of every row `sql` selects. +pub(crate) async fn records( transaction: ControlTransaction, sql: &str, parameters: Vec, diff --git a/crates/sqlbuild-orchestration/src/warehouse_history/_helpers/mod.rs b/crates/sqlbuild-orchestration/src/warehouse_history/_helpers/mod.rs index e0260321e5..51c3c3a651 100644 --- a/crates/sqlbuild-orchestration/src/warehouse_history/_helpers/mod.rs +++ b/crates/sqlbuild-orchestration/src/warehouse_history/_helpers/mod.rs @@ -1,2 +1,3 @@ pub(crate) mod checks; +pub(crate) mod coverage; pub(crate) mod maintenance; diff --git a/crates/sqlbuild-orchestration/src/warehouse_history/main/mod.rs b/crates/sqlbuild-orchestration/src/warehouse_history/main/mod.rs index 5a314e91ca..55736dc22c 100644 --- a/crates/sqlbuild-orchestration/src/warehouse_history/main/mod.rs +++ b/crates/sqlbuild-orchestration/src/warehouse_history/main/mod.rs @@ -3,3 +3,4 @@ pub mod read_person_check_evidence; pub mod read_person_check_sql; pub mod read_person_checks; pub mod read_person_maintenance; +pub mod read_person_microbatch_coverage; diff --git a/crates/sqlbuild-orchestration/src/warehouse_history/main/read_person_microbatch_coverage.rs b/crates/sqlbuild-orchestration/src/warehouse_history/main/read_person_microbatch_coverage.rs new file mode 100644 index 0000000000..a89c679521 --- /dev/null +++ b/crates/sqlbuild-orchestration/src/warehouse_history/main/read_person_microbatch_coverage.rs @@ -0,0 +1,34 @@ +//! Live view permission over the whole target and one model's recorded microbatch coverage. + +use uuid::Uuid; + +use crate::authorization::main::authorize_listing::authorize_listing; +use crate::credentials::models::{AuditContext, DefaultRole}; +use crate::store::errors::ControlError; +use crate::store::main::{begin::begin, commit::commit}; +use crate::store::models::{ControlStore, TransactionMode}; +use crate::warehouse_history::_helpers::coverage::microbatch_coverage; +use crate::warehouse_history::models::{CoverageQuery, MicrobatchCoverage}; + +pub async fn read_person_microbatch_coverage( + store: &ControlStore, + credentials: (&AuditContext, Uuid), + ((location_id, target), (model, options)): ((Uuid, &str), (&str, &CoverageQuery)), + default_role: DefaultRole, +) -> Result, ControlError> { + let (transaction, _) = authorize_listing( + begin(store, TransactionMode::Write).await?, + credentials, + (location_id, target), + default_role, + ) + .await?; + let (transaction, coverage) = microbatch_coverage( + transaction, + (location_id, target), + (model, options.limit.unwrap_or(200)), + ) + .await?; + commit(transaction).await?; + Ok(coverage) +} diff --git a/crates/sqlbuild-orchestration/src/warehouse_history/mod.rs b/crates/sqlbuild-orchestration/src/warehouse_history/mod.rs index 7488923fa9..e87cb64e62 100644 --- a/crates/sqlbuild-orchestration/src/warehouse_history/mod.rs +++ b/crates/sqlbuild-orchestration/src/warehouse_history/mod.rs @@ -1,4 +1,4 @@ -//! Person reads of warehouse-owned history mirrored into the control store: checks and maintenance. +//! Person reads of warehouse-owned history mirrored into the control store: checks, maintenance and microbatch coverage. pub(crate) mod _helpers; pub mod main; diff --git a/crates/sqlbuild-orchestration/src/warehouse_history/models.rs b/crates/sqlbuild-orchestration/src/warehouse_history/models.rs index cca287650b..57c26430cb 100644 --- a/crates/sqlbuild-orchestration/src/warehouse_history/models.rs +++ b/crates/sqlbuild-orchestration/src/warehouse_history/models.rs @@ -112,3 +112,51 @@ pub struct MaintenanceQuery { pub node: Option, pub limit: Option, } + +/// How one recorded stretch of a microbatch model's cursor stands. +#[derive(Clone, Copy, Debug, PartialEq, Eq, Serialize, Deserialize, utoipa::ToSchema)] +#[serde(rename_all = "snake_case")] +pub enum BatchState { + Complete, + /// A hole between completed partitions. + Gap, + /// Completed at another version than the newest replay requirement covering it. + ReplayRequired, +} + +/// One stretch of the cursor, as raw bound texts. +#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize, utoipa::ToSchema)] +pub struct CoverageBatch { + pub start: String, + pub end: String, + pub state: BatchState, + /// The run that executed the latest completion here; absent for gaps and external completions. + pub run_id: Option, +} + +/// Recorded coverage of one model's current relation; tracked only for concurrent microbatch models. +#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize, utoipa::ToSchema)] +pub struct MicrobatchCoverage { + pub model: String, + pub target_name: String, + pub physical_generation_id: String, + pub cursor_column: String, + pub cursor_type: String, + pub batch_size: String, + /// Start of the earliest completed partition. + pub earliest: String, + /// End of the completed range that starts at `earliest` without a hole. + pub contiguous_frontier: Option, + /// Holes between completed partitions, oldest first. + pub known_gaps: Vec, + /// The newest batches, oldest first. + pub batches: Vec, + /// Whether older batches were left out of `batches`. + pub batches_truncated: bool, +} + +#[derive(Clone, Debug, Deserialize)] +#[serde(deny_unknown_fields)] +pub struct CoverageQuery { + pub limit: Option, +} diff --git a/crates/sqlbuild-orchestration/tests/api/test_contracts.rs b/crates/sqlbuild-orchestration/tests/api/test_contracts.rs index 25527b6ad0..4d798f30a5 100644 --- a/crates/sqlbuild-orchestration/tests/api/test_contracts.rs +++ b/crates/sqlbuild-orchestration/tests/api/test_contracts.rs @@ -50,6 +50,8 @@ mod test_location_health; mod test_location_work; #[path = "test_maintenance.rs"] mod test_maintenance; +#[path = "test_microbatch_coverage.rs"] +mod test_microbatch_coverage; #[path = "test_oidc.rs"] mod test_oidc; #[path = "test_oidc_groups.rs"] diff --git a/crates/sqlbuild-orchestration/tests/api/test_microbatch_coverage.rs b/crates/sqlbuild-orchestration/tests/api/test_microbatch_coverage.rs new file mode 100644 index 0000000000..79bffe3bbc --- /dev/null +++ b/crates/sqlbuild-orchestration/tests/api/test_microbatch_coverage.rs @@ -0,0 +1,169 @@ +//! Microbatch coverage through the public API: completed segments, holes and stale versions. + +use reqwest::StatusCode; +use serde_json::{Value, json}; + +use crate::api::tests::helpers::stop_api; +use crate::api::tests::test_types::WarehouseMetadataTestCase; +use crate::api::tests::warehouse_history_helpers::{ + batch, coverage_batch, fixture, get, login, records, seed_target, start_password_api, +}; +use crate::store::tests::store_helpers::{close_test_store, open_test_store, run}; +use crate::store::tests::store_test_types::Backend; +use crate::warehouse_metadata::main::ingest_microbatches::ingest_microbatches; + +const HOUR_0: &str = "2026-01-01 00:00:00+00"; +const HOUR_1: &str = "2026-01-01T09:00:00+08:00"; +const HOUR_2: &str = "2026-01-01T02:00:00+00:00"; +const HOUR_3: &str = "2026-01-01T03:00:00+00:00"; +const HOUR_4: &str = "2026-01-01T04:00:00+00:00"; +const HOUR_5: &str = "2026-01-01T05:00:00+00:00"; +const HOUR_6: &str = "2026-01-01T06:00:00+00:00"; +const HOUR_7: &str = "2026-01-01T07:00:00+00:00"; + +#[test] +fn given_microbatch_history_when_people_read_coverage_then_holes_and_stale_versions_are_marked() { + let test_cases = [ + WarehouseMetadataTestCase { + description: "SQLite microbatch coverage", + backend: Backend::Sqlite, + expected_observations: 20, + }, + WarehouseMetadataTestCase { + description: "PostgreSQL microbatch coverage", + backend: Backend::Postgres, + expected_observations: 20, + }, + ]; + for test_case in test_cases { + run(async { + let fixture_store = open_test_store(test_case.backend).await.expect("store"); + let store = &fixture_store.store; + let (location, session_id) = seed_target(store).await; + let history = records( + "microbatch_history.json", + &fixture("microbatch_coverage.json")["microbatches"], + ); + assert_eq!(history.len(), test_case.expected_observations); + ingest_microbatches( + store, + (location, "development"), + &batch(session_id, "microbatches", history), + ) + .await + .expect("real microbatch ingestion"); + + let (api, client) = start_password_api(store).await; + let viewer = login(&client, &api.url, "orders-viewer").await; + let path = + format!("/api/control/locations/{location}/targets/development/microbatches"); + let read = |model: &str| format!("{path}/{model}"); + + let (status, orders) = get(&client, (&api.url, &viewer), &read("orders")).await; + assert_eq!(status, StatusCode::OK, "{}", test_case.description); + let early_gap = coverage_batch(HOUR_2, HOUR_3, "gap", Value::Null); + let late_gap = coverage_batch(HOUR_5, HOUR_6, "gap", Value::Null); + assert_eq!( + orders, + json!({ + "model": "orders", + "target_name": "orders_\u{1f9fe}", + "physical_generation_id": "orders-generation-1", + "cursor_column": "ordered_at", + "cursor_type": "timestamp", + "batch_size": "1 hour", + "earliest": HOUR_0, + "contiguous_frontier": HOUR_2, + "known_gaps": [early_gap, late_gap], + "batches": [ + coverage_batch(HOUR_0, HOUR_1, "replay_required", json!("run-1")), + coverage_batch(HOUR_1, HOUR_2, "replay_required", json!("run-2")), + early_gap, + coverage_batch(HOUR_3, HOUR_4, "complete", json!("run-5")), + coverage_batch(HOUR_4, HOUR_5, "complete", Value::Null), + late_gap, + coverage_batch(HOUR_6, HOUR_7, "complete", Value::Null), + ], + "batches_truncated": false, + }), + "bounds sort by instant; the newest completion and requirement win; other generations, scopes and models never appear" + ); + + let (_, newest) = get( + &client, + (&api.url, &viewer), + &format!("{}?limit=2", read("orders")), + ) + .await; + assert_eq!( + (&newest["batches"], &newest["batches_truncated"]), + ( + &json!([ + late_gap, + coverage_batch(HOUR_6, HOUR_7, "complete", Value::Null) + ]), + &json!(true) + ) + ); + assert_eq!( + newest["known_gaps"], + json!([early_gap, late_gap]), + "gaps are never truncated" + ); + + let (_, events) = get(&client, (&api.url, &viewer), &read("events")).await; + assert_eq!( + ( + &events["earliest"], + &events["contiguous_frontier"], + &events["batches"] + ), + ( + &json!("2"), + &json!("11"), + &json!([ + coverage_batch("2", "9", "complete", json!("run-a")), + coverage_batch("9", "10", "complete", json!("run-b")), + coverage_batch("10", "11", "complete", json!("run-c")), + ]) + ), + "integer bounds sort by value, not text" + ); + + let (_, payments) = get(&client, (&api.url, &viewer), &read("payments")).await; + assert_eq!( + payments["batches"], + json!([ + coverage_batch( + "2026-01-01T00:00:00+00:00", + "2026-01-01T01:00:00+00:00", + "complete", + json!("run-p3") + ), + coverage_batch( + "2026-01-01T01:00:00+00:00", + "2026-01-01T02:00:00+00:00", + "complete", + json!("run-p2") + ), + ]), + "a completion at a third version ends the requirement, as the executor decides" + ); + + let (status, untracked) = get(&client, (&api.url, &viewer), &read("inventory")).await; + assert_eq!((status, untracked), (StatusCode::OK, Value::Null)); + assert_eq!( + get( + &client, + (&api.url, &viewer), + &read("orders").replace("development", "preview") + ) + .await + .0, + StatusCode::FORBIDDEN + ); + stop_api(api).await; + close_test_store(fixture_store).await; + }); + } +} diff --git a/crates/sqlbuild-orchestration/tests/api/warehouse_history/helpers.rs b/crates/sqlbuild-orchestration/tests/api/warehouse_history/helpers.rs index b7679b1b7d..abe809c055 100644 --- a/crates/sqlbuild-orchestration/tests/api/warehouse_history/helpers.rs +++ b/crates/sqlbuild-orchestration/tests/api/warehouse_history/helpers.rs @@ -212,3 +212,8 @@ pub(crate) fn ids(history: &Value, kind: &str) -> Vec { .map(|record| record["event_id"].as_str().expect("event").to_owned()) .collect() } + +/// One microbatch coverage batch as the API returns it. +pub(crate) fn coverage_batch(start: &str, end: &str, state: &str, run_id: Value) -> Value { + json!({"start": start, "end": end, "state": state, "run_id": run_id}) +} diff --git a/fensu.toml b/fensu.toml index b36a40292e..54ddd88f0a 100644 --- a/fensu.toml +++ b/fensu.toml @@ -1371,7 +1371,7 @@ paths = ["crates/sqlbuild-orchestration/src/api/_helpers/resources/administrator reason = "Each function is one utoipa-annotated axum route: OpenAPI needs one annotated handler per operation, and each body only checks the caller's scope and calls one store operation. Scope checks and store work live in their shared owners." [[dupes.allowlist]] -paths = ["crates/sqlbuild-orchestration/src/authorization/main/read_security_audit.rs", "crates/sqlbuild-orchestration/src/credentials/_helpers/active_session.rs", "crates/sqlbuild-orchestration/src/declarations/main/list_triggers.rs", "crates/sqlbuild-orchestration/src/platform/main/read_location_work.rs", "crates/sqlbuild-orchestration/src/runs/main/list_person_runs.rs", "crates/sqlbuild-orchestration/src/runs/main/list_runs.rs", "crates/sqlbuild-orchestration/src/runs/_helpers/lifecycle/reads.rs", "crates/sqlbuild-orchestration/src/warehouse_history/_helpers/checks.rs", "crates/sqlbuild-orchestration/src/scheduler/main/list_ticks.rs", "crates/sqlbuild-orchestration/src/sensors/_helpers/attempts.rs", "crates/sqlbuild-orchestration/src/sensors/_helpers/facts.rs", "crates/sqlbuild-orchestration/src/snapshots/main/read_current_record.rs", "crates/sqlbuild-orchestration/src/snapshots/main/read_snapshot_documents.rs", "crates/sqlbuild-orchestration/src/snapshots/main/read_snapshot_nodes.rs", "crates/sqlbuild-orchestration/src/warehouse_metadata/main/read_person_catalog.rs", "crates/sqlbuild-orchestration/src/warehouse_history/main/read_person_check.rs", "crates/sqlbuild-orchestration/src/warehouse_history/main/read_person_checks.rs", "crates/sqlbuild-orchestration/src/warehouse_history/main/read_person_maintenance.rs", "crates/sqlbuild-orchestration/src/warehouse_metadata/main/read_person_statistics.rs"] +paths = ["crates/sqlbuild-orchestration/src/authorization/main/read_security_audit.rs", "crates/sqlbuild-orchestration/src/credentials/_helpers/active_session.rs", "crates/sqlbuild-orchestration/src/declarations/main/list_triggers.rs", "crates/sqlbuild-orchestration/src/platform/main/read_location_work.rs", "crates/sqlbuild-orchestration/src/runs/main/list_person_runs.rs", "crates/sqlbuild-orchestration/src/runs/main/list_runs.rs", "crates/sqlbuild-orchestration/src/runs/_helpers/lifecycle/reads.rs", "crates/sqlbuild-orchestration/src/warehouse_history/_helpers/checks.rs", "crates/sqlbuild-orchestration/src/scheduler/main/list_ticks.rs", "crates/sqlbuild-orchestration/src/sensors/_helpers/attempts.rs", "crates/sqlbuild-orchestration/src/sensors/_helpers/facts.rs", "crates/sqlbuild-orchestration/src/snapshots/main/read_current_record.rs", "crates/sqlbuild-orchestration/src/snapshots/main/read_snapshot_documents.rs", "crates/sqlbuild-orchestration/src/snapshots/main/read_snapshot_nodes.rs", "crates/sqlbuild-orchestration/src/warehouse_metadata/main/read_person_catalog.rs", "crates/sqlbuild-orchestration/src/warehouse_history/main/read_person_check.rs", "crates/sqlbuild-orchestration/src/warehouse_history/main/read_person_checks.rs", "crates/sqlbuild-orchestration/src/warehouse_history/main/read_person_maintenance.rs", "crates/sqlbuild-orchestration/src/warehouse_history/main/read_person_microbatch_coverage.rs", "crates/sqlbuild-orchestration/src/warehouse_metadata/main/read_person_statistics.rs"] reason = "Each is one control-store read with its own SQL and record type. Transactions, person authorization and decoding go through the shared begin_authorized, query_records, read_records and query_bodies helpers; what remains alike is the call shape." [[dupes.allowlist]] From 22d47994f6eb98e54b44859a9bb18614b96a140d Mon Sep 17 00:00:00 2001 From: Kevin Longe <273747162+kvlonge@users.noreply.github.com> Date: Fri, 6 Nov 2026 05:59:11 +0000 Subject: [PATCH 4/4] test: keep the worker contract's scheduler lease alive The worker contract held one 30 second scheduler lease without renewing it, so on a slow runner its second claim came after expiry and was fenced. Renew it in the background, as a running scheduler does. --- crates/sqlbuild-orchestration/tests/api/test_worker.rs | 4 +++- 1 file changed, 3 insertions(+), 1 deletion(-) diff --git a/crates/sqlbuild-orchestration/tests/api/test_worker.rs b/crates/sqlbuild-orchestration/tests/api/test_worker.rs index 0d21031a7c..ffc06aa6ce 100644 --- a/crates/sqlbuild-orchestration/tests/api/test_worker.rs +++ b/crates/sqlbuild-orchestration/tests/api/test_worker.rs @@ -8,7 +8,7 @@ use crate::runs::main::list_runs::list_runs; use crate::runs::models::{ NodeOutcome, RunAuthority, RunChange, RunChangeRequest, RunOutcome, RunStatus, }; -use crate::scheduler::tests::contracts::helpers::acquire_lease; +use crate::scheduler::tests::contracts::helpers::{acquire_lease, hold_lease}; use crate::snapshots::main::publish_snapshot::publish_snapshot; use crate::store::errors::ControlError; use crate::store::main::migrate_store::migrate_store; @@ -68,6 +68,7 @@ fn given_a_claimed_run_when_a_worker_connects_then_verification_precedes_start_a .await .expect("acquire") .expect("leader"); + let holder = hold_lease(store, &lease); let claim = claim_next_run(store, &lease, &BTreeMap::new(), None) .await .expect("claim") @@ -329,6 +330,7 @@ fn given_a_claimed_run_when_a_worker_connects_then_verification_precedes_start_a "{}", test_case.description ); + holder.abort(); drop(session); stop_api(api).await; close_test_store(fixture).await;