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/3] 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/3] 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 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 3/3] 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;