diff --git a/CHANGELOG.md b/CHANGELOG.md index 406b588d4..5bde0f873 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -8,6 +8,50 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 Releases before v0.27.0 predate the changelog. ## [Unreleased] +### Added + +- A schema-drift guard keeps the SQLite (`control/lite/schema.sql`) and + Postgres (`control/pg/schema.sql`) control schemas aligned: same tables, + column names, foreign keys and indexes, with every deliberate difference + listed and reasoned in `control/schema_drift.rs` (partition children, the + xid8 outbox columns and their ordering index, the pg-only + `job_messages.job_timeout_s`, the shared `timeline_id`, and the + `runner_sessions.runner_id` FK). The guard fails on an undocumented + difference and on a stale entry, so aligning the backends shrinks the list + instead of letting it rot. +- Coverage for id and time storage on both backends: a run id and job + round-trip in lowercase-canonical form, a sub-microsecond instant reads + back as whole microseconds (nanosecond `bigint` stamps keep full + precision), SQLite accepts several attempts of one job sharing a + `timeline_id` (the deliberate pg-`UNIQUE` divergence) and its + `job_requests_timeline_cascade` trigger prunes the shared timeline with the + last request. + +### Changed + +- SQLite's `jobs_ready` key is now `(pool_key, priority DESC, run_order, + job_order)` — exactly the claim / ready-queue `ORDER BY`, the key Postgres + already carries. `namespace_id` sat between the pool key and the priority, + so SQLite sorted the last three ORDER BY terms on every poll; the claim + query now reads the ready front in index order (`EXPLAIN QUERY PLAN` goes + from `USE TEMP B-TREE FOR LAST 3 TERMS OF ORDER BY` to no sort). SQLite + also gains `job_requests_attempts (run_id, job_id, request_id DESC)`, the + same index Postgres has, which serves the latest-attempt lookups and the + `jobs` -> `job_requests` cascade that the partial inflight index cannot. + Both are fresh-database changes: an existing database keeps working + unchanged, because a missing index is never an error. +- The control schema no longer carries objects nothing uses: + `run_submissions.secret_refs`, `jobs.not_before`, + `provision_requests.lease_owner`/`leased_until`, + `runner_sessions.engine_node_id`, `log_files.byte_count`/`line_count`, + `artifacts.upload_token_hash` and Postgres's `runners_labels` GIN index + (label matching reads `runners.labels` in Rust and uses no operator a GIN + index serves). No code referenced any of them, so an existing control + database that still has the columns keeps working. The namespace + platform columns (`cell_generation`, `config_version`, + `namespace_limits.max_job_timeout_minutes`/`priority_tier`/ + `run_history_retention_days`, `namespace_policies`) stay and are marked + platform-owned in the schema. ### Security diff --git a/crates/preloop-runner-server/src/control/lite/schema.sql b/crates/preloop-runner-server/src/control/lite/schema.sql index c47e5b7c5..29efb7306 100644 --- a/crates/preloop-runner-server/src/control/lite/schema.sql +++ b/crates/preloop-runner-server/src/control/lite/schema.sql @@ -21,6 +21,8 @@ CREATE TABLE namespaces ( namespace_id TEXT PRIMARY KEY, state TEXT NOT NULL DEFAULT 'active' CHECK (state IN ('active','suspended','draining','deleted')), + -- Platform-owned: cell fencing and config push bookkeeping; the engine + -- neither reads nor writes these two. cell_generation INTEGER NOT NULL DEFAULT 1, config_version INTEGER NOT NULL DEFAULT 0, created_at INTEGER NOT NULL DEFAULT (CAST(unixepoch('subsec') * 1000000 AS INTEGER)), @@ -33,6 +35,8 @@ CREATE TABLE namespace_limits ( max_running_jobs INTEGER, submit_rate_per_minute INTEGER, max_jobs_per_run INTEGER, + -- Platform-owned: written by the hosted platform; the engine does not + -- enforce these three yet. max_job_timeout_minutes INTEGER, priority_tier INTEGER NOT NULL DEFAULT 0, run_history_retention_days INTEGER @@ -45,6 +49,8 @@ CREATE TABLE namespace_pool_limits ( PRIMARY KEY (namespace_id, pool_key) ); +-- Platform-owned: the hosted platform writes per-tenant admission policy +-- here; the engine does not read it yet. CREATE TABLE namespace_policies ( namespace_id TEXT PRIMARY KEY REFERENCES namespaces(namespace_id) ON DELETE CASCADE, fork_pr_policy TEXT NOT NULL DEFAULT 'untrusted' CHECK (fork_pr_policy IN @@ -130,7 +136,6 @@ CREATE TABLE run_submissions ( github_context TEXT NOT NULL, workspace_snapshot TEXT, snapshot_timing TEXT, - secret_refs TEXT NOT NULL DEFAULT '{}', -- Record-level per-job maps the agreed schema has no table for -- (old backend's run_jobs): job_base_ids, job_names, job_needs, -- job_check_run_ids, caller_plans, jobs_list, reusable_calls, @@ -171,7 +176,6 @@ CREATE TABLE jobs ( priority INTEGER NOT NULL DEFAULT 0, run_order INTEGER NOT NULL DEFAULT 0, job_order INTEGER NOT NULL DEFAULT 0, - not_before INTEGER, enqueued_at INTEGER, claimed_by_runner_id INTEGER, claimed_at INTEGER, @@ -194,7 +198,11 @@ CREATE TABLE jobs ( FOREIGN KEY (run_id, parent_job_id) REFERENCES jobs(run_id, job_id) ON DELETE CASCADE DEFERRABLE INITIALLY DEFERRED ); -CREATE INDEX jobs_ready ON jobs(pool_key, namespace_id, priority DESC, run_order, job_order) +-- Claim order: `SELECT .. WHERE queue_state = 'ready' ORDER BY pool_key, +-- priority DESC, run_order, job_order`. The key columns are exactly the +-- ORDER BY (pg's index carries the same key), so the ready front is read in +-- order instead of sorting the whole ready set on every poll. +CREATE INDEX jobs_ready ON jobs(pool_key, priority DESC, run_order, job_order) WHERE queue_state = 'ready'; CREATE INDEX jobs_pending_expansion ON jobs(enqueued_at) WHERE queue_state = 'pending_expansion'; CREATE INDEX jobs_run_active ON jobs(run_id, queue_state) WHERE queue_state <> 'none'; @@ -281,6 +289,11 @@ CREATE TABLE job_requests ( ); CREATE UNIQUE INDEX job_requests_inflight ON job_requests(run_id, job_id) WHERE result IS NULL; CREATE INDEX job_requests_session ON job_requests(session_id) WHERE result IS NULL; +-- Latest-attempt lookups (`WHERE run_id = ? AND job_id = ? ORDER BY +-- request_id DESC LIMIT 1`) and the jobs -> job_requests cascade: the partial +-- inflight index cannot serve settled attempts. `request_id` is the rowid; +-- naming it keeps the key identical to pg's index. +CREATE INDEX job_requests_attempts ON job_requests(run_id, job_id, request_id DESC); CREATE TABLE job_leases ( request_id INTEGER PRIMARY KEY REFERENCES job_requests(request_id) ON DELETE CASCADE, @@ -345,8 +358,6 @@ CREATE TABLE log_files ( run_id TEXT NOT NULL REFERENCES runs(run_id) ON DELETE CASCADE, plan_id TEXT NOT NULL REFERENCES job_requests(agent_job_id) ON DELETE CASCADE, log_id INTEGER NOT NULL CHECK (log_id > 0), - byte_count INTEGER NOT NULL DEFAULT 0, - line_count INTEGER NOT NULL DEFAULT 0, updated_at INTEGER NOT NULL DEFAULT (CAST(unixepoch('subsec') * 1000000 AS INTEGER)), UNIQUE (plan_id, log_id) ); @@ -378,7 +389,6 @@ CREATE TABLE runner_sessions ( protocol TEXT NOT NULL CHECK (protocol IN ('broker','azdo')), client_id TEXT, verified INTEGER NOT NULL DEFAULT 0, - engine_node_id TEXT, created_at INTEGER NOT NULL DEFAULT (CAST(unixepoch('subsec') * 1000000 AS INTEGER)), last_seen_at INTEGER NOT NULL DEFAULT (CAST(unixepoch('subsec') * 1000000 AS INTEGER)) ); @@ -426,8 +436,6 @@ CREATE TABLE provision_requests ( pool_key TEXT NOT NULL, labels TEXT NOT NULL, requested_at INTEGER NOT NULL DEFAULT (CAST(unixepoch('subsec') * 1000000 AS INTEGER)), - leased_until INTEGER, - lease_owner TEXT, PRIMARY KEY (run_id, job_id), FOREIGN KEY (run_id, job_id) REFERENCES jobs(run_id, job_id) ON DELETE CASCADE ); @@ -565,6 +573,8 @@ CREATE TABLE check_run_updates ( CREATE INDEX check_run_updates_queue ON check_run_updates(installation_id, not_before); -- ── Artifacts ──────────────────────────────────────────────────────── +-- Artifact catalog rows. Bytes live in the file-backed ArtifactStore; the +-- only statement today is the run-archive `DELETE FROM artifacts`. CREATE TABLE artifacts ( artifact_id INTEGER PRIMARY KEY AUTOINCREMENT, namespace_id TEXT NOT NULL, @@ -575,7 +585,6 @@ CREATE TABLE artifacts ( size_bytes INTEGER, digest TEXT, storage_key TEXT NOT NULL, - upload_token_hash BLOB, created_at INTEGER NOT NULL DEFAULT (CAST(unixepoch('subsec') * 1000000 AS INTEGER)), finalized_at INTEGER, expires_at INTEGER, diff --git a/crates/preloop-runner-server/src/control/lite/testview.rs b/crates/preloop-runner-server/src/control/lite/testview.rs index 3124d2401..7210e8eea 100644 --- a/crates/preloop-runner-server/src/control/lite/testview.rs +++ b/crates/preloop-runner-server/src/control/lite/testview.rs @@ -1005,6 +1005,19 @@ impl TestDb<'_> { self.0.execute(sql, params) } + /// `EXPLAIN QUERY PLAN` detail lines for one statement, so index tests + /// can assert which access path (and whether a sort) the planner picks. + pub fn query_plan(&self, sql: &str) -> Vec { + let mut stmt = self + .0 + .prepare(&format!("EXPLAIN QUERY PLAN {sql}")) + .expect("prepare EXPLAIN QUERY PLAN"); + let rows = stmt + .query_map([], |row| row.get::<_, String>(3)) + .expect("run EXPLAIN QUERY PLAN"); + rows.map(|row| row.expect("plan row")).collect() + } + /// `job_assignments` staleness: `assigned_at`/`first_assigned_at`. pub fn set_assignment_times( &self, diff --git a/crates/preloop-runner-server/src/control/mod.rs b/crates/preloop-runner-server/src/control/mod.rs index 507e7b841..6484258a0 100644 --- a/crates/preloop-runner-server/src/control/mod.rs +++ b/crates/preloop-runner-server/src/control/mod.rs @@ -34,6 +34,8 @@ pub(crate) mod logic; pub mod logic; pub(crate) mod pg; #[cfg(test)] +mod schema_drift; +#[cfg(test)] mod tests; #[cfg(any(test, feature = "test-support"))] pub mod testview; diff --git a/crates/preloop-runner-server/src/control/pg/schema.sql b/crates/preloop-runner-server/src/control/pg/schema.sql index 5119a5574..30fd0ac58 100644 --- a/crates/preloop-runner-server/src/control/pg/schema.sql +++ b/crates/preloop-runner-server/src/control/pg/schema.sql @@ -48,6 +48,8 @@ CREATE TABLE namespaces ( -- suspended/deleted: neither (queued jobs stay queued, nothing starts). state text NOT NULL DEFAULT 'active' CHECK (state IN ('active','suspended','draining','deleted')), + -- Platform-owned: cell fencing and config push bookkeeping; the engine + -- neither reads nor writes these two. cell_generation bigint NOT NULL DEFAULT 1, -- bumped on cell move; fences stale writers config_version bigint NOT NULL DEFAULT 0, -- last platform push applied created_at timestamptz NOT NULL DEFAULT now(), @@ -60,6 +62,8 @@ CREATE TABLE namespaces ( -- predicate (a capped claim locks this row first, so nodes cannot -- overshoot). Not yet read: max_job_timeout_minutes, priority_tier, -- run_history_retention_days. +-- Those three are platform-owned: written by the hosted platform, not yet +-- enforced by the engine. CREATE TABLE namespace_limits ( namespace_id text PRIMARY KEY REFERENCES namespaces(namespace_id) ON DELETE CASCADE, max_queued_jobs integer, @@ -80,6 +84,8 @@ CREATE TABLE namespace_pool_limits ( ); -- Admission and job-build restrictions. +-- Platform-owned: the hosted platform writes per-tenant admission policy +-- here; the engine does not read it yet. CREATE TABLE namespace_policies ( namespace_id text PRIMARY KEY REFERENCES namespaces(namespace_id) ON DELETE CASCADE, fork_pr_policy text NOT NULL DEFAULT 'untrusted' CHECK (fork_pr_policy IN @@ -169,16 +175,14 @@ CREATE TABLE workflow_run_numbers ( PRIMARY KEY (namespace_id, repository, workflow_path) ); --- The request a run was created from. No secret values: `secret_refs` --- names which secrets (and scopes) the run may use; values are resolved --- from the SecretProvider when a job is acquired, never stored here. +-- The request a run was created from. No secret values are ever stored: +-- secrets are resolved from the SecretProvider when a job is acquired. CREATE TABLE run_submissions ( run_id uuid PRIMARY KEY REFERENCES runs(run_id) ON DELETE CASCADE, submission jsonb NOT NULL, -- WorkflowSubmission minus secrets github_context jsonb NOT NULL, workspace_snapshot jsonb, snapshot_timing jsonb, -- duration_ms, object_count, pack_bytes - secret_refs jsonb NOT NULL DEFAULT '{}', -- name -> {scope, version} -- Record-level per-job maps the table layout has no column for (jobs -- with no `jobs` row yet — a check run minted before its matrix leg -- materializes). Mirrors lite's `run_submissions.record_details`. @@ -225,7 +229,6 @@ CREATE TABLE jobs ( priority integer NOT NULL DEFAULT 0, run_order bigint NOT NULL DEFAULT 0, job_order integer NOT NULL DEFAULT 0, - not_before timestamptz, enqueued_at timestamptz, claimed_by_runner_id bigint, claimed_at timestamptz, @@ -414,14 +417,12 @@ CREATE TABLE timeline_records ( -- Per-plan log ids fit the official runner's 32-bit TaskLog.Id and remain -- stable across nodes. The unique pair arbitrates concurrent allocations. --- Content lives in file segments; this row tracks result-service counters. +-- Content and sizes live in file segments; this row only allocates ids. CREATE TABLE log_files ( log_key text PRIMARY KEY, run_id uuid NOT NULL REFERENCES runs(run_id) ON DELETE CASCADE, plan_id uuid NOT NULL REFERENCES job_requests(agent_job_id) ON DELETE CASCADE, log_id integer NOT NULL CHECK (log_id > 0), - byte_count bigint NOT NULL DEFAULT 0, - line_count bigint NOT NULL DEFAULT 0, updated_at timestamptz NOT NULL DEFAULT now(), UNIQUE (plan_id, log_id) ); @@ -443,7 +444,6 @@ CREATE TABLE runners ( registered_at timestamptz NOT NULL DEFAULT now(), last_seen_at timestamptz ); -CREATE INDEX runners_labels ON runners USING gin (labels); CREATE TABLE runner_sessions ( session_id uuid PRIMARY KEY, @@ -451,7 +451,6 @@ CREATE TABLE runner_sessions ( protocol text NOT NULL CHECK (protocol IN ('broker','azdo')), client_id text, verified boolean NOT NULL DEFAULT false, - engine_node_id text, -- node holding the long-poll (wake routing) created_at timestamptz NOT NULL DEFAULT now(), last_seen_at timestamptz NOT NULL DEFAULT now() ); @@ -505,8 +504,6 @@ CREATE TABLE provision_requests ( pool_key text NOT NULL, labels jsonb NOT NULL, requested_at timestamptz NOT NULL DEFAULT now(), - leased_until timestamptz, - lease_owner text, PRIMARY KEY (run_id, job_id), FOREIGN KEY (run_id, job_id) REFERENCES jobs(run_id, job_id) ON DELETE CASCADE ); @@ -669,8 +666,9 @@ CREATE TABLE check_run_updates ( CREATE INDEX check_run_updates_queue ON check_run_updates(installation_id, not_before); -- ── Artifacts (replaces the artifact part of the `meta` blob) ──────── --- Blobs live in object storage; these rows are the shared index. Upload --- tokens are stored as hashes, never raw. +-- Blobs live in object storage; upload state lives in the file-backed +-- ArtifactStore. These rows are the catalog; the only statement today is +-- the run-archive DELETE. -- -- The Actions cache is deliberately NOT here: it is a bounded cache -- colocated with the runner hosts (preloop-cache CAS: key index -> @@ -686,7 +684,6 @@ CREATE TABLE artifacts ( size_bytes bigint, digest text, storage_key text NOT NULL, - upload_token_hash bytea, created_at timestamptz NOT NULL DEFAULT now(), finalized_at timestamptz, expires_at timestamptz, diff --git a/crates/preloop-runner-server/src/control/schema_drift.rs b/crates/preloop-runner-server/src/control/schema_drift.rs new file mode 100644 index 000000000..d638cb68b --- /dev/null +++ b/crates/preloop-runner-server/src/control/schema_drift.rs @@ -0,0 +1,401 @@ +//! Schema-drift guard between the two copies of the control schema. +//! +//! `lite/schema.sql` is the SQLite translation of `pg/schema.sql` (the +//! source of truth, mirrored into `docs/control-schema.sql`). Both must +//! expose the same tables, the same column names, the same foreign keys and +//! the same indexes — only storage types differ (uuid/text -> TEXT, +//! timestamptz -> INTEGER microseconds, jsonb -> TEXT, bytea -> BLOB, +//! identity -> INTEGER PRIMARY KEY AUTOINCREMENT). +//! +//! [`DOCUMENTED_DIFFERENCES`] is the complete list of deliberate exceptions; +//! every entry carries the reason it exists. A difference that is not listed +//! fails [`control_schema_differences_are_documented`], and a listed +//! difference that no longer exists fails too — stale entries must be +//! deleted, so the guard shrinks itself whenever the backends are aligned. +//! +//! This guard is parse-level only. It cannot see storage semantics (a column +//! UNIQUE here is a constraint there), so it does not bless any difference it +//! cannot express — see the `UNIQUE`/foreign-key checks, which are the parts +//! of the drift that matter for correctness. +//! +//! This is not a migration tool: a database created at any other stamped +//! `schema_version` is still refused at open, and changing that is a separate +//! release blocker (`lite/mod.rs` / `pg/mod.rs`). Until versioned migrations +//! exist, this module's job is to keep the *fresh-database* schemas honest +//! and to name every place the two backends disagree. + +use std::collections::{BTreeMap, BTreeSet}; + +/// The SQLite translation (see the file header for the type mapping). +const LITE_SCHEMA: &str = include_str!("lite/schema.sql"); +/// The Postgres source of truth. +const PG_SCHEMA: &str = include_str!("pg/schema.sql"); + +/// Deliberate structural differences between the two schemas, keyed exactly +/// as [`differences`] reports them. Keep the keys verbatim: a typo makes the +/// guard fail twice (undocumented difference + stale entry). +const DOCUMENTED_DIFFERENCES: &[(&str, &str)] = &[ + // ── Partitioning: Postgres-only ────────────────────────────────── + ( + "table attempt_history_default: pg only", + "partition child: pg partitions the *_history tables by run_created_at and drops whole \ + partitions; SQLite has no partitioning and keeps one table. The logical table and its \ + columns are identical.", + ), + ( + "table job_history_default: pg only", + "partition child of job_history (see attempt_history_default).", + ), + ( + "table outbox_events_default: pg only", + "partition child of outbox_events (see attempt_history_default).", + ), + ( + "table run_history_default: pg only", + "partition child of run_history (see attempt_history_default).", + ), + ( + "table step_history_default: pg only", + "partition child of step_history (see attempt_history_default).", + ), + // ── Outbox ordering: single writer vs xid8 safe point ──────────── + ( + "table outbox_events column txid: pg only", + "pg orders the outbox by (txid, event_id) below pg_snapshot_xmin to read only finished \ + transactions. SQLite has exactly one writer (BEGIN IMMEDIATE), so commit order is \ + event_id order and no safe-point column exists. Owned by the durable-outbox PR: do not \ + drop or rename on either side.", + ), + ( + "table outbox_events column origin: pg only", + "pg stamps the writing node (`preloop.origin`) so a node skips its own rows; the single \ + SQLite writer has no other node to skip.", + ), + ( + "table consumer_offsets column last_txid: pg only", + "pg bookmarks the outbox by xid8 safe point; SQLite bookmarks by last_event_id. Owned by \ + the durable-outbox PR: do not drop or rename on either side.", + ), + ( + "index outbox_events_read: pg only", + "the (txid, event_id) reader index; there is no txid column on SQLite (see above).", + ), + // ── Columns and constraints that encode a deliberate behavioral split ─ + ( + "table job_messages column job_timeout_s: pg only", + "pg extracts `timeout-minutes` once at write because the reaper reads it for every \ + unfinished attempt on every tick and detoasting the template cost ~15x the rest of the \ + query. Lite re-extracts with json_extract from a non-toasted TEXT column. Aligning the \ + two needs a schema-version bump with a tested migration (existing databases have no \ + column and the reaper must not fail on them).", + ), + ( + "table job_requests column timeline_id inline UNIQUE: pg only", + "intentional: a timeline is shared across the attempts of one job. Pg makes the column \ + UNIQUE so `timelines` can FK to it; SQLite keeps several job_requests rows on one \ + timeline_id and deletes the timeline through the `job_requests_timeline_cascade` trigger \ + when the last request goes.", + ), + ( + "table timelines foreign key timeline_id -> job_requests(timeline_id): pg only", + "intentional: the reverse-direction FK for the pg UNIQUE above cannot be expressed on a \ + shared, non-unique SQLite column; `job_requests_timeline_cascade` owns the lifecycle.", + ), + ( + "table runner_sessions foreign key runner_id -> runners(runner_id): pg only", + "intentional: the legacy AgentSession/distributedtask create may name an agent id whose \ + registration arrives later (`require_live_runner: false` in runner_lifecycle.rs). SQLite \ + runs with `PRAGMA foreign_keys = ON`, so an FK would reject that row; lite instead \ + checks liveness inside the insert transaction when the caller requires it, reads through \ + a LEFT JOIN, and deletes the sessions before the runner row (lifecycle.rs). Note: the pg \ + FK means the legacy create cannot work against Postgres — tracked as a parity finding.", + ), +]; + +#[test] +fn control_schema_differences_are_documented() { + let differences = differences(); + let documented: BTreeMap<&str, &str> = DOCUMENTED_DIFFERENCES.iter().copied().collect(); + let mut problems = Vec::new(); + for difference in &differences { + if !documented.contains_key(difference.as_str()) { + problems.push(format!("undocumented difference: {difference}")); + } + } + for (difference, _reason) in DOCUMENTED_DIFFERENCES { + if !differences.contains(*difference) { + problems.push(format!( + "stale entry, delete it from DOCUMENTED_DIFFERENCES: {difference}" + )); + } + } + assert!( + problems.is_empty(), + "the SQLite and Postgres control schemas drifted ({} difference(s)):\n{}", + differences.len(), + problems.join("\n") + ); +} + +// ── The parsed schemas ────────────────────────────────────────────────── + +#[derive(Debug, Default, Clone)] +struct Table { + columns: BTreeSet, + /// Columns declared with an inline `UNIQUE`. + unique_columns: BTreeSet, + /// `column -> table(columns)` for inline `REFERENCES`, `a,b -> table(x,y)` + /// for table-level `FOREIGN KEY` clauses. The `ON DELETE` action is not + /// part of the key (it is behavioral, not structural, and both schemas + /// pick the action to match their own cascade rules). + foreign_keys: BTreeSet, +} + +#[derive(Debug, Default)] +struct Schema { + tables: BTreeMap, + /// Index name -> ordered column list (ASC/DESC and the access method are + /// not part of the comparison; the guard still catches column-order and + /// column-set drift because the list is ordered). + indexes: BTreeMap>, +} + +fn differences() -> BTreeSet { + let lite = parse(LITE_SCHEMA); + let pg = parse(PG_SCHEMA); + let mut out = BTreeSet::new(); + for name in lite.tables.keys() { + if !pg.tables.contains_key(name) { + out.insert(format!("table {name}: lite only")); + } + } + for name in pg.tables.keys() { + if !lite.tables.contains_key(name) { + out.insert(format!("table {name}: pg only")); + } + } + for (name, lite_table) in &lite.tables { + let Some(pg_table) = pg.tables.get(name) else { + continue; + }; + diff_members( + &lite_table.columns, + &pg_table.columns, + |item, side| format!("table {name} column {item}: {side} only"), + &mut out, + ); + diff_members( + &lite_table.unique_columns, + &pg_table.unique_columns, + |item, side| format!("table {name} column {item} inline UNIQUE: {side} only"), + &mut out, + ); + diff_members( + &lite_table.foreign_keys, + &pg_table.foreign_keys, + |item, side| format!("table {name} foreign key {item}: {side} only"), + &mut out, + ); + } + for name in lite.indexes.keys() { + match pg.indexes.get(name) { + None => { + out.insert(format!("index {name}: lite only")); + } + Some(pg_columns) => { + let lite_columns = &lite.indexes[name]; + if lite_columns != pg_columns { + out.insert(format!( + "index {name} columns: lite {lite_columns:?} vs pg {pg_columns:?}" + )); + } + } + } + } + for name in pg.indexes.keys() { + if !lite.indexes.contains_key(name) { + out.insert(format!("index {name}: pg only")); + } + } + out +} + +fn diff_members( + lite: &BTreeSet, + pg: &BTreeSet, + key: impl Fn(&str, &str) -> String, + out: &mut BTreeSet, +) { + for item in lite.difference(pg) { + out.insert(key(item, "lite")); + } + for item in pg.difference(lite) { + out.insert(key(item, "pg")); + } +} + +/// Extract the printable schema from one DDL file: tables (with columns, +/// inline UNIQUE flags and foreign keys) and indexes. Comments and string +/// literals never span a line in these files; `--` always starts a comment. +fn parse(sql: &str) -> Schema { + let mut schema = Schema::default(); + let mut current: Option<(String, Table)> = None; + let mut depth: i32 = 0; + for raw in sql.lines() { + let trimmed = strip_comment(raw).trim(); + if trimmed.is_empty() { + continue; + } + if current.is_some() { + if depth == 1 && !trimmed.starts_with(')') { + let table = &mut current.as_mut().expect("checked above").1; + parse_table_body_line(trimmed, table); + } + depth += paren_delta(trimmed); + if depth <= 0 { + let (name, table) = current.take().expect("checked above"); + schema.tables.insert(name, table); + } + continue; + } + if let Some(rest) = trimmed.strip_prefix("CREATE TABLE ") { + let (name, tail) = split_ident(rest); + let depth_after_header = paren_delta(&tail); + let table = Table::default(); + if depth_after_header > 0 { + depth = depth_after_header; + current = Some((name, table)); + } else { + // `CREATE TABLE x PARTITION OF y ...` (no body) or a + // single-line table. + schema.tables.insert(name, table); + } + continue; + } + let index = trimmed + .strip_prefix("CREATE UNIQUE INDEX ") + .or_else(|| trimmed.strip_prefix("CREATE INDEX ")); + if let Some(rest) = index + && let Some((name, columns)) = parse_index(rest) + { + schema.indexes.insert(name, columns); + } + } + schema +} + +/// One line inside a `CREATE TABLE (...)` body: a column definition, a +/// table-level constraint, or a continuation line (`ON DELETE ...`). +fn parse_table_body_line(line: &str, table: &mut Table) { + let (word, rest) = split_ident(line); + let keyword = word.to_ascii_uppercase(); + if matches!( + keyword.as_str(), + "PRIMARY" + | "FOREIGN" + | "UNIQUE" + | "CHECK" + | "CONSTRAINT" + | "ON" + | "DEFERRABLE" + | "INITIALLY" + ) { + if keyword == "FOREIGN" + && let Some(foreign_key) = table_foreign_key(&rest) + { + table.foreign_keys.insert(foreign_key); + } + return; + } + if !table.columns.insert(word.clone()) { + return; + } + if find_word(&rest, "UNIQUE").is_some() { + table.unique_columns.insert(word.clone()); + } + if let Some(target) = reference_target(&rest) { + table.foreign_keys.insert(format!("{word} -> {target}")); + } +} + +fn table_foreign_key(rest: &str) -> Option { + let open = rest.find('(')?; + let close = rest[open..].find(')')? + open; + let columns = normalize_columns(&rest[open + 1..close]); + let target = reference_target(&rest[close + 1..])?; + Some(format!("{columns} -> {target}")) +} + +fn reference_target(text: &str) -> Option { + let at = find_word(text, "REFERENCES")?; + let after = text[at + "REFERENCES".len()..].trim_start(); + let open = after.find('(')?; + let close = after[open..].find(')')? + open; + let table = after[..open].trim(); + let columns = normalize_columns(&after[open + 1..close]); + Some(format!("{table}({columns})")) +} + +fn parse_index(rest: &str) -> Option<(String, Vec)> { + let (name, tail) = split_ident(rest); + let open = tail.find('(')?; + let close = tail[open..].find(')')? + open; + let columns = tail[open + 1..close] + .split(',') + .map(|column| { + let column = column.trim(); + let column = column + .strip_suffix("DESC") + .or_else(|| column.strip_suffix("ASC")) + .unwrap_or(column); + column.trim().to_ascii_lowercase() + }) + .filter(|column| !column.is_empty()) + .collect(); + Some((name, columns)) +} + +// ── Text helpers ──────────────────────────────────────────────────────── + +fn strip_comment(line: &str) -> &str { + match line.find("--") { + Some(at) => &line[..at], + None => line, + } +} + +fn paren_delta(line: &str) -> i32 { + line.matches('(').count() as i32 - line.matches(')').count() as i32 +} + +/// `("CREATE TABLE", "namespaces (")`-style split on the first run of +/// whitespace. +fn split_ident(text: &str) -> (String, String) { + let text = text.trim(); + match text.find(char::is_whitespace) { + Some(at) => (text[..at].to_owned(), text[at..].trim().to_owned()), + None => (text.to_owned(), String::new()), + } +} + +fn normalize_columns(columns: &str) -> String { + columns + .split(',') + .map(str::trim) + .collect::>() + .join(",") +} + +/// Byte offset of `needle` in `haystack` when bounded by non-identifier +/// characters on both sides. +fn find_word(haystack: &str, needle: &str) -> Option { + for (at, _) in haystack.match_indices(needle) { + let before = haystack[..at].chars().next_back(); + let after = haystack[at + needle.len()..].chars().next(); + let is_boundary = |ch: Option| !ch.is_some_and(|c| c.is_alphanumeric() || c == '_'); + if is_boundary(before) && is_boundary(after) { + return Some(at); + } + } + None +} diff --git a/crates/preloop-runner-server/src/control/tests.rs b/crates/preloop-runner-server/src/control/tests.rs index 0f08806dc..6cfa6cb39 100644 --- a/crates/preloop-runner-server/src/control/tests.rs +++ b/crates/preloop-runner-server/src/control/tests.rs @@ -5277,6 +5277,69 @@ pub(crate) mod suite { outcome_b.run_number ); } + + /// Ids and instants have one canonical stored form on both backends: run + /// uuids lowercase (RFC 4122 text form), instants whole microseconds + /// since the Unix epoch. A sub-microsecond input is truncated toward the + /// epoch (`as_micros` / `timestamp_micros`), never rounded; nanosecond + /// `bigint` stamps keep full precision. + /// + /// The provenance of the run's `created_at` still differs and is recorded + /// here instead of hidden: SQLite stores the submitted value (truncated + /// to µs), Postgres takes the database clock (`DEFAULT now()`). Only the + /// precision rule and the nanosecond column are asserted identical. + pub(crate) async fn ids_and_timestamps_are_stored_canonically(backend: &dyn ControlBackend) { + let nanos = 1_700_000_000_123_456_789_i64; + let run_id = RunId(uuid::Uuid::parse_str("0F8FAD5B-D9CB-469F-A165-70867728950E").unwrap()); + let mut record = super::run_record(run_id); + // Upper-case spelling, deliberately: the stored form is the canonical + // lowercase one on both backends. + record.created_at = chrono::DateTime::from_timestamp_nanos(nanos); + record.fork_approval_requested_at_unix_nanos = Some(nanos); + record + .job_names + .insert(JobId("build".to_owned()), "build".to_owned()); + backend + .submit_run(SubmitRun { + namespace: "default".to_owned(), + record, + jobs: vec![submit_job(run_id, "build", 1)], + workflow_concurrency: None, + empty_concurrency_group: false, + check_hostable: false, + }) + .await + .unwrap(); + + let loaded = backend.run_record(run_id).await.unwrap(); + assert_eq!( + loaded.run_id.0.to_string(), + "0f8fad5b-d9cb-469f-a165-70867728950e", + "a run id must round-trip in the lowercase canonical form" + ); + assert_eq!( + loaded.job_names.get(&JobId("build".to_owned())), + Some(&"build".to_owned()), + "the job id round-trips verbatim: {:?}", + loaded.job_names + ); + let subsec = loaded.created_at.timestamp_subsec_nanos(); + assert_eq!( + subsec % 1000, + 0, + "microsecond columns must not keep sub-microsecond digits ({subsec} ns)" + ); + assert_eq!( + loaded.created_at, + chrono::DateTime::from_timestamp_micros(loaded.created_at.timestamp_micros()).unwrap(), + "the stored instant must be its own microsecond truncation" + ); + assert_eq!( + loaded.fork_approval_requested_at_unix_nanos, + Some(nanos), + "nanosecond-typed stamps must keep full precision" + ); + } } // ── SQLite ────────────────────────────────────────────────────────────── @@ -6454,6 +6517,14 @@ mod pg { }; suite::duplicate_run_number_is_reallocated(&backend).await; } + + #[tokio::test] + async fn ids_and_timestamps_are_stored_canonically() { + let Some((_pg, backend)) = backend().await else { + return skip_no_postgres(); + }; + suite::ids_and_timestamps_are_stored_canonically(&backend).await; + } } // ── New SQLite backend (`control::lite`) ──────────────────────────────── // @@ -7525,6 +7596,152 @@ mod lite { async fn duplicate_run_number_is_reallocated() { suite::duplicate_run_number_is_reallocated(&LiteBackend::in_memory().unwrap()).await; } + + #[tokio::test] + async fn ids_and_timestamps_are_stored_canonically() { + suite::ids_and_timestamps_are_stored_canonically(&LiteBackend::in_memory().unwrap()).await; + } + + /// `job_requests.timeline_id` is deliberately not UNIQUE: several attempts + /// of one job share a timeline (pg cannot — its column is UNIQUE and + /// `timelines` FKs to one request). The + /// `job_requests_timeline_cascade` trigger removes the timeline row only + /// when the last request referencing it is deleted. + #[tokio::test] + async fn timeline_is_shared_across_attempts_and_pruned_by_trigger() { + fn timeline_rows(tx: &crate::control::lite::TestDb<'_>, timeline_id: &str) -> i64 { + tx.0.query_row( + "SELECT count(*) FROM timelines WHERE timeline_id = ?1", + rusqlite::params![timeline_id], + |row| row.get::<_, i64>(0), + ) + .unwrap() + } + + let backend = LiteBackend::in_memory().unwrap(); + let run_id = RunId::new(); + backend + .submit_run(submit_run(run_id, vec![submit_job(run_id, "build", 1)])) + .await + .unwrap(); + let run_key = run_id.0.to_string(); + let timeline_id = uuid::Uuid::new_v4().to_string(); + let retry_agent = uuid::Uuid::new_v4().to_string(); + backend + .test_db_mutate(|tx| { + // Submission already minted one unclaimed attempt; settle it + // onto a shared timeline, then add the retry on the same + // timeline. Sharing is only meaningful once the previous + // attempt is settled — that is also what keeps the partial + // inflight index ("one open attempt per job") satisfied. + let first_agent: String = + tx.0.query_row( + "SELECT agent_job_id FROM job_requests \ + WHERE run_id = ?1 AND job_id = 'build'", + rusqlite::params![run_key], + |row| row.get(0), + ) + .unwrap(); + tx.execute( + "INSERT INTO timelines (timeline_id, change_id) VALUES (?1, 0)", + rusqlite::params![timeline_id], + ) + .unwrap(); + tx.execute( + "UPDATE job_requests SET timeline_id = ?2, result = 'failure' \ + WHERE agent_job_id = ?1", + rusqlite::params![first_agent, timeline_id], + ) + .unwrap(); + tx.execute( + "INSERT INTO job_requests (run_id, job_id, namespace_id, \ + agent_job_id, timeline_id, claimed_at) \ + VALUES (?1, 'build', 'default', ?2, ?3, 0)", + rusqlite::params![run_key, retry_agent, timeline_id], + ) + .expect("two attempts of one job may share the timeline id"); + assert_eq!( + timeline_rows(tx, &timeline_id), + 1, + "one shared timeline row" + ); + tx.execute( + "DELETE FROM job_requests WHERE agent_job_id = ?1", + rusqlite::params![retry_agent], + ) + .unwrap(); + assert_eq!( + timeline_rows(tx, &timeline_id), + 1, + "the timeline outlives the retry (the settled attempt still references it)" + ); + tx.execute( + "DELETE FROM job_requests WHERE agent_job_id = ?1", + rusqlite::params![first_agent], + ) + .unwrap(); + assert_eq!( + timeline_rows(tx, &timeline_id), + 0, + "the cascade trigger prunes the orphaned timeline" + ); + }) + .unwrap(); + } + + /// The claim query's `ORDER BY` must be served by `jobs_ready` without a + /// temp B-tree: the key order is the whole point of the index (pg carries + /// the same key). While `namespace_id` sat second, SQLite sorted the last + /// three ORDER BY terms on every poll. + #[tokio::test] + async fn jobs_ready_serves_the_claim_order_without_a_sort() { + let backend = LiteBackend::in_memory().unwrap(); + let plan = backend + .test_db_mutate(|tx| { + tx.query_plan(&format!( + "SELECT j.run_id FROM jobs j \ + WHERE j.queue_state = 'ready' AND ({}) \ + ORDER BY j.pool_key, j.priority DESC, j.run_order, j.job_order \ + LIMIT 64 OFFSET 0", + crate::control::types::NAMESPACE_ADMITS_CLAIM + )) + }) + .unwrap(); + assert!( + plan.iter().any(|detail| detail.contains("jobs_ready")), + "the claim query must use the jobs_ready index: {plan:?}" + ); + assert!( + !plan.iter().any(|detail| detail.contains("B-TREE")), + "the ready-queue order must come from the index, not a sort: {plan:?}" + ); + } + + /// Latest-attempt lookups must use `job_requests_attempts`: the partial + /// inflight index cannot serve settled attempts, and the row order must + /// come from the index. + #[tokio::test] + async fn attempt_lookups_use_the_attempts_index() { + let backend = LiteBackend::in_memory().unwrap(); + let plan = backend + .test_db_mutate(|tx| { + tx.query_plan( + "SELECT request_id FROM job_requests \ + WHERE run_id = '00000000-0000-0000-0000-000000000000' AND job_id = 'build' \ + ORDER BY request_id DESC LIMIT 1", + ) + }) + .unwrap(); + assert!( + plan.iter() + .any(|detail| detail.contains("job_requests_attempts")), + "latest-attempt lookups must use job_requests_attempts: {plan:?}" + ); + assert!( + !plan.iter().any(|detail| detail.contains("B-TREE")), + "the attempt order must come from the index, not a sort: {plan:?}" + ); + } } // ── Schema copies and legacy identity ─────────────────────────────────── diff --git a/docs/control-schema.sql b/docs/control-schema.sql index ef400c3c5..17b14c0b6 100644 --- a/docs/control-schema.sql +++ b/docs/control-schema.sql @@ -47,6 +47,8 @@ CREATE TABLE namespaces ( -- suspended/deleted: neither (queued jobs stay queued, nothing starts). state text NOT NULL DEFAULT 'active' CHECK (state IN ('active','suspended','draining','deleted')), + -- Platform-owned: cell fencing and config push bookkeeping; the engine + -- neither reads nor writes these two. cell_generation bigint NOT NULL DEFAULT 1, -- bumped on cell move; fences stale writers config_version bigint NOT NULL DEFAULT 0, -- last platform push applied created_at timestamptz NOT NULL DEFAULT now(), @@ -59,6 +61,8 @@ CREATE TABLE namespaces ( -- predicate (a capped claim locks this row first, so nodes cannot -- overshoot). Not yet read: max_job_timeout_minutes, priority_tier, -- run_history_retention_days. +-- Those three are platform-owned: written by the hosted platform, not yet +-- enforced by the engine. CREATE TABLE namespace_limits ( namespace_id text PRIMARY KEY REFERENCES namespaces(namespace_id) ON DELETE CASCADE, max_queued_jobs integer, @@ -79,6 +83,8 @@ CREATE TABLE namespace_pool_limits ( ); -- Admission and job-build restrictions. +-- Platform-owned: the hosted platform writes per-tenant admission policy +-- here; the engine does not read it yet. CREATE TABLE namespace_policies ( namespace_id text PRIMARY KEY REFERENCES namespaces(namespace_id) ON DELETE CASCADE, fork_pr_policy text NOT NULL DEFAULT 'untrusted' CHECK (fork_pr_policy IN @@ -168,16 +174,14 @@ CREATE TABLE workflow_run_numbers ( PRIMARY KEY (namespace_id, repository, workflow_path) ); --- The request a run was created from. No secret values: `secret_refs` --- names which secrets (and scopes) the run may use; values are resolved --- from the SecretProvider when a job is acquired, never stored here. +-- The request a run was created from. No secret values are ever stored: +-- secrets are resolved from the SecretProvider when a job is acquired. CREATE TABLE run_submissions ( run_id uuid PRIMARY KEY REFERENCES runs(run_id) ON DELETE CASCADE, submission jsonb NOT NULL, -- WorkflowSubmission minus secrets github_context jsonb NOT NULL, workspace_snapshot jsonb, snapshot_timing jsonb, -- duration_ms, object_count, pack_bytes - secret_refs jsonb NOT NULL DEFAULT '{}', -- name -> {scope, version} -- Record-level per-job maps the table layout has no column for (jobs -- with no `jobs` row yet — a check run minted before its matrix leg -- materializes). Mirrors lite's `run_submissions.record_details`. @@ -224,7 +228,6 @@ CREATE TABLE jobs ( priority integer NOT NULL DEFAULT 0, run_order bigint NOT NULL DEFAULT 0, job_order integer NOT NULL DEFAULT 0, - not_before timestamptz, enqueued_at timestamptz, claimed_by_runner_id bigint, claimed_at timestamptz, @@ -413,14 +416,12 @@ CREATE TABLE timeline_records ( -- Per-plan log ids fit the official runner's 32-bit TaskLog.Id and remain -- stable across nodes. The unique pair arbitrates concurrent allocations. --- Content lives in file segments; this row tracks result-service counters. +-- Content and sizes live in file segments; this row only allocates ids. CREATE TABLE log_files ( log_key text PRIMARY KEY, run_id uuid NOT NULL REFERENCES runs(run_id) ON DELETE CASCADE, plan_id uuid NOT NULL REFERENCES job_requests(agent_job_id) ON DELETE CASCADE, log_id integer NOT NULL CHECK (log_id > 0), - byte_count bigint NOT NULL DEFAULT 0, - line_count bigint NOT NULL DEFAULT 0, updated_at timestamptz NOT NULL DEFAULT now(), UNIQUE (plan_id, log_id) ); @@ -442,7 +443,6 @@ CREATE TABLE runners ( registered_at timestamptz NOT NULL DEFAULT now(), last_seen_at timestamptz ); -CREATE INDEX runners_labels ON runners USING gin (labels); CREATE TABLE runner_sessions ( session_id uuid PRIMARY KEY, @@ -450,7 +450,6 @@ CREATE TABLE runner_sessions ( protocol text NOT NULL CHECK (protocol IN ('broker','azdo')), client_id text, verified boolean NOT NULL DEFAULT false, - engine_node_id text, -- node holding the long-poll (wake routing) created_at timestamptz NOT NULL DEFAULT now(), last_seen_at timestamptz NOT NULL DEFAULT now() ); @@ -504,8 +503,6 @@ CREATE TABLE provision_requests ( pool_key text NOT NULL, labels jsonb NOT NULL, requested_at timestamptz NOT NULL DEFAULT now(), - leased_until timestamptz, - lease_owner text, PRIMARY KEY (run_id, job_id), FOREIGN KEY (run_id, job_id) REFERENCES jobs(run_id, job_id) ON DELETE CASCADE ); @@ -668,8 +665,9 @@ CREATE TABLE check_run_updates ( CREATE INDEX check_run_updates_queue ON check_run_updates(installation_id, not_before); -- ── Artifacts (replaces the artifact part of the `meta` blob) ──────── --- Blobs live in object storage; these rows are the shared index. Upload --- tokens are stored as hashes, never raw. +-- Blobs live in object storage; upload state lives in the file-backed +-- ArtifactStore. These rows are the catalog; the only statement today is +-- the run-archive DELETE. -- -- The Actions cache is deliberately NOT here: it is a bounded cache -- colocated with the runner hosts (preloop-cache CAS: key index -> @@ -685,7 +683,6 @@ CREATE TABLE artifacts ( size_bytes bigint, digest text, storage_key text NOT NULL, - upload_token_hash bytea, created_at timestamptz NOT NULL DEFAULT now(), finalized_at timestamptz, expires_at timestamptz,