Skip to main content

runner_manager_domain/
store.rs

1// owner: b2-sqlite-persistence
2
3//! SQLite persistence for configuration and recovery metadata.
4//!
5//! This is the only module in this crate that performs I/O, and it is a
6//! deliberate exception rather than a loosened rule: everything `b1` owns stays
7//! decidable with no network, no filesystem and no clock, and this module is the
8//! seam where those decisions are made durable. [`Store`] is a trait for exactly
9//! that reason — `b1`'s logic and the `testkit` fixtures remain usable with no
10//! database at all, and only the code that genuinely needs durability names
11//! [`SqliteStore`].
12//!
13//! # What is stored, and what is not
14//!
15//! **No credential of any kind is stored here.** The user access token lives in
16//! the machine-scoped secret store (`d2`), and the encoded JIT configuration
17//! lives in a restrictive temporary file that is deleted immediately after
18//! handoff (`05-infrastructure.md`). There is no column for either, and
19//! [`SqliteStore::dump_text`] exists partly so the security gate can prove that
20//! against a populated database rather than against this paragraph.
21//!
22//! # The three guarantees this module is responsible for
23//!
24//! 1. **Every load re-validates.** A row is never trusted. Policies are rebuilt
25//!    through [`ScalePolicy::from_persisted`], which re-runs D19's shape rules
26//!    and `min <= max`; attempts through [`RunnerAttempt::from_persisted`], which
27//!    re-runs the state/outcome/timestamp pairing; hosts through [`Host::new`]
28//!    and [`RefreshInterval::from_secs`]. A hand-edited database therefore cannot
29//!    inject a configuration the domain would refuse to construct in memory.
30//! 2. **Schema migrations are forward-only and versioned, and an unknown newer
31//!    version fails closed.** See [`SCHEMA_VERSION`] and [`MIGRATIONS`].
32//! 3. **`ScalePolicy::revision` is an optimistic-concurrency token.**
33//!    [`Store::update_policy`] matches on the revision the caller read and
34//!    reports [`StoreError::StaleRevision`] when someone else got there first.
35//!    That is what stops the TUI and a concurrent CLI invocation from silently
36//!    overwriting each other, and it is tested with two real concurrent writers
37//!    rather than by reasoning about the transaction.
38//!
39//! # Where the database lives
40//!
41//! Nowhere this module decides. [`SqliteStore::open`] takes a path and uses it.
42//! Resolving the platform application-data directory is `d1`'s job
43//! (`05-infrastructure.md`), and so is creating it: a missing parent directory is
44//! reported as [`StoreError::Open`] here rather than silently created, because a
45//! store that creates directories can create them in the wrong place.
46//!
47//! # A note on the two mapping directions
48//!
49//! [`PersistedPolicy`] and [`PersistedAttempt`] exist so the Rust half of the
50//! column mapping is checked by name. Their own documentation points out that
51//! the check stops at the field name — `PersistedAttempt { created_at:
52//! row.get("last_state_change_at")?, .. }` still compiles. This module closes
53//! that gap from both ends: every write binds by name with `:column`, every read
54//! reads by name with `row.get("column")`, so the column name sits literally
55//! beside the field name at each of the two crossings, and
56//! `tests::every_column_lands_in_the_field_of_the_same_name` loads a row of
57//! deliberately distinguishable values and asserts each landed where its name
58//! says it did.
59
60use std::fmt;
61use std::num::NonZeroU16;
62use std::path::{Path, PathBuf};
63use std::sync::atomic::{AtomicU64, Ordering};
64use std::sync::{Mutex, MutexGuard, PoisonError};
65use std::time::Duration;
66
67use rusqlite::types::{Value, ValueRef};
68use rusqlite::{Connection, OptionalExtension, Row, ToSql, TransactionBehavior, named_params};
69use serde::Serialize;
70use serde::de::DeserializeOwned;
71use uuid::Uuid;
72
73use crate::attempt::{AttemptError, AttemptOutcome, AttemptState, PersistedAttempt, RunnerAttempt};
74use crate::model::{
75    Arch, AttemptId, CachePolicy, Clock, Host, HostId, HostLabel, Os, PolicyId, RefreshInterval,
76    ScaleTarget, StartMode, SystemClock, TargetScope, Timestamp, ValidationError,
77};
78use crate::path::LocalAbsolutePath;
79use crate::policy::{PersistedPolicy, PolicyError, PolicyState, RoutingLabels, ScalePolicy};
80use crate::workspace::{WorkspaceError, WorkspaceKind};
81
82// ---------------------------------------------------------------------------
83// Errors
84// ---------------------------------------------------------------------------
85
86/// Everything that can go wrong between a domain value and a database row.
87///
88/// The variants are split the way a *caller* has to branch, not the way the
89/// implementation happens to fail. In particular [`StoreError::StaleRevision`] is
90/// its own variant rather than a flavour of [`StoreError::Sqlite`], because a
91/// concurrent edit is an ordinary outcome the CLI and the TUI must report as
92/// "someone else changed this; re-read and try again", while an I/O failure is
93/// not. [`StoreError::is_conflict`] is the predicate for that branch.
94#[derive(Debug, thiserror::Error)]
95pub enum StoreError {
96    /// The database file could not be opened. The parent directory belongs to
97    /// `d1`; this module never creates it.
98    #[error("the database at {path} could not be opened: {source}")]
99    Open {
100        path: PathBuf,
101        #[source]
102        source: rusqlite::Error,
103    },
104
105    /// Any other SQLite failure: a real I/O error, a locked database, a
106    /// malformed file.
107    #[error(transparent)]
108    Sqlite(#[from] rusqlite::Error),
109
110    /// The database was written by a newer build of this product.
111    ///
112    /// This fails closed on purpose. A newer version may have added a column
113    /// this build does not write — which this build would then drop on its next
114    /// write — or changed the meaning of one it does. Guessing is how a
115    /// downgrade silently corrupts a configuration.
116    #[error(
117        "this database is at schema version {found}, but this build of \
118         runner-manager understands version {supported}; upgrade runner-manager \
119         rather than running it against a database from a newer version"
120    )]
121    SchemaTooNew { found: u32, supported: u32 },
122
123    /// A migration did not apply. The transaction around it rolled back, so the
124    /// database is still at the previous version.
125    #[error("schema migration {version} ({name}) failed and was rolled back: {source}")]
126    Migration {
127        version: u32,
128        name: &'static str,
129        #[source]
130        source: rusqlite::Error,
131    },
132
133    /// A write lost an optimistic-concurrency race. **Nothing was written.**
134    ///
135    /// The caller must re-read the policy and re-apply its change; it must not
136    /// retry the value it holds, because that value was derived from a revision
137    /// that no longer exists.
138    #[error(
139        "policy {id} was written against revision {expected}, but the stored \
140         revision is now {found}; another process changed it first and nothing \
141         was written"
142    )]
143    StaleRevision {
144        id: PolicyId,
145        expected: u64,
146        found: u64,
147    },
148
149    /// Active work changed after an operator observed it for a disable.
150    /// The policy update and this predicate execute under one SQLite write
151    /// transaction, so no attempt journal write can cross the check.
152    #[error(
153        "policy {id} was confirmed with {expected} active runner(s), but now has \
154         {found}; nothing was written"
155    )]
156    ActiveCountChanged {
157        id: PolicyId,
158        expected: u16,
159        found: u16,
160    },
161
162    /// A host runner-root write was built from an override that is no longer
163    /// the stored one. **Nothing was written.**
164    ///
165    /// This is the host counterpart of [`StoreError::StaleRevision`], and it
166    /// exists because `hosts` carries no revision column: `03-migration-rollout`
167    /// requires the host mutation to compare "the expected old override" and
168    /// update *only* that column, so that a capacity or service-mode change made
169    /// between the operator's read and this write is not silently rolled back by
170    /// a whole-record [`Store::put_host`].
171    ///
172    /// Both paths are rendered rather than optional, so a message reads the same
173    /// way whichever direction the change went: an unset override is "the
174    /// platform default".
175    #[error(
176        "host {id} was written against runner root {expected}, but the stored \
177         override is now {found}; another process changed it first and nothing \
178         was written"
179    )]
180    RunnerRootChanged {
181        id: HostId,
182        expected: String,
183        found: String,
184    },
185
186    /// Uncleaned attempts changed after an operator observed the count a path
187    /// mutation was refused or permitted on. **Nothing was written.**
188    ///
189    /// Distinct from [`StoreError::ActiveCountChanged`], and the difference is
190    /// the whole point of the variant. *Active* excludes a terminal attempt;
191    /// *uncleaned* includes one, because a `finished` attempt whose cleanup has
192    /// not run still owns the directory under the root being moved, and a
193    /// persistent one still holds its slot lease
194    /// (`04-security-recovery.md`: "A host root setting cannot change while any
195    /// ephemeral attempt is active or unresolved").
196    ///
197    /// `subject` names the host or policy the count was taken for, already
198    /// rendered, because the two callers count different sets and an operator
199    /// reading the message needs to know which.
200    #[error(
201        "{subject} was confirmed with {expected} uncleaned attempt(s), but now \
202         has {found}; nothing was written"
203    )]
204    UncleanedCountChanged {
205        subject: String,
206        expected: u16,
207        found: u16,
208    },
209
210    /// Two uncleaned persistent attempts cannot hold one slot.
211    ///
212    /// Raised when a journal write collides with the partial unique index
213    /// `one_uncleaned_persistent_attempt_per_slot`. The allocation lock in `c2`
214    /// coordinates slot *selection*; this is the durable guard that catches the
215    /// race the lock cannot see — a second process, or a restart that lost the
216    /// lock — and `04-security-recovery.md` names it as the control for "two
217    /// attempts use one slot concurrently".
218    ///
219    /// **Nothing was written**: the statement is a single INSERT, so SQLite
220    /// rolls it back whole and the existing lease is untouched.
221    #[error(
222        "policy {policy} already holds an uncleaned attempt in persistent slot \
223         s{slot}; one slot is leased to at most one uncleaned attempt and \
224         nothing was written"
225    )]
226    SlotAlreadyLeased { policy: PolicyId, slot: u16 },
227
228    /// The row a write was aimed at is not there.
229    #[error("no {what} with id {id} is in the database")]
230    NotFound { what: &'static str, id: String },
231
232    /// An insert collided with an existing primary key.
233    #[error("a {what} with id {id} is already in the database")]
234    AlreadyExists { what: &'static str, id: String },
235
236    /// A stored policy is not a legal policy. This is the hand-edited-database
237    /// case: D19's shape rules and `min <= max` are re-run on every load.
238    #[error("the stored policy {id} is not a legal policy: {source}")]
239    CorruptPolicy {
240        id: PolicyId,
241        #[source]
242        source: PolicyError,
243    },
244
245    /// A stored attempt is not a legal attempt: its state, outcome and
246    /// timestamps do not pair the way this crate's own transitions pair them.
247    #[error("the stored attempt {id} is not a legal attempt: {source}")]
248    CorruptAttempt {
249        id: AttemptId,
250        #[source]
251        source: AttemptError,
252    },
253
254    /// A stored host does not satisfy a domain constraint — a blank display
255    /// name, or a refresh interval under the documented floor.
256    #[error("the stored host {id} is not a legal host: {source}")]
257    CorruptHost {
258        id: HostId,
259        #[source]
260        source: ValidationError,
261    },
262
263    /// A stored host's configured runner root is not a shape this product will
264    /// place a runner under.
265    ///
266    /// Separate from [`StoreError::CorruptHost`] because the source error is a
267    /// different vocabulary: `hosts.runner_root_override` is re-parsed through
268    /// [`LocalAbsolutePath::new`], so a hand-edited `\\nas\builds`, a relative
269    /// path, a bare drive root, or a Windows path in a database opened on Linux
270    /// fails closed here with the reason attached (D10). A path is not a
271    /// credential — see `crates/domain/src/path.rs` — so the offending text may
272    /// travel in the message, which is what makes the refusal actionable.
273    #[error("the stored host {id} has an unusable configured runner root: {source}")]
274    CorruptHostWorkspace {
275        id: HostId,
276        #[source]
277        source: WorkspaceError,
278    },
279
280    /// One column holds something that is not the kind of value it is declared
281    /// to hold. The row is named so an operator can find and fix it.
282    ///
283    /// **`value` never repeats the whole payload, and how much it repeats
284    /// depends on which column this is.** The row id is what an operator needs
285    /// to find the row; the payload only helps them recognise it, and repeating
286    /// all of it turns this error into a disclosure the moment it reaches a log.
287    ///
288    /// `table` and `column` are carried for the operator's sake and are also
289    /// what decides the echo: a column whose shape the schema fixes gets a
290    /// clipped echo of at most [`ECHO_LIMIT`] characters, and one that may hold
291    /// text the agent captured from a failure gets position only, with none of
292    /// the payload. The rule and the measurement behind it are on
293    /// `FREE_FORM_COLUMNS`, beside the decoder that applies it. (Named rather
294    /// than linked: it is private, and a link from here would not resolve for a
295    /// reader of the public docs.)
296    #[error("{table}.{column} of row {id} holds {value}, which is not {expected}")]
297    CorruptColumn {
298        table: &'static str,
299        column: &'static str,
300        id: String,
301        /// At most [`ECHO_LIMIT`] characters of the offending payload for a
302        /// constrained column, and none of it for a free-form one.
303        value: String,
304        expected: &'static str,
305    },
306
307    /// An integer that does not fit in a SQLite integer.
308    ///
309    /// SQLite has no unsigned 64-bit type, so a `u64` above `i64::MAX` has no
310    /// representation. Refused rather than saturated: saturating stores one
311    /// number and reads a different one back, silently, and the two values the
312    /// domain carries as `u64` -- `installation_id` and `github_runner_id` --
313    /// both come from GitHub, so a caller can reach this without doing anything
314    /// unusual.
315    #[error(
316        "{what} is {value}, which does not fit in a SQLite integer; SQLite \
317         integers are signed 64-bit and this store will not silently truncate one"
318    )]
319    UnrepresentableInteger { what: &'static str, value: u64 },
320
321    /// A runtime path that is not valid UTF-8 and therefore cannot be stored as
322    /// text.
323    ///
324    /// Lossy conversion is deliberately **not** used: `e3` deletes the runtime
325    /// directory this path names, and a path mangled by U+FFFD substitution
326    /// either fails to delete or names a different directory.
327    #[error("attempt {attempt} has a runtime path that is not valid UTF-8: {path:?}")]
328    UnrepresentablePath { attempt: AttemptId, path: PathBuf },
329}
330
331impl StoreError {
332    /// Whether this is an optimistic-concurrency conflict rather than a failure.
333    ///
334    /// The Definition of Done requires that "a stale-`revision` write is rejected
335    /// and the caller can distinguish it from an I/O error". This is that
336    /// distinction, exposed so a caller need not match on the variant shape to
337    /// make it.
338    ///
339    /// The three fences the workspace mutations add are conflicts on exactly the
340    /// same footing: each means "someone else got there first, re-read and try
341    /// again", and none of them means the database is unwell.
342    /// [`StoreError::SlotAlreadyLeased`] is included because that is what an
343    /// allocator that lost a race to a slot must do — pick another one — rather
344    /// than surface an I/O failure to an operator.
345    #[must_use]
346    pub const fn is_conflict(&self) -> bool {
347        matches!(
348            self,
349            StoreError::StaleRevision { .. }
350                | StoreError::ActiveCountChanged { .. }
351                | StoreError::RunnerRootChanged { .. }
352                | StoreError::UncleanedCountChanged { .. }
353                | StoreError::SlotAlreadyLeased { .. }
354        )
355    }
356}
357
358// ---------------------------------------------------------------------------
359// Schema and migrations
360// ---------------------------------------------------------------------------
361
362/// One forward-only schema step.
363#[derive(Debug, Clone, Copy)]
364struct Migration {
365    version: u32,
366    name: &'static str,
367    sql: &'static str,
368}
369
370/// The ordered, forward-only migration chain.
371///
372/// The runner, [`apply_migrations`], is a general one: its step-skipping
373/// behaviour is exercised against a synthetic chain in
374/// `tests::a_database_one_version_behind_gets_only_the_missing_step`, and
375/// against the production chain itself in
376/// `tests::a_version_one_database_migrates_through_the_whole_chain` and
377/// `tests::a_version_two_database_migrates_every_row_to_ephemeral`, which stop
378/// at `&MIGRATIONS[..1]` / `&MIGRATIONS[..2]`, write rows at that older shape,
379/// and then reopen through [`SqliteStore::open`].
380///
381/// Adding a step means adding a numbered `.sql` file beside this module and one
382/// entry here. It never means editing an applied file; see the header of
383/// `store/migrations/0001_initial_schema.sql`.
384const MIGRATIONS: &[Migration] = &[
385    Migration {
386        version: 1,
387        name: "initial_schema",
388        sql: include_str!("store/migrations/0001_initial_schema.sql"),
389    },
390    Migration {
391        version: 2,
392        name: "policy_host_label",
393        sql: include_str!("store/migrations/0002_policy_host_label.sql"),
394    },
395    Migration {
396        version: 3,
397        name: "workspace_locations",
398        sql: include_str!("store/migrations/0003_workspace_locations.sql"),
399    },
400];
401
402/// The schema version this build writes and understands.
403///
404/// A database above this is refused with [`StoreError::SchemaTooNew`]; a database
405/// below it is migrated up on open. Both directions are decided from the
406/// `schema_migrations` table, which records every applied step and when.
407pub const SCHEMA_VERSION: u32 = 3;
408
409/// Created outside the numbered chain, because the chain needs somewhere to
410/// record itself before its first step runs.
411const BOOTSTRAP_SQL: &str = "\
412CREATE TABLE IF NOT EXISTS schema_migrations (
413    version    INTEGER NOT NULL PRIMARY KEY,
414    name       TEXT    NOT NULL,
415    applied_at TEXT    NOT NULL
416) STRICT;";
417
418/// Every table this module reads, in the order [`SqliteStore::dump_text`] prints
419/// them.
420const TABLES: &[&str] = &["schema_migrations", "hosts", "policies", "attempts"];
421
422fn current_version(conn: &Connection) -> Result<u32, StoreError> {
423    let max: Option<i64> =
424        conn.query_row("SELECT MAX(version) FROM schema_migrations", [], |row| {
425            row.get(0)
426        })?;
427    // A negative recorded version is corrupted bookkeeping, and it is reported
428    // as that. Both directions of the old `unwrap_or(u32::MAX)` were wrong in
429    // the same way: it did fail closed, which is right, but it failed closed
430    // saying "this database is at schema version 4294967295", which is a number
431    // no database has ever been at. The operator's next move is to look at
432    // `schema_migrations`, and this says so.
433    match max {
434        None => Ok(0),
435        Some(raw) => u32::try_from(raw).map_err(|_| StoreError::CorruptColumn {
436            table: "schema_migrations",
437            column: "version",
438            id: raw.to_string(),
439            value: clip(&raw.to_string()),
440            expected: "a schema version this build could have written",
441        }),
442    }
443}
444
445/// Apply every step in `migrations` this database has not seen, in order.
446///
447/// Each step runs inside its own immediate transaction *together with* the row
448/// that records it, so a step that fails leaves the database at the previous
449/// version rather than half-migrated behind a version number claiming otherwise.
450fn apply_migrations(
451    conn: &mut Connection,
452    migrations: &[Migration],
453    clock: &dyn Clock,
454) -> Result<u32, StoreError> {
455    conn.execute_batch(BOOTSTRAP_SQL)?;
456
457    let supported = migrations.last().map_or(0, |m| m.version);
458    let found = current_version(conn)?;
459    if found > supported {
460        return Err(StoreError::SchemaTooNew { found, supported });
461    }
462
463    for migration in migrations.iter().filter(|m| m.version > found) {
464        let tx = conn.transaction_with_behavior(TransactionBehavior::Immediate)?;
465        let record = |source| StoreError::Migration {
466            version: migration.version,
467            name: migration.name,
468            source,
469        };
470        tx.execute_batch(migration.sql).map_err(record)?;
471        tx.execute(
472            "INSERT INTO schema_migrations (version, name, applied_at) \
473             VALUES (:version, :name, :applied_at)",
474            named_params! {
475                ":version": i64::from(migration.version),
476                ":name": migration.name,
477                ":applied_at": timestamp_to_text(clock.now()),
478            },
479        )
480        .map_err(record)?;
481        tx.commit()?;
482    }
483
484    Ok(supported)
485}
486
487// ---------------------------------------------------------------------------
488// The store trait
489// ---------------------------------------------------------------------------
490
491/// Durable storage for the three things that must survive a restart.
492///
493/// A trait rather than a concrete type so that `b1`'s logic and the `testkit`
494/// fixtures stay usable with no database — a capacity calculation or a recovery
495/// decision needs neither a file nor this trait — and so that a caller can be
496/// written against storage without being written against SQLite.
497///
498/// `Send + Sync` because the agent holds one of these across tasks while the TUI
499/// reads through the same handle. [`SqliteStore`] earns it with an internal
500/// mutex; a test double should do the same rather than being `!Sync` and forcing
501/// every caller to change shape.
502pub trait Store: fmt::Debug + Send + Sync {
503    /// Insert or replace this host.
504    ///
505    /// # Errors
506    /// [`StoreError::Sqlite`] on an I/O failure.
507    fn put_host(&self, host: &Host) -> Result<(), StoreError>;
508
509    /// One host, re-validated.
510    ///
511    /// # Errors
512    /// [`StoreError::CorruptHost`] or [`StoreError::CorruptColumn`] for a row the
513    /// domain refuses.
514    fn host(&self, id: HostId) -> Result<Option<Host>, StoreError>;
515
516    /// Every host, re-validated.
517    ///
518    /// # Errors
519    /// As [`Store::host`].
520    fn hosts(&self) -> Result<Vec<Host>, StoreError>;
521
522    /// Move the configured runner root, and **only** that column, while both the
523    /// override the caller read and the uncleaned ephemeral count it observed
524    /// still hold.
525    ///
526    /// `expected` is the override the caller **read**, and `new_root` the one it
527    /// wants stored; `None` on either side means "the platform default". A
528    /// mutation from and to the same value is a no-op that still runs both
529    /// predicates, which is what makes `host reset-runtime-root` safe to run
530    /// twice.
531    ///
532    /// **Why this is not [`Store::put_host`] with a changed field.**
533    /// `02-target-architecture.md`: "The store exposes a targeted host-root
534    /// mutation rather than writing a stale whole `Host` value. In one SQLite
535    /// transaction it compares the previously read override, confirms the count
536    /// of uncleaned ephemeral attempts, and updates only `runner_root_override`.
537    /// This prevents a simultaneous capacity or service-mode change from being
538    /// overwritten." A whole-record upsert built from a `Host` read seconds ago
539    /// would silently roll back a `host set-capacity` that landed in between,
540    /// and no revision column exists on `hosts` to catch it.
541    ///
542    /// **What is counted.** Every attempt in the journal whose workspace is
543    /// ephemeral and whose state is not `cleaned` — see
544    /// [`Store::uncleaned_ephemeral_attempts`], which is the read a caller uses
545    /// to obtain the number it passes here, so the two cannot disagree about
546    /// the set. Implementations must evaluate both predicates in the same write
547    /// transaction as the update; a separate read followed by a write does not
548    /// satisfy this contract.
549    ///
550    /// # Errors
551    /// [`StoreError::RunnerRootChanged`] when the stored override moved,
552    /// [`StoreError::UncleanedCountChanged`] when the count moved,
553    /// [`StoreError::NotFound`] when the host row is gone. In every case
554    /// nothing is written.
555    fn set_runner_root_override(
556        &self,
557        id: HostId,
558        expected: Option<&LocalAbsolutePath>,
559        new_root: Option<&LocalAbsolutePath>,
560        expected_uncleaned: u16,
561    ) -> Result<(), StoreError>;
562
563    /// Add a policy that is not there yet.
564    ///
565    /// # Errors
566    /// [`StoreError::AlreadyExists`] when the id is taken. Use
567    /// [`Store::update_policy`] to change an existing policy: this call carries
568    /// no revision check because there is no previous revision to check against.
569    fn insert_policy(&self, policy: &ScalePolicy) -> Result<(), StoreError>;
570
571    /// Write a changed policy, but only if nobody else changed it first.
572    ///
573    /// `expected_revision` is the revision the caller **read**, not the one the
574    /// policy now carries: every successful domain mutation advances
575    /// [`ScalePolicy::revision`], so a caller that loaded revision 3 and called
576    /// `set_max_capacity` holds revision 4 and passes 3 here. The write matches
577    /// on 3 and stores 4.
578    ///
579    /// # Errors
580    /// [`StoreError::StaleRevision`] when the stored revision is not
581    /// `expected_revision` — nothing is written, and the caller must re-read
582    /// rather than retry — or [`StoreError::NotFound`] when the row is gone.
583    fn update_policy(&self, policy: &ScalePolicy, expected_revision: u64)
584    -> Result<(), StoreError>;
585
586    /// Atomically update a policy only while its revision and active-attempt
587    /// count are exactly the values the caller observed. Implementations must
588    /// evaluate both predicates in the same write transaction as the update;
589    /// composing [`Store::attempts_for_policy`] and [`Store::update_policy`]
590    /// does not satisfy this contract.
591    fn update_policy_confirming_active_count(
592        &self,
593        policy: &ScalePolicy,
594        expected_revision: u64,
595        expected_active: u16,
596    ) -> Result<(), StoreError>;
597
598    /// Atomically update a policy only while its revision and **uncleaned**
599    /// attempt count are exactly the values the caller observed.
600    ///
601    /// The fence `repo set-workspace` needs, and the reason it is not
602    /// [`Store::update_policy_confirming_active_count`]: an attempt that has
603    /// concluded but not yet been cleaned is *not* active, and it is exactly the
604    /// attempt a workspace-path change must be refused behind. It still owns the
605    /// directory under the old root, and if it is persistent it still holds its
606    /// slot lease. `04-security-recovery.md`: "A repository path setting cannot
607    /// change while any attempt for that policy is active **or unresolved**."
608    ///
609    /// `03-migration-rollout.md` states the transaction boundary this
610    /// implements: "The policy store operation compares its revision and
611    /// confirms the uncleaned policy-attempt count. Both checks happen inside
612    /// the same SQLite write transaction as the mutation. The existing
613    /// whole-record `put_host` and active-count-only policy guard are not
614    /// sufficient for these commands." Composing
615    /// [`Store::uncleaned_attempts_for_policy`] and [`Store::update_policy`]
616    /// therefore does not satisfy this contract, even though that read is where
617    /// the caller gets the number it passes here.
618    ///
619    /// # Errors
620    /// [`StoreError::StaleRevision`], [`StoreError::UncleanedCountChanged`], or
621    /// [`StoreError::NotFound`]. In every case nothing is written.
622    fn update_policy_confirming_uncleaned_count(
623        &self,
624        policy: &ScalePolicy,
625        expected_revision: u64,
626        expected_uncleaned: u16,
627    ) -> Result<(), StoreError>;
628
629    /// Delete a policy, subject to the same revision check as a write.
630    ///
631    /// Deleting is a mutation like any other and races the same way: an operator
632    /// removing a repository while the TUI enables it must not silently win.
633    ///
634    /// Attempts belonging to the policy are deliberately left in place; see the
635    /// note on `attempts.policy_id` in the schema.
636    ///
637    /// # Errors
638    /// As [`Store::update_policy`].
639    fn remove_policy(&self, id: PolicyId, expected_revision: u64) -> Result<(), StoreError>;
640
641    /// One policy, re-validated.
642    ///
643    /// # Errors
644    /// [`StoreError::CorruptPolicy`] for a row violating D19's shape or
645    /// `min <= max`; [`StoreError::CorruptColumn`] for an unreadable column.
646    fn policy(&self, id: PolicyId) -> Result<Option<ScalePolicy>, StoreError>;
647
648    /// Every policy, re-validated.
649    ///
650    /// # Errors
651    /// As [`Store::policy`]. One corrupt row fails the whole call rather than
652    /// being skipped: a silently short policy list is a host that quietly stops
653    /// serving a repository, which is the failure nobody notices.
654    fn policies(&self) -> Result<Vec<ScalePolicy>, StoreError>;
655
656    /// Journal an attempt, inserting it or updating it in place.
657    ///
658    /// The journal has one writer — the agent holds the single-instance lock
659    /// (`05-infrastructure.md`) — so there is no revision token here. `created_at`
660    /// is written once at insert and is never overwritten by a later call, which
661    /// is the storage half of the domain's "`created_at` never moves".
662    ///
663    /// # Errors
664    /// [`StoreError::UnrepresentablePath`] for a non-UTF-8 runtime path,
665    /// otherwise [`StoreError::Sqlite`].
666    fn record_attempt(&self, attempt: &RunnerAttempt) -> Result<(), StoreError>;
667
668    /// One attempt, re-validated.
669    ///
670    /// # Errors
671    /// [`StoreError::CorruptAttempt`] for a state/outcome/timestamp combination
672    /// this crate's transitions cannot produce.
673    fn attempt(&self, id: AttemptId) -> Result<Option<RunnerAttempt>, StoreError>;
674
675    /// Every attempt, oldest first. This is the input to `e3`'s startup
676    /// recovery.
677    ///
678    /// # Errors
679    /// As [`Store::attempt`].
680    fn attempts(&self) -> Result<Vec<RunnerAttempt>, StoreError>;
681
682    /// Every attempt of one policy, oldest first.
683    ///
684    /// # Errors
685    /// As [`Store::attempt`].
686    fn attempts_for_policy(&self, policy_id: PolicyId) -> Result<Vec<RunnerAttempt>, StoreError>;
687
688    /// The attempts of one policy that still occupy a host capacity slot.
689    ///
690    /// "Active" is [`AttemptState::counts_against_capacity`] — every
691    /// non-terminal state — and this is the narrower of the two questions a
692    /// workspace mutation asks. It is here beside its counterpart so that the
693    /// distinction is visible at the trait rather than reconstructed by each
694    /// caller from [`Store::attempts_for_policy`] and a filter each writes
695    /// slightly differently.
696    ///
697    /// # Errors
698    /// As [`Store::attempt`].
699    fn active_attempts_for_policy(
700        &self,
701        policy_id: PolicyId,
702    ) -> Result<Vec<RunnerAttempt>, StoreError>;
703
704    /// The attempts of one policy that have not been cleaned, oldest first.
705    ///
706    /// A superset of [`Store::active_attempts_for_policy`]: it also holds the
707    /// terminal attempts whose cleanup has not completed. Those are invisible to
708    /// capacity and decisive for a path change, which is the distinction
709    /// `04-security-recovery.md` draws between "active" and "unresolved".
710    ///
711    /// This is the read that produces `expected_uncleaned` for
712    /// [`Store::update_policy_confirming_uncleaned_count`], and `c2`'s allocator
713    /// input: "Load uncleaned attempts for the policy"
714    /// (`02-target-architecture.md`, "Slot allocation").
715    ///
716    /// # Errors
717    /// As [`Store::attempt`].
718    fn uncleaned_attempts_for_policy(
719        &self,
720        policy_id: PolicyId,
721    ) -> Result<Vec<RunnerAttempt>, StoreError>;
722
723    /// The durable slot leases one policy holds, oldest first.
724    ///
725    /// Every uncleaned **persistent** attempt, which is the same set the partial
726    /// unique index `one_uncleaned_persistent_attempt_per_slot` enforces
727    /// uniqueness over. It deliberately includes a terminal attempt whose
728    /// cleanup failed: "Every persistent attempt whose state is not `cleaned` is
729    /// a durable slot lease, including a terminal attempt whose cleanup failed"
730    /// (`02-target-architecture.md`). Every returned attempt therefore answers
731    /// `true` to [`RunnerAttempt::holds_slot_lease`] and carries a slot.
732    ///
733    /// The filesystem is never consulted to answer this. Invariant 6: "Uncleaned
734    /// attempt rows, the database lease constraint, and the allocation lock
735    /// remain authoritative; the filesystem is never scanned to infer ownership
736    /// or capacity."
737    ///
738    /// # Errors
739    /// As [`Store::attempt`].
740    fn slot_leases_for_policy(&self, policy_id: PolicyId)
741    -> Result<Vec<RunnerAttempt>, StoreError>;
742
743    /// Every uncleaned **ephemeral** attempt on this host, oldest first.
744    ///
745    /// The read that produces `expected_uncleaned` for
746    /// [`Store::set_runner_root_override`], and the reason it is host-wide
747    /// rather than per-policy: these attempts are the ones whose directories sit
748    /// under the host runner root, and an attempt that outlived its policy row
749    /// still owns one (see the note on `attempts.policy_id` in the schema). A
750    /// count scoped to the policies of one host would drop exactly those, and
751    /// `04-security-recovery.md` requires unknown-policy attempts to keep their
752    /// fail-closed ownership behaviour rather than to become invisible.
753    ///
754    /// # Errors
755    /// As [`Store::attempt`].
756    fn uncleaned_ephemeral_attempts(&self) -> Result<Vec<RunnerAttempt>, StoreError>;
757
758    /// Forget one attempt. Returns whether a row was removed.
759    ///
760    /// # Errors
761    /// [`StoreError::Sqlite`] on an I/O failure.
762    fn remove_attempt(&self, id: AttemptId) -> Result<bool, StoreError>;
763}
764
765// ---------------------------------------------------------------------------
766// Attempt sets
767// ---------------------------------------------------------------------------
768//
769// Four questions are asked of the `attempts` table by the mutations and queries
770// above, and three of them are new. Their `WHERE` fragments are built here, from
771// the domain's own predicates and the domain's own serde tokens, rather than
772// written out at each call site.
773//
774// **Why derived rather than spelled.** The active-set fragment used to be the
775// literal `state IN ('allocated', 'jit_received', 'starting', 'idle', 'busy')`,
776// written twice in one function. That is a copy of
777// `AttemptState::counts_against_capacity` maintained by hand: a tenth attempt
778// state added to the domain would be counted by the formula in `capacity` and
779// silently *not* counted by this fence, and no test that does not already know
780// to look would say so. Deriving it means the domain decides once, and
781// `tests::the_attempt_set_predicates_follow_the_domain` asserts the two agree
782// state by state.
783
784/// `WHERE`-fragment for the attempts that still occupy a capacity slot.
785fn active_sql() -> String {
786    let states = AttemptState::ALL
787        .into_iter()
788        .filter(|state| state.counts_against_capacity())
789        .map(|state| format!("'{}'", token(&state)))
790        .collect::<Vec<_>>()
791        .join(", ");
792    format!("state IN ({states})")
793}
794
795/// `WHERE`-fragment for the attempts whose cleanup has not completed.
796///
797/// Deliberately `<> 'cleaned'` and not "is not terminal": a `finished` attempt
798/// that has not been cleaned still owns its directory and, when persistent, its
799/// slot lease.
800fn uncleaned_sql() -> String {
801    format!("state <> '{}'", token(&AttemptState::Cleaned))
802}
803
804/// `WHERE`-fragment for the uncleaned attempts of one workspace kind.
805fn uncleaned_of_kind_sql(kind: WorkspaceKind) -> String {
806    format!(
807        "workspace_mode = '{}' AND {}",
808        token(&kind),
809        uncleaned_sql()
810    )
811}
812
813/// The ceiling every guarded attempt count saturates at, on both sides.
814///
815/// Named once because it is stated twice in two notations — as an integer
816/// literal inside SQL by [`attempt_count_sql`], and as a Rust narrowing by
817/// [`clamped_count`] — and the fences are only sound while the two agree.
818const ATTEMPT_COUNT_CEILING: u16 = u16::MAX;
819
820/// The scalar sub-select counting the attempts matching `predicate`.
821///
822/// Clamped to [`ATTEMPT_COUNT_CEILING`] in SQL so the comparison against the
823/// caller's `u16` is representable on both sides rather than wrapping: a journal
824/// holding more than 65 535 uncleaned attempts for one subject reports the
825/// ceiling, which is the same figure the diagnosis reports and the same one a
826/// caller's own saturating count would produce. The clamp is therefore a
827/// saturation, not a fence in its own right — at the ceiling the two sides can
828/// still agree — and it is chosen deliberately, because the alternative is a
829/// refusal whose message reads "expected 65535, but now has 65535". Nothing in
830/// this product creates 65 535 concurrent attempts for one policy; host capacity
831/// is a `NonZeroU16` bound checked long before here.
832fn attempt_count_sql(predicate: &str) -> String {
833    format!("SELECT MIN(COUNT(*), {ATTEMPT_COUNT_CEILING}) FROM attempts WHERE {predicate}")
834}
835
836/// The narrowing partner of [`attempt_count_sql`]'s clamp.
837///
838/// The sub-select already saturates, so the `unwrap_or` is unreachable for any
839/// value SQLite can return through it; it is spelled out rather than
840/// `expect`-ed so that a future caller counting through some other statement
841/// still saturates instead of panicking.
842fn clamped_count(found: i64) -> u16 {
843    u16::try_from(found).unwrap_or(ATTEMPT_COUNT_CEILING)
844}
845
846/// Which set of attempts a guarded policy write counts.
847///
848/// A closed enum rather than a `&str` argument, so the SQL a caller can reach is
849/// one of two constants built here and never text a caller supplies.
850#[derive(Debug, Clone, Copy, PartialEq, Eq)]
851enum CountedAttempts {
852    /// Still occupying a capacity slot — the fence `set-scale --enabled false`
853    /// needs.
854    Active,
855    /// Not yet cleaned, terminal or otherwise — the fence `repo set-workspace`
856    /// needs.
857    Uncleaned,
858}
859
860impl CountedAttempts {
861    /// The scalar sub-select counting this set for the policy bound as `:id`.
862    ///
863    /// Scoped to one policy; [`attempt_count_sql`] owns the statement shape and
864    /// the saturation contract it shares with the host-wide fence.
865    fn count_sql(self) -> String {
866        let predicate = match self {
867            CountedAttempts::Active => active_sql(),
868            CountedAttempts::Uncleaned => uncleaned_sql(),
869        };
870        attempt_count_sql(&format!("policy_id = :id AND {predicate}"))
871    }
872
873    /// The conflict this set reports when the count moved under the write.
874    fn count_changed(self, id: PolicyId, expected: u16, found: u16) -> StoreError {
875        match self {
876            CountedAttempts::Active => StoreError::ActiveCountChanged {
877                id,
878                expected,
879                found,
880            },
881            CountedAttempts::Uncleaned => StoreError::UncleanedCountChanged {
882                subject: format!("policy {id}"),
883                expected,
884                found,
885            },
886        }
887    }
888}
889
890// ---------------------------------------------------------------------------
891// The rusqlite implementation
892// ---------------------------------------------------------------------------
893
894/// The rusqlite-backed [`Store`].
895///
896/// One connection behind a mutex. SQLite serialises writers anyway, so a
897/// connection pool would buy concurrency the database does not offer; what the
898/// mutex buys is `Sync`, so the agent can hold one handle across tasks.
899///
900/// Opened with `synchronous = FULL` and a request for WAL. `FULL` because this
901/// journal exists precisely to survive an unclean stop: a handful of fsyncs per
902/// runner attempt is not a cost worth trading for the chance of losing the last
903/// write before a power cut. WAL so that a reader — the TUI — does not block the
904/// agent's journal writes.
905///
906/// **The WAL half is a request, not a guarantee**, which is why
907/// [`Self::journal_mode`] exists to report what actually happened. SQLite falls
908/// back to `delete` where the directory cannot host WAL's shared-memory file,
909/// and says so in the pragma's return row rather than by failing.
910pub struct SqliteStore {
911    conn: Mutex<Connection>,
912    path: Option<PathBuf>,
913    schema_version: u32,
914    /// The journal mode this database actually ended up in, as SQLite reported
915    /// it. Not necessarily `wal`; see [`SqliteStore::journal_mode`].
916    journal_mode: String,
917    clock_skew_repairs: AtomicU64,
918}
919
920impl fmt::Debug for SqliteStore {
921    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
922        f.debug_struct("SqliteStore")
923            .field("path", &self.path)
924            .field("schema_version", &self.schema_version)
925            .field("journal_mode", &self.journal_mode)
926            .field(
927                "clock_skew_repairs",
928                &self.clock_skew_repairs.load(Ordering::Relaxed),
929            )
930            .finish()
931    }
932}
933
934impl SqliteStore {
935    /// Open (or create) the database at `path` and migrate it to
936    /// [`SCHEMA_VERSION`].
937    ///
938    /// The path is used exactly as given. Resolving the platform
939    /// application-data directory and creating it is `d1`'s job; a missing parent
940    /// directory is reported rather than created.
941    ///
942    /// # Errors
943    /// [`StoreError::Open`] when the file cannot be opened,
944    /// [`StoreError::SchemaTooNew`] when the database came from a newer build,
945    /// [`StoreError::Migration`] when a step fails.
946    pub fn open(path: impl AsRef<Path>) -> Result<Self, StoreError> {
947        let path = path.as_ref();
948        let conn = Connection::open(path).map_err(|source| StoreError::Open {
949            path: path.to_path_buf(),
950            source,
951        })?;
952        Self::with_migrations(conn, Some(path.to_path_buf()), MIGRATIONS)
953    }
954
955    /// An anonymous in-memory database, migrated to [`SCHEMA_VERSION`].
956    ///
957    /// For tests and for a dry run. It is private to this one connection — a
958    /// second `open_in_memory` is a different database — so it cannot stand in
959    /// for a file store in a test about two concurrent writers.
960    ///
961    /// # Errors
962    /// As [`SqliteStore::open`].
963    pub fn open_in_memory() -> Result<Self, StoreError> {
964        let conn = Connection::open_in_memory().map_err(|source| StoreError::Open {
965            path: PathBuf::from(":memory:"),
966            source,
967        })?;
968        Self::with_migrations(conn, None, MIGRATIONS)
969    }
970
971    fn with_migrations(
972        mut conn: Connection,
973        path: Option<PathBuf>,
974        migrations: &[Migration],
975    ) -> Result<Self, StoreError> {
976        // A row-returning pragma, so it cannot go through `execute`, and the row
977        // it returns is the mode the database **ended up in** rather than the
978        // one that was asked for. Discarding it was a real gap: where WAL is
979        // unavailable -- it needs shared memory, which a network-mounted
980        // application data directory or some container `/tmp` does not provide
981        // -- SQLite quietly leaves the database in `delete` and says so in this
982        // row. Two things then went wrong at once. This type's own
983        // documentation promises "WAL so that a reader -- the TUI -- does not
984        // block the agent's journal writes", and that promise silently stopped
985        // holding in production with nothing anywhere to say so; and the `-wal`
986        // assertion in `tests/store_journal.rs` failed on an otherwise healthy
987        // build without explaining why.
988        //
989        // Recorded and warned about rather than refused. The journal is still
990        // *correct* in `delete` mode, only less concurrent, and an operator
991        // whose application data directory sits on a network mount wants a
992        // working agent more than a principled refusal to start. The fact is
993        // exposed through `SqliteStore::journal_mode` so a test can ask instead
994        // of assuming and an operator can see it in a support bundle. An
995        // in-memory database answers `memory`, which is correct and exempt.
996        let journal_mode: String =
997            conn.query_row("PRAGMA journal_mode = WAL", [], |row| row.get(0))?;
998        if path.is_some() && !journal_mode.eq_ignore_ascii_case("wal") {
999            tracing::warn!(
1000                path = ?path,
1001                journal_mode = %journal_mode,
1002                "this database did not enter WAL mode, so a reader will block \
1003                 the agent's journal writes. The usual cause is a directory \
1004                 that cannot host WAL's shared-memory file, such as a network \
1005                 mount."
1006            );
1007        }
1008        conn.pragma_update(None, "synchronous", "FULL")?;
1009        // Two processes will contend (the TUI and a CLI invocation), and the
1010        // loser of a write lock should wait briefly rather than fail: a
1011        // `database is locked` error surfaced to an operator who did nothing
1012        // wrong is indistinguishable from a bug. It is also what makes the
1013        // stale-revision answer deterministic instead of racing SQLITE_BUSY.
1014        conn.busy_timeout(Duration::from_secs(5))?;
1015        // No foreign key exists today (see the schema), but this is a
1016        // per-connection setting that silently defaults to off, so setting it
1017        // here is what would make a future migration's key actually enforced.
1018        conn.pragma_update(None, "foreign_keys", true)?;
1019
1020        // `applied_at` is an audit stamp, not a decision input, which is why the
1021        // production clock is acceptable here and nowhere else in this crate.
1022        let schema_version = apply_migrations(&mut conn, migrations, &SystemClock)?;
1023
1024        Ok(Self {
1025            conn: Mutex::new(conn),
1026            path,
1027            schema_version,
1028            journal_mode,
1029            clock_skew_repairs: AtomicU64::new(0),
1030        })
1031    }
1032
1033    /// The schema version this database is at.
1034    #[must_use]
1035    pub const fn schema_version(&self) -> u32 {
1036        self.schema_version
1037    }
1038
1039    /// The journal mode this database is actually in, lowercased by SQLite.
1040    ///
1041    /// `wal` for a healthy file store, `memory` for an in-memory one, and
1042    /// something else — `delete`, usually — where the directory cannot host
1043    /// WAL's shared-memory file. That last case is not a failure but it does
1044    /// mean a reader blocks the agent's journal writes, so it is worth showing
1045    /// an operator rather than assuming.
1046    #[must_use]
1047    pub fn journal_mode(&self) -> &str {
1048        &self.journal_mode
1049    }
1050
1051    /// Whether a reader can read this database without blocking the agent's
1052    /// writes.
1053    ///
1054    /// True exactly when [`Self::journal_mode`] is `wal`. An in-memory store is
1055    /// **not** included: it is private to one connection, so the question does
1056    /// not arise for it.
1057    #[must_use]
1058    pub fn readers_do_not_block_writers(&self) -> bool {
1059        self.journal_mode.eq_ignore_ascii_case("wal")
1060    }
1061
1062    /// The path this store was opened from, or `None` for an in-memory one.
1063    #[must_use]
1064    pub fn path(&self) -> Option<&Path> {
1065        self.path.as_deref()
1066    }
1067
1068    /// How many attempt timestamps this store has repaired for backwards clock
1069    /// movement since it was opened.
1070    ///
1071    /// See [`SqliteStore::normalise`] for what is repaired and why. A non-zero
1072    /// value means this machine's clock stepped backwards while an attempt was in
1073    /// flight; each repair is also logged at `warn`.
1074    #[must_use]
1075    pub fn clock_skew_repairs(&self) -> u64 {
1076        self.clock_skew_repairs.load(Ordering::Relaxed)
1077    }
1078
1079    /// Every row of every table, as text, in a deterministic order.
1080    ///
1081    /// Two callers. An operator support bundle, and the security gate: the
1082    /// Definition of Done requires that "a grep of every fixture database and its
1083    /// dump finds no token-shaped value", and a dump produced here is a testable
1084    /// artifact where a `sqlite3 .dump` invocation in a shell script is not.
1085    ///
1086    /// This is safe to attach to a bug report **because no column carries a
1087    /// credential**, not because anything here redacts one. If a column ever
1088    /// does, this function becomes a disclosure and the schema is what has to
1089    /// change.
1090    ///
1091    /// # Errors
1092    /// [`StoreError::Sqlite`] on an I/O failure.
1093    pub fn dump_text(&self) -> Result<String, StoreError> {
1094        use std::fmt::Write as _;
1095
1096        let conn = self.lock();
1097        let mut out = String::new();
1098        let _ = writeln!(out, "-- schema version {}", self.schema_version);
1099        for table in TABLES {
1100            let _ = writeln!(out, "-- table {table}");
1101            let mut stmt = conn.prepare(&format!("SELECT * FROM \"{table}\""))?;
1102            let columns: Vec<String> = stmt.column_names().into_iter().map(String::from).collect();
1103            let mut rows = stmt.query([])?;
1104            while let Some(row) = rows.next()? {
1105                for (index, column) in columns.iter().enumerate() {
1106                    if index > 0 {
1107                        out.push_str(", ");
1108                    }
1109                    let _ = write!(out, "{table}.{column}={}", render(row.get_ref(index)?));
1110                }
1111                out.push('\n');
1112            }
1113        }
1114        Ok(out)
1115    }
1116
1117    fn lock(&self) -> MutexGuard<'_, Connection> {
1118        // A panicking writer cannot leave SQLite half-written: every write here
1119        // runs inside a statement or a transaction whose guard rolls back on
1120        // drop. So the poison flag says a Rust caller panicked, not that the
1121        // database is inconsistent, and refusing to serve the database over it
1122        // would turn one panic into a permanently unusable installation.
1123        self.conn.lock().unwrap_or_else(PoisonError::into_inner)
1124    }
1125
1126    /// Repair a persisted attempt whose timestamps run backwards, and say so.
1127    ///
1128    /// **The hazard.** [`RunnerAttempt`]'s in-memory transitions accept any `now`
1129    /// with no ordering check, while [`RunnerAttempt::from_persisted`] refuses
1130    /// `last_state_change_at < created_at` and `terminal_at < created_at`. So the
1131    /// domain can build in memory a value its own loader will not accept. The
1132    /// realistic trigger is not a hand-edited row: it is the wall clock stepping
1133    /// backwards between two transitions — an NTP correction, a VM snapshot
1134    /// restore, or an operator changing the clock on the home PC this product
1135    /// targets.
1136    ///
1137    /// **Why this clamps rather than rejecting or quarantining.** Rejecting
1138    /// leaves a row that neither this store nor the agent can ever load again;
1139    /// the attempt's capacity slot goes with it, its runtime directory is never
1140    /// cleaned, and there is no repair path short of an operator editing SQLite
1141    /// by hand. Quarantining preserves the evidence but has the same operational
1142    /// effect — the attempt becomes invisible to recovery, so a live child
1143    /// process goes unsupervised. Clamping costs one bounded thing: a recovery
1144    /// timeout measured from `created_at` rather than from a slightly earlier
1145    /// instant, which errs towards concluding a stuck attempt and giving its slot
1146    /// back, never towards holding it.
1147    ///
1148    /// **It is applied on the way in as well as on the way out.** Repairing only
1149    /// on load would still write the unloadable row, and the next reader — a
1150    /// different build, a different tool, an operator's `sqlite3` — would have to
1151    /// know about this function to make sense of it. Repairing on write keeps the
1152    /// database self-consistent; repairing on load handles rows this build did
1153    /// not write.
1154    ///
1155    /// **The clean fix is upstream and is not `b2`'s to make.** If
1156    /// `RunnerAttempt::move_to` and `::conclude` clamped `now` to at least
1157    /// `created_at`, the out-of-order value could not exist in memory at all and
1158    /// `from_persisted`'s strictness would be exactly right with nothing here to
1159    /// compensate for it. That is a change to `crates/domain/src/attempt.rs`,
1160    /// which `b1` owns; it is reported rather than worked around silently, and
1161    /// this is the defensive measure in the meantime.
1162    fn normalise(&self, mut fields: PersistedAttempt) -> PersistedAttempt {
1163        if fields.last_state_change_at < fields.created_at {
1164            tracing::warn!(
1165                attempt = %fields.id,
1166                created_at = %fields.created_at,
1167                last_state_change_at = %fields.last_state_change_at,
1168                "attempt last_state_change_at precedes created_at; the host clock \
1169                 stepped backwards. Clamping to created_at so the attempt stays \
1170                 recoverable; its recovery timeouts now measure from allocation."
1171            );
1172            fields.last_state_change_at = fields.created_at;
1173            self.clock_skew_repairs.fetch_add(1, Ordering::Relaxed);
1174        }
1175        if let Some(terminal_at) = fields.terminal_at
1176            && terminal_at < fields.created_at
1177        {
1178            tracing::warn!(
1179                attempt = %fields.id,
1180                created_at = %fields.created_at,
1181                terminal_at = %terminal_at,
1182                "attempt terminal_at precedes created_at; the host clock stepped \
1183                 backwards. Clamping to created_at."
1184            );
1185            fields.terminal_at = Some(fields.created_at);
1186            self.clock_skew_repairs.fetch_add(1, Ordering::Relaxed);
1187        }
1188        fields
1189    }
1190
1191    fn update_policy_confirming_active_count_with(
1192        &self,
1193        policy: &ScalePolicy,
1194        expected_revision: u64,
1195        expected_active: u16,
1196        after_write_fence: impl FnOnce(),
1197    ) -> Result<(), StoreError> {
1198        self.update_policy_confirming_count_with(
1199            policy,
1200            expected_revision,
1201            expected_active,
1202            CountedAttempts::Active,
1203            after_write_fence,
1204        )
1205    }
1206
1207    /// The shared body of both guarded policy writes.
1208    ///
1209    /// The two differ in one thing — which attempts they count — and everything
1210    /// else about them has to be identical: the same column list, the same
1211    /// IMMEDIATE fence, the same "diagnose only after the write matched nothing"
1212    /// ordering. Written twice they would drift, and the drift would be
1213    /// invisible until a column added to one UPDATE went missing from the other.
1214    /// [`CountedAttempts`] carries the only difference, as a closed enum whose
1215    /// two SQL fragments are constants rather than caller-supplied text.
1216    fn update_policy_confirming_count_with(
1217        &self,
1218        policy: &ScalePolicy,
1219        expected_revision: u64,
1220        expected_count: u16,
1221        counted: CountedAttempts,
1222        after_write_fence: impl FnOnce(),
1223    ) -> Result<(), StoreError> {
1224        let fields = policy.to_persisted();
1225        let mut params = policy_params(&fields)?;
1226        params.push((
1227            ":expected_revision",
1228            int(u64_to_sql("the expected revision", expected_revision)?),
1229        ));
1230        params.push((":expected_count", int(i64::from(expected_count))));
1231
1232        let count_sql = counted.count_sql();
1233        let mut conn = self.lock();
1234        // IMMEDIATE fences every attempt insert/update/delete on every other
1235        // connection before the count predicate is evaluated. The count and
1236        // revision predicates are part of the UPDATE itself, so there is no
1237        // check-to-write interval inside this transaction either.
1238        let tx = conn.transaction_with_behavior(TransactionBehavior::Immediate)?;
1239        after_write_fence();
1240        let changed = tx.execute(
1241            &format!(
1242                "UPDATE policies SET
1243                     target_scope    = :target_scope,
1244                     target_slug     = :target_slug,
1245                     installation_id = :installation_id,
1246                     host_id         = :host_id,
1247                     requested_host_label = :requested_host_label,
1248                     routing_labels  = :routing_labels,
1249                     min_capacity    = :min_capacity,
1250                     max_capacity    = :max_capacity,
1251                     enabled         = :enabled,
1252                     state           = :state,
1253                     cache_policy    = :cache_policy,
1254                     workspace_mode  = :workspace_mode,
1255                     workspace_path  = :workspace_path,
1256                     revision        = :revision
1257                 WHERE id = :id
1258                   AND revision = :expected_revision
1259                   AND :expected_count = ({count_sql})"
1260            ),
1261            &bind(&params)[..],
1262        )?;
1263        if changed == 0 {
1264            let revision_conflict = conflict_or_missing(&tx, fields.id, expected_revision)?;
1265            match revision_conflict {
1266                StoreError::StaleRevision {
1267                    expected, found, ..
1268                } if expected == found => {
1269                    let found: i64 = tx.query_row(
1270                        &count_sql,
1271                        named_params! { ":id": uuid_text(fields.id.as_uuid()) },
1272                        |row| row.get(0),
1273                    )?;
1274                    return Err(counted.count_changed(
1275                        fields.id,
1276                        expected_count,
1277                        clamped_count(found),
1278                    ));
1279                }
1280                other => return Err(other),
1281            }
1282        }
1283        tx.commit()?;
1284        Ok(())
1285    }
1286
1287    fn attempt_from_row(&self, row: &Row<'_>) -> Result<RunnerAttempt, StoreError> {
1288        let fields = self.normalise(persisted_attempt_from_row(row)?);
1289        let id = fields.id;
1290        RunnerAttempt::from_persisted(fields)
1291            .map_err(|source| StoreError::CorruptAttempt { id, source })
1292    }
1293
1294    /// Every attempt satisfying `predicate`, oldest first.
1295    ///
1296    /// The four attempt queries differ only in their `WHERE` fragment and their
1297    /// bindings; sharing the body is what keeps them agreeing on the ordering
1298    /// and, more importantly, on going through [`Self::attempt_from_row`] — a
1299    /// query that decoded a row itself would skip the clock-skew repair and the
1300    /// load-time revalidation every other read applies.
1301    fn attempts_where(
1302        &self,
1303        predicate: &str,
1304        params: &[(&str, &dyn ToSql)],
1305    ) -> Result<Vec<RunnerAttempt>, StoreError> {
1306        let conn = self.lock();
1307        let mut stmt = conn.prepare(&format!(
1308            "SELECT * FROM attempts WHERE {predicate} ORDER BY created_at, id"
1309        ))?;
1310        let mut rows = stmt.query(params)?;
1311        let mut out = Vec::new();
1312        while let Some(row) = rows.next()? {
1313            out.push(self.attempt_from_row(row)?);
1314        }
1315        Ok(out)
1316    }
1317
1318    /// Every attempt of one policy also satisfying `and`, oldest first.
1319    ///
1320    /// The binding is the part the four per-policy queries must not restate:
1321    /// the parameter name and the `&dyn ToSql` cast have to agree across all of
1322    /// them, and the only thing that actually varies between them is the
1323    /// fragment `and` supplies.
1324    fn attempts_of_policy(
1325        &self,
1326        policy_id: PolicyId,
1327        and: Option<&str>,
1328    ) -> Result<Vec<RunnerAttempt>, StoreError> {
1329        let id = uuid_text(policy_id.as_uuid());
1330        let predicate = and.map_or_else(
1331            || "policy_id = :policy_id".to_string(),
1332            |and| format!("policy_id = :policy_id AND {and}"),
1333        );
1334        self.attempts_where(&predicate, &[(":policy_id", &id as &dyn ToSql)])
1335    }
1336}
1337
1338impl Store for SqliteStore {
1339    fn put_host(&self, host: &Host) -> Result<(), StoreError> {
1340        let conn = self.lock();
1341        conn.execute(
1342            "INSERT INTO hosts (
1343                 id, display_name, os, architecture, host_capacity,
1344                 service_start_mode, refresh_interval_secs, runner_root_override,
1345                 created_at
1346             ) VALUES (
1347                 :id, :display_name, :os, :architecture, :host_capacity,
1348                 :service_start_mode, :refresh_interval_secs, :runner_root_override,
1349                 :created_at
1350             )
1351             ON CONFLICT(id) DO UPDATE SET
1352                 display_name          = excluded.display_name,
1353                 os                    = excluded.os,
1354                 architecture          = excluded.architecture,
1355                 host_capacity         = excluded.host_capacity,
1356                 service_start_mode    = excluded.service_start_mode,
1357                 refresh_interval_secs = excluded.refresh_interval_secs,
1358                 runner_root_override  = excluded.runner_root_override,
1359                 created_at            = excluded.created_at",
1360            // `created_at` **is** in this DO UPDATE list, and `record_attempt`
1361            // deliberately leaves it out of its own. The asymmetry is intended
1362            // and the two columns are not the same kind of thing.
1363            //
1364            // An attempt's `created_at` is a domain fact with a rule attached:
1365            // "created_at never moves", enforced by
1366            // `RunnerAttempt::from_persisted`, which refuses a row whose other
1367            // timestamps precede it. Journal writes happen repeatedly over one
1368            // attempt's life, so excluding the column is what keeps the value
1369            // written at allocation authoritative.
1370            //
1371            // A host's `created_at` is the record of when this host was
1372            // registered, and `put_host` is a whole-record upsert of a value the
1373            // caller assembled -- there is no partial-update path and no
1374            // ordering rule against it. Writing back what the caller holds keeps
1375            // the row equal to the `Host` it was given, which is what
1376            // `a_host_round_trips_byte_identically_in_every_configuration`
1377            // asserts. Excluding it would silently discard a correction an
1378            // operator made on purpose.
1379            //
1380            // `runner_root_override` is written here for the same reason —
1381            // this is a whole-record upsert and the row must equal the `Host` it
1382            // was handed — and that is precisely why it is *not* how `host
1383            // set-runtime-root` writes. A caller that read a `Host`, changed the
1384            // override on it and called this would also write back the capacity
1385            // and service mode it read, rolling back anything that changed in
1386            // between. `Store::set_runner_root_override` is the targeted,
1387            // fenced mutation for that command.
1388            named_params! {
1389                ":id": uuid_text(host.id.as_uuid()),
1390                ":display_name": host.display_name.as_str(),
1391                ":os": token(&host.os),
1392                ":architecture": token(&host.architecture),
1393                ":host_capacity": i64::from(host.host_capacity.get()),
1394                ":service_start_mode": token(&host.service_start_mode),
1395                ":refresh_interval_secs": i64::from(host.refresh_interval.as_secs()),
1396                ":runner_root_override": host
1397                    .runner_root_override
1398                    .as_ref()
1399                    .map(LocalAbsolutePath::as_str),
1400                ":created_at": timestamp_to_text(host.created_at),
1401            },
1402        )?;
1403        Ok(())
1404    }
1405
1406    fn host(&self, id: HostId) -> Result<Option<Host>, StoreError> {
1407        let conn = self.lock();
1408        let mut stmt = conn.prepare("SELECT * FROM hosts WHERE id = :id")?;
1409        let mut rows = stmt.query(named_params! { ":id": uuid_text(id.as_uuid()) })?;
1410        match rows.next()? {
1411            Some(row) => Ok(Some(host_from_row(row)?)),
1412            None => Ok(None),
1413        }
1414    }
1415
1416    fn hosts(&self) -> Result<Vec<Host>, StoreError> {
1417        let conn = self.lock();
1418        let mut stmt = conn.prepare("SELECT * FROM hosts ORDER BY created_at, id")?;
1419        let mut rows = stmt.query([])?;
1420        let mut out = Vec::new();
1421        while let Some(row) = rows.next()? {
1422            out.push(host_from_row(row)?);
1423        }
1424        Ok(out)
1425    }
1426
1427    fn set_runner_root_override(
1428        &self,
1429        id: HostId,
1430        expected: Option<&LocalAbsolutePath>,
1431        new_root: Option<&LocalAbsolutePath>,
1432        expected_uncleaned: u16,
1433    ) -> Result<(), StoreError> {
1434        // Host-wide, not scoped to this host's policies: see the contract note
1435        // on `Store::uncleaned_ephemeral_attempts` for why an attempt that
1436        // outlived its policy row still has to count.
1437        let count_sql = attempt_count_sql(&uncleaned_of_kind_sql(WorkspaceKind::Ephemeral));
1438
1439        let mut conn = self.lock();
1440        // IMMEDIATE for the reason `update_policy` gives: the diagnosis below
1441        // reads inside this transaction, so the write lock is taken up front
1442        // rather than being upgraded mid-transaction into a `SQLITE_BUSY` the
1443        // busy handler refuses to retry. It is also what fences a concurrent
1444        // attempt write out of the count predicate.
1445        let tx = conn.transaction_with_behavior(TransactionBehavior::Immediate)?;
1446        // `IS` rather than `=`: the override is NULL for every host that has
1447        // never been configured, and `NULL = NULL` is NULL, so `=` would refuse
1448        // the most ordinary mutation there is -- the first `host
1449        // set-runtime-root` on a fresh install.
1450        let changed = tx.execute(
1451            &format!(
1452                "UPDATE hosts SET runner_root_override = :new_root
1453                  WHERE id = :id
1454                    AND runner_root_override IS :expected
1455                    AND :expected_uncleaned = ({count_sql})"
1456            ),
1457            named_params! {
1458                ":id": uuid_text(id.as_uuid()),
1459                ":new_root": new_root.map(LocalAbsolutePath::as_str),
1460                ":expected": expected.map(LocalAbsolutePath::as_str),
1461                ":expected_uncleaned": i64::from(expected_uncleaned),
1462            },
1463        )?;
1464        if changed == 0 {
1465            // The transaction is dropped, and therefore rolled back, on return.
1466            let stored: Option<Option<String>> = tx
1467                .query_row(
1468                    "SELECT runner_root_override FROM hosts WHERE id = :id",
1469                    named_params! { ":id": uuid_text(id.as_uuid()) },
1470                    |row| row.get(0),
1471                )
1472                .optional()?;
1473            let Some(stored) = stored else {
1474                return Err(StoreError::NotFound {
1475                    what: "host",
1476                    id: id.to_string(),
1477                });
1478            };
1479            // The override is compared first, so that when both moved the
1480            // operator is told about the one that names a directory. The stored
1481            // text is echoed as it is rather than re-parsed: a row this build
1482            // would refuse to load is exactly the row whose value the message
1483            // has to show.
1484            if stored.as_deref() != expected.map(LocalAbsolutePath::as_str) {
1485                return Err(StoreError::RunnerRootChanged {
1486                    id,
1487                    expected: render_root(expected.map(LocalAbsolutePath::as_str)),
1488                    found: render_root(stored.as_deref()),
1489                });
1490            }
1491            let found: i64 = tx.query_row(&count_sql, [], |row| row.get(0))?;
1492            return Err(StoreError::UncleanedCountChanged {
1493                subject: format!("host {id}"),
1494                expected: expected_uncleaned,
1495                found: clamped_count(found),
1496            });
1497        }
1498        tx.commit()?;
1499        Ok(())
1500    }
1501
1502    fn insert_policy(&self, policy: &ScalePolicy) -> Result<(), StoreError> {
1503        let fields = policy.to_persisted();
1504        let params = policy_params(&fields)?;
1505        let conn = self.lock();
1506        conn.execute(
1507            "INSERT INTO policies (
1508                 id, target_scope, target_slug, installation_id, host_id,
1509                 requested_host_label, routing_labels, min_capacity, max_capacity, enabled, state,
1510                 cache_policy, workspace_mode, workspace_path, revision
1511             ) VALUES (
1512                 :id, :target_scope, :target_slug, :installation_id, :host_id,
1513                 :requested_host_label, :routing_labels, :min_capacity, :max_capacity, :enabled, :state,
1514                 :cache_policy, :workspace_mode, :workspace_path, :revision
1515             )",
1516            &bind(&params)[..],
1517        )
1518        .map_err(|source| {
1519            if is_constraint_violation(&source) {
1520                StoreError::AlreadyExists {
1521                    what: "policy",
1522                    id: fields.id.to_string(),
1523                }
1524            } else {
1525                StoreError::Sqlite(source)
1526            }
1527        })?;
1528        Ok(())
1529    }
1530
1531    fn update_policy(
1532        &self,
1533        policy: &ScalePolicy,
1534        expected_revision: u64,
1535    ) -> Result<(), StoreError> {
1536        let fields = policy.to_persisted();
1537        let mut params = policy_params(&fields)?;
1538        params.push((
1539            ":expected_revision",
1540            int(u64_to_sql("the expected revision", expected_revision)?),
1541        ));
1542
1543        let mut conn = self.lock();
1544        // IMMEDIATE, not the default DEFERRED -- but not for the reason this
1545        // comment used to give, which was checkable and wrong.
1546        //
1547        // It claimed that DEFERRED would make two racing writers produce
1548        // SQLITE_BUSY on the loser instead of a clean stale-revision answer.
1549        // That hazard is real in SQLite (`SQLITE_BUSY_SNAPSHOT`, which the busy
1550        // handler deliberately refuses to retry, because retrying would hand
1551        // the reader a snapshot that has already moved) but it needs the *read*
1552        // to be inside the transaction. Here it is not: the caller read the
1553        // policy through `Store::policy`, in a separate implicit transaction
1554        // that has already ended, and this transaction runs the UPDATE as its
1555        // first statement. So the ordinary busy handler applies, the loser waits
1556        // out `busy_timeout` and then matches against the winner's revision and
1557        // gets `StaleRevision`. Measured: with `Deferred` here,
1558        // `two_concurrent_writers_race_and_exactly_one_wins` passes 15 runs out
1559        // of 15.
1560        //
1561        // What IMMEDIATE buys is that the paragraph above becomes true the day
1562        // the read moves inside -- a re-read to report the current revision, a
1563        // check-then-write, a batched multi-policy update. That is an ordinary
1564        // refactor whose failure mode is a raw `database is locked` in an
1565        // operator's face instead of the conflict this store promises to
1566        // distinguish, and no test here would catch it, because both writers
1567        // have to interleave *within* the transaction to show it. Taking the
1568        // write lock up front costs one uncontended acquisition and removes the
1569        // hazard before anyone can introduce it.
1570        let tx = conn.transaction_with_behavior(TransactionBehavior::Immediate)?;
1571        let changed = tx.execute(
1572            "UPDATE policies SET
1573                 target_scope    = :target_scope,
1574                 target_slug     = :target_slug,
1575                 installation_id = :installation_id,
1576                 host_id         = :host_id,
1577                 requested_host_label = :requested_host_label,
1578                 routing_labels  = :routing_labels,
1579                 min_capacity    = :min_capacity,
1580                 max_capacity    = :max_capacity,
1581                 enabled         = :enabled,
1582                 state           = :state,
1583                 cache_policy    = :cache_policy,
1584                 workspace_mode  = :workspace_mode,
1585                 workspace_path  = :workspace_path,
1586                 revision        = :revision
1587             WHERE id = :id AND revision = :expected_revision",
1588            &bind(&params)[..],
1589        )?;
1590        if changed == 0 {
1591            // The transaction is dropped, and therefore rolled back, on return.
1592            return Err(conflict_or_missing(&tx, fields.id, expected_revision)?);
1593        }
1594        tx.commit()?;
1595        Ok(())
1596    }
1597
1598    fn update_policy_confirming_active_count(
1599        &self,
1600        policy: &ScalePolicy,
1601        expected_revision: u64,
1602        expected_active: u16,
1603    ) -> Result<(), StoreError> {
1604        self.update_policy_confirming_active_count_with(
1605            policy,
1606            expected_revision,
1607            expected_active,
1608            || {},
1609        )
1610    }
1611
1612    fn update_policy_confirming_uncleaned_count(
1613        &self,
1614        policy: &ScalePolicy,
1615        expected_revision: u64,
1616        expected_uncleaned: u16,
1617    ) -> Result<(), StoreError> {
1618        self.update_policy_confirming_count_with(
1619            policy,
1620            expected_revision,
1621            expected_uncleaned,
1622            CountedAttempts::Uncleaned,
1623            || {},
1624        )
1625    }
1626
1627    fn remove_policy(&self, id: PolicyId, expected_revision: u64) -> Result<(), StoreError> {
1628        let mut conn = self.lock();
1629        let tx = conn.transaction_with_behavior(TransactionBehavior::Immediate)?;
1630        let changed = tx.execute(
1631            "DELETE FROM policies WHERE id = :id AND revision = :expected_revision",
1632            named_params! {
1633                ":id": uuid_text(id.as_uuid()),
1634                ":expected_revision": u64_to_sql("the expected revision", expected_revision)?,
1635            },
1636        )?;
1637        if changed == 0 {
1638            return Err(conflict_or_missing(&tx, id, expected_revision)?);
1639        }
1640        tx.commit()?;
1641        Ok(())
1642    }
1643
1644    fn policy(&self, id: PolicyId) -> Result<Option<ScalePolicy>, StoreError> {
1645        let conn = self.lock();
1646        let mut stmt = conn.prepare("SELECT * FROM policies WHERE id = :id")?;
1647        let mut rows = stmt.query(named_params! { ":id": uuid_text(id.as_uuid()) })?;
1648        match rows.next()? {
1649            Some(row) => Ok(Some(policy_from_row(row)?)),
1650            None => Ok(None),
1651        }
1652    }
1653
1654    fn policies(&self) -> Result<Vec<ScalePolicy>, StoreError> {
1655        let conn = self.lock();
1656        let mut stmt = conn.prepare("SELECT * FROM policies ORDER BY id")?;
1657        let mut rows = stmt.query([])?;
1658        let mut out = Vec::new();
1659        while let Some(row) = rows.next()? {
1660            out.push(policy_from_row(row)?);
1661        }
1662        Ok(out)
1663    }
1664
1665    fn record_attempt(&self, attempt: &RunnerAttempt) -> Result<(), StoreError> {
1666        let fields = self.normalise(attempt.to_persisted());
1667        let params = attempt_params(&fields)?;
1668        let conn = self.lock();
1669        conn.execute(
1670            "INSERT INTO attempts (
1671                 id, policy_id, github_runner_id, state, outcome, process_id,
1672                 runtime_path, workspace_mode, workspace_slot,
1673                 created_at, terminal_at, last_state_change_at
1674             ) VALUES (
1675                 :id, :policy_id, :github_runner_id, :state, :outcome, :process_id,
1676                 :runtime_path, :workspace_mode, :workspace_slot,
1677                 :created_at, :terminal_at, :last_state_change_at
1678             )
1679             ON CONFLICT(id) DO UPDATE SET
1680                 policy_id            = excluded.policy_id,
1681                 github_runner_id     = excluded.github_runner_id,
1682                 state                = excluded.state,
1683                 outcome              = excluded.outcome,
1684                 process_id           = excluded.process_id,
1685                 runtime_path         = excluded.runtime_path,
1686                 terminal_at          = excluded.terminal_at,
1687                 last_state_change_at = excluded.last_state_change_at",
1688            // `created_at` is absent from the DO UPDATE list on purpose: the
1689            // domain says it never moves, so the value written at allocation is
1690            // the one that stands and no later journal write can rewrite it.
1691            // `put_host` above does the opposite with its own `created_at`, and
1692            // its comment says why the two are not the same case.
1693            //
1694            // `workspace_mode` and `workspace_slot` are absent for the same
1695            // reason and a sharper one. They are the *immutable allocation
1696            // fact* -- `02-target-architecture.md`: "The workspace kind and slot
1697            // number tell recovery which cleanup algorithm is legal. Neither may
1698            // change after allocation" -- and `RunnerAttempt` exposes no
1699            // mutator for either. Leaving them out of the update makes that a
1700            // property of the journal too, so the row a crash leaves behind
1701            // still names the algorithm the directory was created under, and no
1702            // later write can move a live lease onto a different slot.
1703            &bind(&params)[..],
1704        )
1705        .map_err(|source| slot_lease_error(attempt, source))?;
1706        Ok(())
1707    }
1708
1709    fn attempt(&self, id: AttemptId) -> Result<Option<RunnerAttempt>, StoreError> {
1710        let conn = self.lock();
1711        let mut stmt = conn.prepare("SELECT * FROM attempts WHERE id = :id")?;
1712        let mut rows = stmt.query(named_params! { ":id": uuid_text(id.as_uuid()) })?;
1713        match rows.next()? {
1714            Some(row) => Ok(Some(self.attempt_from_row(row)?)),
1715            None => Ok(None),
1716        }
1717    }
1718
1719    fn attempts(&self) -> Result<Vec<RunnerAttempt>, StoreError> {
1720        let conn = self.lock();
1721        let mut stmt = conn.prepare("SELECT * FROM attempts ORDER BY created_at, id")?;
1722        let mut rows = stmt.query([])?;
1723        let mut out = Vec::new();
1724        while let Some(row) = rows.next()? {
1725            out.push(self.attempt_from_row(row)?);
1726        }
1727        Ok(out)
1728    }
1729
1730    fn attempts_for_policy(&self, policy_id: PolicyId) -> Result<Vec<RunnerAttempt>, StoreError> {
1731        self.attempts_of_policy(policy_id, None)
1732    }
1733
1734    fn active_attempts_for_policy(
1735        &self,
1736        policy_id: PolicyId,
1737    ) -> Result<Vec<RunnerAttempt>, StoreError> {
1738        self.attempts_of_policy(policy_id, Some(&active_sql()))
1739    }
1740
1741    fn uncleaned_attempts_for_policy(
1742        &self,
1743        policy_id: PolicyId,
1744    ) -> Result<Vec<RunnerAttempt>, StoreError> {
1745        self.attempts_of_policy(policy_id, Some(&uncleaned_sql()))
1746    }
1747
1748    fn slot_leases_for_policy(
1749        &self,
1750        policy_id: PolicyId,
1751    ) -> Result<Vec<RunnerAttempt>, StoreError> {
1752        self.attempts_of_policy(
1753            policy_id,
1754            Some(&uncleaned_of_kind_sql(WorkspaceKind::Persistent)),
1755        )
1756    }
1757
1758    fn uncleaned_ephemeral_attempts(&self) -> Result<Vec<RunnerAttempt>, StoreError> {
1759        self.attempts_where(&uncleaned_of_kind_sql(WorkspaceKind::Ephemeral), &[])
1760    }
1761
1762    fn remove_attempt(&self, id: AttemptId) -> Result<bool, StoreError> {
1763        let conn = self.lock();
1764        let changed = conn.execute(
1765            "DELETE FROM attempts WHERE id = :id",
1766            named_params! { ":id": uuid_text(id.as_uuid()) },
1767        )?;
1768        Ok(changed > 0)
1769    }
1770}
1771
1772// ---------------------------------------------------------------------------
1773// Domain -> row
1774// ---------------------------------------------------------------------------
1775
1776/// Named bindings for one statement.
1777///
1778/// Owned [`Value`]s rather than references so that the insert and the update can
1779/// share one binding list. They bind the same twelve columns, and a duplicated
1780/// `named_params!` between them is exactly how two statements drift into
1781/// disagreeing about which column holds what.
1782type NamedParams = Vec<(&'static str, Value)>;
1783
1784fn bind(params: &NamedParams) -> Vec<(&str, &dyn ToSql)> {
1785    params
1786        .iter()
1787        .map(|(name, value)| (*name, value as &dyn ToSql))
1788        .collect()
1789}
1790
1791fn text(value: impl Into<String>) -> Value {
1792    Value::Text(value.into())
1793}
1794
1795fn int(value: i64) -> Value {
1796    Value::Integer(value)
1797}
1798
1799fn opt_text(value: Option<String>) -> Value {
1800    value.map_or(Value::Null, Value::Text)
1801}
1802
1803fn opt_int(value: Option<i64>) -> Value {
1804    value.map_or(Value::Null, Value::Integer)
1805}
1806
1807fn policy_params(fields: &PersistedPolicy) -> Result<NamedParams, StoreError> {
1808    Ok(vec![
1809        (":id", text(uuid_text(fields.id.as_uuid()))),
1810        (":target_scope", text(token(&fields.target.scope()))),
1811        (":target_slug", text(fields.target.slug())),
1812        (
1813            ":installation_id",
1814            int(u64_to_sql(
1815                "policies.installation_id",
1816                fields.installation_id,
1817            )?),
1818        ),
1819        (":host_id", text(uuid_text(fields.host_id.as_uuid()))),
1820        (
1821            ":requested_host_label",
1822            text(fields.requested_host_label.to_string()),
1823        ),
1824        (
1825            ":routing_labels",
1826            opt_text(fields.routing_labels.as_ref().map(json)),
1827        ),
1828        (":min_capacity", int(i64::from(fields.min_capacity))),
1829        (
1830            ":max_capacity",
1831            opt_int(fields.max_capacity.map(|m| i64::from(m.get()))),
1832        ),
1833        (":enabled", int(i64::from(fields.enabled))),
1834        (":state", text(token(&fields.state))),
1835        (":cache_policy", text(token(&fields.cache_policy))),
1836        (":workspace_mode", text(token(&fields.workspace_kind))),
1837        (
1838            ":workspace_path",
1839            opt_text(
1840                fields
1841                    .workspace_root
1842                    .as_ref()
1843                    .map(|root| root.as_str().to_string()),
1844            ),
1845        ),
1846        (
1847            ":revision",
1848            int(u64_to_sql("policies.revision", fields.revision)?),
1849        ),
1850    ])
1851}
1852
1853fn attempt_params(fields: &PersistedAttempt) -> Result<NamedParams, StoreError> {
1854    let runtime_path =
1855        fields
1856            .runtime_path
1857            .to_str()
1858            .ok_or_else(|| StoreError::UnrepresentablePath {
1859                attempt: fields.id,
1860                path: fields.runtime_path.clone(),
1861            })?;
1862    Ok(vec![
1863        (":id", text(uuid_text(fields.id.as_uuid()))),
1864        (":policy_id", text(uuid_text(fields.policy_id.as_uuid()))),
1865        (
1866            ":github_runner_id",
1867            opt_int(
1868                fields
1869                    .github_runner_id
1870                    .map(|id| u64_to_sql("attempts.github_runner_id", id))
1871                    .transpose()?,
1872            ),
1873        ),
1874        (":state", text(token(&fields.state))),
1875        (":outcome", opt_text(fields.outcome.as_ref().map(json))),
1876        (":process_id", opt_int(fields.process_id.map(i64::from))),
1877        (":runtime_path", text(runtime_path)),
1878        (":workspace_mode", text(token(&fields.workspace_kind))),
1879        (
1880            ":workspace_slot",
1881            opt_int(fields.workspace_slot.map(i64::from)),
1882        ),
1883        (":created_at", text(timestamp_to_text(fields.created_at))),
1884        (
1885            ":terminal_at",
1886            opt_text(fields.terminal_at.map(timestamp_to_text)),
1887        ),
1888        (
1889            ":last_state_change_at",
1890            text(timestamp_to_text(fields.last_state_change_at)),
1891        ),
1892    ])
1893}
1894
1895// ---------------------------------------------------------------------------
1896// Row -> domain
1897// ---------------------------------------------------------------------------
1898
1899fn host_from_row(row: &Row<'_>) -> Result<Host, StoreError> {
1900    const TABLE: &str = "hosts";
1901    let id = HostId::from_uuid(uuid_column(row, TABLE, "id")?);
1902    let key = id.to_string();
1903
1904    let display_name: String = row.get("display_name")?;
1905    let os: Os = token_column(row, TABLE, "os", &key)?;
1906    let architecture: Arch = token_column(row, TABLE, "architecture", &key)?;
1907    let host_capacity = NonZeroU16::new(u16_column(row, TABLE, "host_capacity", &key)?).ok_or(
1908        StoreError::CorruptColumn {
1909            table: TABLE,
1910            column: "host_capacity",
1911            id: key.clone(),
1912            value: "0".to_string(),
1913            expected: "a non-zero capacity; a host that declares zero is not a \
1914                       configured host but a disabled one",
1915        },
1916    )?;
1917    let service_start_mode: StartMode = token_column(row, TABLE, "service_start_mode", &key)?;
1918    let refresh_interval_secs = u16_column(row, TABLE, "refresh_interval_secs", &key)?;
1919    // Re-parsed through the *native* entry point, so a hand-edited UNC share, a
1920    // bare drive root, a `..` component, or a Windows path in a database opened
1921    // on Linux fails closed here rather than becoming a directory this product
1922    // creates runners under (D10). NULL is the ordinary state: it means "use the
1923    // platform default", which is resolved at runtime and is deliberately not
1924    // stored.
1925    let runner_root_override = row
1926        .get::<_, Option<String>>("runner_root_override")?
1927        .map(LocalAbsolutePath::new)
1928        .transpose()
1929        .map_err(|source| StoreError::CorruptHostWorkspace {
1930            id,
1931            source: WorkspaceError::from(source),
1932        })?;
1933    let created_at = timestamp_column(row, TABLE, "created_at", &key)?;
1934
1935    // `Host::new` re-runs the display-name rule and `RefreshInterval::from_secs`
1936    // re-runs the 30-second floor, so a hand-edited row cannot install a host
1937    // with a blank name or one that polls every second.
1938    let mut host = Host::new(
1939        id,
1940        &display_name,
1941        os,
1942        architecture,
1943        host_capacity,
1944        created_at,
1945    )
1946    .map_err(|source| StoreError::CorruptHost { id, source })?;
1947    host.service_start_mode = service_start_mode;
1948    host.refresh_interval = RefreshInterval::from_secs(refresh_interval_secs)
1949        .map_err(|source| StoreError::CorruptHost { id, source })?;
1950    host.runner_root_override = runner_root_override;
1951    Ok(host)
1952}
1953
1954fn policy_from_row(row: &Row<'_>) -> Result<ScalePolicy, StoreError> {
1955    const TABLE: &str = "policies";
1956    let id = PolicyId::from_uuid(uuid_column(row, TABLE, "id")?);
1957    let key = id.to_string();
1958
1959    let scope: TargetScope = token_column(row, TABLE, "target_scope", &key)?;
1960    let slug: String = row.get("target_slug")?;
1961    // Rebuilt through the real constructors, so GitHub's naming rules run again
1962    // and a scope/slug pair that cannot exist — `organization` holding `o/r` —
1963    // is refused rather than loaded as an organization with a slash in its name.
1964    let target = match scope {
1965        TargetScope::Repository => ScaleTarget::repository(&slug),
1966        TargetScope::Organization => ScaleTarget::organization(&slug),
1967    }
1968    .map_err(|source| StoreError::CorruptPolicy {
1969        id,
1970        source: PolicyError::Invalid(source),
1971    })?;
1972
1973    let routing_labels: Option<RoutingLabels> =
1974        json_column(row, TABLE, "routing_labels", &key, "a routing label set")?;
1975    let max_capacity = match u16_option_column(row, TABLE, "max_capacity", &key)? {
1976        Some(raw) => Some(NonZeroU16::new(raw).ok_or(StoreError::CorruptColumn {
1977            table: TABLE,
1978            column: "max_capacity",
1979            id: key.clone(),
1980            value: "0".to_string(),
1981            expected: "a non-zero ceiling; an autoscale policy that may start no \
1982                       runner is a monitor-only policy and stores NULL here",
1983        })?),
1984        None => None,
1985    };
1986
1987    let fields = PersistedPolicy {
1988        id,
1989        target,
1990        installation_id: u64_column(row, TABLE, "installation_id", &key)?,
1991        host_id: HostId::from_uuid(uuid_column(row, TABLE, "host_id")?),
1992        requested_host_label: HostLabel::new(row.get::<_, String>("requested_host_label")?)
1993            .map_err(|source| StoreError::CorruptPolicy {
1994                id,
1995                source: PolicyError::Invalid(source),
1996            })?,
1997        routing_labels,
1998        min_capacity: u16_column(row, TABLE, "min_capacity", &key)?,
1999        max_capacity,
2000        enabled: bool_column(row, TABLE, "enabled", &key)?,
2001        state: token_column::<PolicyState>(row, TABLE, "state", &key)?,
2002        cache_policy: token_column::<CachePolicy>(row, TABLE, "cache_policy", &key)?,
2003        // The two columns are read separately and paired by
2004        // `WorkspacePolicy::from_persisted` inside `ScalePolicy::from_persisted`
2005        // below, which is what refuses persistent-without-path,
2006        // ephemeral-with-path, and an organization policy claiming to retain a
2007        // workspace (D7). Migration 3 gave every historical row
2008        // `'ephemeral'`/NULL, which is the pair that rebuilds as
2009        // `WorkspacePolicy::Ephemeral`.
2010        workspace_kind: token_column::<WorkspaceKind>(row, TABLE, "workspace_mode", &key)?,
2011        workspace_root: row
2012            .get::<_, Option<String>>("workspace_path")?
2013            .map(LocalAbsolutePath::new)
2014            .transpose()
2015            .map_err(|source| StoreError::CorruptPolicy {
2016                id,
2017                source: PolicyError::Workspace(WorkspaceError::from(source)),
2018            })?,
2019        revision: u64_column(row, TABLE, "revision", &key)?,
2020    };
2021
2022    // D19's shape rules and `min <= max` run here, on every load.
2023    ScalePolicy::from_persisted(fields).map_err(|source| StoreError::CorruptPolicy { id, source })
2024}
2025
2026fn persisted_attempt_from_row(row: &Row<'_>) -> Result<PersistedAttempt, StoreError> {
2027    const TABLE: &str = "attempts";
2028    let id = AttemptId::from_uuid(uuid_column(row, TABLE, "id")?);
2029    let key = id.to_string();
2030
2031    let runtime_path: String = row.get("runtime_path")?;
2032    Ok(PersistedAttempt {
2033        id,
2034        policy_id: PolicyId::from_uuid(uuid_column(row, TABLE, "policy_id")?),
2035        github_runner_id: u64_option_column(row, TABLE, "github_runner_id", &key)?,
2036        state: token_column::<AttemptState>(row, TABLE, "state", &key)?,
2037        outcome: json_column::<AttemptOutcome>(row, TABLE, "outcome", &key, "an attempt outcome")?,
2038        process_id: u32_option_column(row, TABLE, "process_id", &key)?,
2039        runtime_path: PathBuf::from(runtime_path),
2040        // As for policies, the pair is rebuilt by the domain --
2041        // `AttemptWorkspace::from_persisted`, called first thing in
2042        // `RunnerAttempt::from_persisted` -- because it decides which cleanup
2043        // algorithm is legal on `runtime_path`. The slot is read as a raw `u16`
2044        // on purpose: a stored `0` is a refusal there rather than an
2045        // unrepresentable value that panics here.
2046        workspace_kind: token_column::<WorkspaceKind>(row, TABLE, "workspace_mode", &key)?,
2047        workspace_slot: u16_option_column(row, TABLE, "workspace_slot", &key)?,
2048        created_at: timestamp_column(row, TABLE, "created_at", &key)?,
2049        terminal_at: timestamp_option_column(row, TABLE, "terminal_at", &key)?,
2050        last_state_change_at: timestamp_column(row, TABLE, "last_state_change_at", &key)?,
2051    })
2052}
2053
2054/// Why a revision-guarded write matched no row: someone else won, or the row is
2055/// gone. Both are ordinary and the caller does different things about them.
2056fn conflict_or_missing(
2057    tx: &rusqlite::Transaction<'_>,
2058    id: PolicyId,
2059    expected: u64,
2060) -> Result<StoreError, StoreError> {
2061    let found: Option<i64> = tx
2062        .query_row(
2063            "SELECT revision FROM policies WHERE id = :id",
2064            named_params! { ":id": uuid_text(id.as_uuid()) },
2065            |row| row.get(0),
2066        )
2067        .optional()?;
2068    Ok(match found {
2069        // The revision is read raw here rather than through `u64_column`, so it
2070        // is the one place a corrupt value could slip past the check every other
2071        // column gets. It used to be coerced with `unwrap_or(0)`, which turned a
2072        // hand-edited `-1` into the message "written against revision 0, but the
2073        // stored revision is now 0" -- a self-contradiction that reads as a bug
2074        // in this code and tells an operator nothing about the row that actually
2075        // needs fixing.
2076        Some(found) => match u64::try_from(found) {
2077            Ok(found) => StoreError::StaleRevision {
2078                id,
2079                expected,
2080                found,
2081            },
2082            Err(_) => StoreError::CorruptColumn {
2083                table: "policies",
2084                column: "revision",
2085                id: id.to_string(),
2086                value: clip(&found.to_string()),
2087                expected: "a non-negative integer",
2088            },
2089        },
2090        None => StoreError::NotFound {
2091            what: "policy",
2092            id: id.to_string(),
2093        },
2094    })
2095}
2096
2097// ---------------------------------------------------------------------------
2098// Encoding helpers
2099// ---------------------------------------------------------------------------
2100
2101/// A unit enum's on-disk token, taken from the domain's own serde naming.
2102///
2103/// Deriving the token from `serde` rather than writing a `match` here means the
2104/// stored token and the JSON token cannot drift apart, and
2105/// `tests::the_on_disk_tokens_are_pinned` fixes the actual strings so a rename in
2106/// the domain breaks a test rather than silently changing the on-disk format.
2107///
2108/// # Panics
2109/// If `T` does not serialise to a JSON string. Every caller passes a unit-only
2110/// enum and the pinning test covers each one.
2111fn token<T: Serialize + fmt::Debug>(value: &T) -> String {
2112    match serde_json::to_value(value) {
2113        Ok(serde_json::Value::String(token)) => token,
2114        other => panic!(
2115            "{value:?} must serialise to a JSON string to be stored as a column \
2116             token, got {other:?}"
2117        ),
2118    }
2119}
2120
2121/// # Panics
2122/// If `T`'s `Serialize` fails, which for the types stored here would mean an
2123/// unrepresentable value rather than an I/O error.
2124fn json<T: Serialize + fmt::Debug>(value: &T) -> String {
2125    serde_json::to_string(value)
2126        .unwrap_or_else(|e| panic!("{value:?} must serialise to JSON for storage: {e}"))
2127}
2128
2129/// RFC 3339, always nanosecond precision, always `Z`.
2130///
2131/// Fixed width so the text sorts in instant order (`ORDER BY created_at` is a
2132/// real query here), and full precision so the round trip is exact: a format that
2133/// dropped sub-second digits would make `assert_eq!` on a reloaded [`Timestamp`]
2134/// fail for any value that came from the system clock.
2135fn timestamp_to_text(value: Timestamp) -> String {
2136    value.to_rfc3339_opts(chrono::SecondsFormat::Nanos, true)
2137}
2138
2139fn uuid_text(value: &Uuid) -> String {
2140    value.hyphenated().to_string()
2141}
2142
2143/// SQLite integers are signed 64-bit, so a `u64` above `i64::MAX` has no
2144/// representation.
2145///
2146/// This refuses rather than saturating. Saturating would write `i64::MAX` and
2147/// read `i64::MAX` back, so the value the caller stored and the value it later
2148/// loaded would differ with nothing to say so — and the path is reachable, not
2149/// theoretical: `RunnerAttempt::registered_idle` takes any `u64` as the GitHub
2150/// runner id, and `ScalePolicy` takes any `u64` as the installation id.
2151fn u64_to_sql(what: &'static str, value: u64) -> Result<i64, StoreError> {
2152    i64::try_from(value).map_err(|_| StoreError::UnrepresentableInteger { what, value })
2153}
2154
2155/// The most of a **constrained** column's payload an error message will ever
2156/// repeat.
2157///
2158/// Sixty characters identifies a value without reproducing it: a malformed
2159/// timestamp, an unrecognised token and a truncated UUID are all shorter than
2160/// this, and anything longer is a payload rather than a value.
2161///
2162/// It is not a limit that protects a *free-form* column, and it was once applied
2163/// to one. The private `FREE_FORM_COLUMNS` below carries the measurement of why
2164/// a fixed character budget is the wrong instrument there, and what replaced it.
2165pub const ECHO_LIMIT: usize = 60;
2166
2167/// The columns whose payload includes text the **agent captured from a
2168/// failure**, and which an error message therefore may not echo at all.
2169///
2170/// **Why this is a per-column list and not one budget for every column.**
2171/// [`ECHO_LIMIT`] is a fixed character budget with no idea which column it is
2172/// echoing, and over `attempts.outcome` that is not a small imprecision. The
2173/// column holds the JSON of [`AttemptOutcome`], and reaching the free-form
2174/// `FailureReason::Other(String)` inside it costs a serde prefix of
2175/// `{"outcome":"failed","reason":{"other":"` — thirty-nine characters — which
2176/// leaves roughly twenty-one of the caller's own string inside a sixty-character
2177/// budget. That is a partial echo of a forty-character `ghu_…` token, and a
2178/// **complete** echo of any secret shorter than about twenty-one characters. The
2179/// rationale the budget was written with — "anything longer is a payload rather
2180/// than a value" — quietly assumes secrets are long, and short ones are the case
2181/// it lets straight through.
2182///
2183/// **The second-order failure is worse than the first.** The argument for
2184/// clipping leaned on `d1`'s redacting log sink catching a prefixed token that
2185/// slipped out. But that sink matches on *shape*, and a token cut after
2186/// twenty-one characters may no longer have the shape it matches — so clipping
2187/// can take a token the sink would have redacted and hand it on as a fragment
2188/// the sink will not. A partial echo is not a safer echo here; it is an echo
2189/// with the downstream control removed.
2190///
2191/// **What is echoed instead.** For a column on this list,
2192/// [`StoreError::CorruptColumn`] reports position only — how many bytes the
2193/// column holds and, where serde recorded one, the position the parse gave up
2194/// at (see [`position_only`], which measures how often it does) — and none of
2195/// the payload. That costs nothing at every other column: `uuid_column`,
2196/// `token_column`, `parse_timestamp` and `current_version` all decode values
2197/// whose shape the *schema* fixes, so no captured text can reach them, and they
2198/// keep the full [`clip`] echo the diagnosability argument was made for.
2199///
2200/// **Why this list has one entry.** `attempts.outcome` is the column
2201/// `the_token_scanner_can_actually_fail` in `tests/store_journal.rs` proves is a
2202/// carrier, by planting a `ghu_…` in exactly that field.
2203/// `policies.routing_labels` is the other column read through `json_column`, and
2204/// it is off the list on a narrower rule than "free-form".
2205///
2206/// **The rule is: text the *agent* captured from a failure.** Not "text a caller
2207/// chose", which was how this was once written and which does not separate the
2208/// two columns at all — [`crate::model::Label`] admits 256 characters and
2209/// rejects only commas and control characters, so a token *is* a valid label and
2210/// `routing_labels` *is* text a caller chose. What distinguishes them is where
2211/// the text comes from. `FailureReason::Other` is filled from whatever the agent
2212/// found while a start went wrong — a subprocess's stderr, an HTTP body, an
2213/// error a library formatted — none of which the agent inspects before writing
2214/// it down, and any of which can have swept up a token. `routing_labels` is
2215/// typed by an operator into a scale policy as configuration, is read back and
2216/// acted on as configuration, and reaching a credential into it takes a
2217/// deliberate act rather than an accident of capture.
2218///
2219/// That is an argument about the source of the text, and it is the whole of the
2220/// argument. It is deliberately *not* supported by "no test plants a credential
2221/// there": no test plants one in most columns, and a column nobody has attacked
2222/// is not thereby a column that cannot carry a secret. If `routing_labels` ever
2223/// starts being populated from something the agent captured rather than
2224/// something an operator typed, it belongs on this list — which is a list, and
2225/// not a hard-coded pair, so that adding it costs nothing in the decoder.
2226const FREE_FORM_COLUMNS: &[(&str, &str)] = &[("attempts", "outcome")];
2227
2228/// Whether this column may hold text the agent captured from a failure.
2229fn carries_free_form_text(table: &str, column: &str) -> bool {
2230    FREE_FORM_COLUMNS
2231        .iter()
2232        .any(|(t, c)| *t == table && *c == column)
2233}
2234
2235/// One constrained column's payload, trimmed to something safe to put in an
2236/// error message.
2237///
2238/// **Only for a column whose shape the schema fixes.** A column that can hold
2239/// text the agent captured from a failure goes through [`position_only`]
2240/// instead; see [`FREE_FORM_COLUMNS`] for the measurement that separates the
2241/// two.
2242///
2243/// Clipping rather than dropping the value entirely: an operator handed only a
2244/// row id has to go and read the row, and the first sixty characters are
2245/// usually enough to see what went wrong. The byte length is reported so a
2246/// truncated echo cannot be mistaken for the whole value.
2247fn clip(raw: &str) -> String {
2248    match raw.char_indices().nth(ECHO_LIMIT) {
2249        None => raw.to_string(),
2250        Some((cut, _)) => format!(
2251            "{}... ({} bytes in total, truncated)",
2252            &raw[..cut],
2253            raw.len()
2254        ),
2255    }
2256}
2257
2258/// Everything an error may say about a [`FREE_FORM_COLUMNS`] payload: how much
2259/// of it there is, and — when serde recorded one — where it stopped being
2260/// parseable.
2261///
2262/// Neither figure is derived from the *content* of the value, so no part of it
2263/// can travel in the message — which is the point, since a prefix of it is
2264/// exactly what would evade `d1`'s shape-matching sink downstream.
2265///
2266/// It is still enough to work with. A byte count separates "this column is
2267/// empty" from "this column holds a megabyte", and the row id — which is what an
2268/// operator actually needs in order to go and look — is carried by
2269/// [`StoreError::CorruptColumn`] itself and is unaffected.
2270///
2271/// **A position is reported only when serde has one, which on this column is the
2272/// minority of failures.** [`AttemptOutcome`] is an internally tagged enum, so
2273/// serde buffers the object's content and re-reads it from memory to dispatch on
2274/// the tag. Every error raised inside that buffer has lost its place in the
2275/// original text, and `serde_json` reports `line 0, column 0` for it — a
2276/// sentinel meaning "unknown", not a position, since real ones are 1-based.
2277/// Measured over this column:
2278///
2279/// | input | `classify()` | line | column |
2280/// |---|---|---|---|
2281/// | object truncated mid-write | `Eof` | 1 | 54 |
2282/// | unknown `reason` variant | `Data` | **0** | **0** |
2283/// | `reason` of the wrong type | `Data` | **0** | **0** |
2284/// | `reason` missing | `Data` | **0** | **0** |
2285/// | `other` holding a number | `Data` | **0** | **0** |
2286/// | unknown *outcome* tag | `Data` | 1 | 21 |
2287/// | trailing characters | `Syntax` | 1 | 24 |
2288///
2289/// One probe string per row, so the exact column figures are properties of those
2290/// strings and not constants; what the table is about is which rows have a
2291/// position at all, and zero is not one of the answers a 1-based position can
2292/// legitimately take.
2293///
2294/// The four positionless rows are the *likely* ones in practice — schema drift,
2295/// a variant name an older build wrote, a hand-edited journal — and the row that
2296/// does carry a position is the rarer torn write. So this said "it stops parsing
2297/// at line 0, column 0" on the common path, which reads as a real position
2298/// pointing at the payload's first character and is not one. It now says the
2299/// position was not recorded.
2300///
2301/// **The discriminator is `line() == 0`, not `classify()`.** The table is why:
2302/// an unknown *outcome* tag is a `Data` error and still carries a real position,
2303/// because serde reads the tag straight from the input stream before it buffers
2304/// the rest of the object. Branching on `classify()` would throw that position
2305/// away.
2306///
2307/// **None of this is a leak, and that is the part worth being precise about.**
2308/// This function reads `raw.len()`, `error.line()` and `error.column()` and
2309/// never `error.to_string()` — which on exactly the positionless path *does*
2310/// carry the payload, as ``unknown variant `ghs_9tokenish` ``. Both branches are
2311/// payload-free; what differed was only how honest the message was about what it
2312/// knew.
2313fn position_only(raw: &str, error: &serde_json::Error) -> String {
2314    if error.line() == 0 {
2315        format!(
2316            "a {}-byte payload that is not echoed (serde records no position \
2317             for this failure, so where in the payload it went wrong is not \
2318             known)",
2319            raw.len()
2320        )
2321    } else {
2322        format!(
2323            "a {}-byte payload that is not echoed (it stops parsing at line {}, column {})",
2324            raw.len(),
2325            error.line(),
2326            error.column()
2327        )
2328    }
2329}
2330
2331/// How a configured runner root reads in an error message.
2332///
2333/// `None` is not "nothing"; it is the platform default, resolved at runtime by
2334/// `b1`. Rendering it as an empty string or as `None` would make
2335/// [`StoreError::RunnerRootChanged`] read as though a value had gone missing,
2336/// when what it means is that the host is back on its default.
2337fn render_root(value: Option<&str>) -> String {
2338    value.map_or_else(|| "the platform default".to_string(), clip)
2339}
2340
2341/// Name a constraint violation on the attempt journal for what it can only be.
2342///
2343/// `attempts` carries two unique constraints. The primary key is handled by the
2344/// statement's own `ON CONFLICT(id) DO UPDATE`, so it cannot surface here; what
2345/// is left is `one_uncleaned_persistent_attempt_per_slot`, the partial index
2346/// migration 3 adds.
2347///
2348/// **The decision is made from the value being written, not by parsing SQLite's
2349/// message.** An attempt that holds no slot lease cannot violate a partial index
2350/// restricted to uncleaned persistent rows, so a constraint failure on one of
2351/// those is passed through as an ordinary [`StoreError::Sqlite`] rather than
2352/// being mislabelled. Matching on the error string would work today and break
2353/// silently the first time SQLite reworded it.
2354///
2355/// "Which rows the index covers" is asked of [`RunnerAttempt::holds_slot_lease`]
2356/// rather than re-derived from the persisted columns. The same predicate is
2357/// frozen a third time in the migration's partial `WHERE`, which the forward-only
2358/// rule keeps from drifting; these two are the pair that could, so only one of
2359/// them decides.
2360fn slot_lease_error(attempt: &RunnerAttempt, source: rusqlite::Error) -> StoreError {
2361    match (
2362        attempt.workspace().slot_number(),
2363        is_constraint_violation(&source),
2364    ) {
2365        (Some(slot), true) if attempt.holds_slot_lease() => StoreError::SlotAlreadyLeased {
2366            policy: attempt.policy_id,
2367            slot,
2368        },
2369        _ => StoreError::Sqlite(source),
2370    }
2371}
2372
2373fn is_constraint_violation(error: &rusqlite::Error) -> bool {
2374    matches!(
2375        error,
2376        rusqlite::Error::SqliteFailure(inner, _)
2377            if inner.code == rusqlite::ErrorCode::ConstraintViolation
2378    )
2379}
2380
2381fn render(value: ValueRef<'_>) -> String {
2382    match value {
2383        ValueRef::Null => "NULL".to_string(),
2384        ValueRef::Integer(i) => i.to_string(),
2385        ValueRef::Real(f) => f.to_string(),
2386        ValueRef::Text(bytes) => String::from_utf8_lossy(bytes).into_owned(),
2387        ValueRef::Blob(bytes) => bytes.iter().map(|b| format!("{b:02x}")).collect(),
2388    }
2389}
2390
2391// ---------------------------------------------------------------------------
2392// Decoding helpers
2393// ---------------------------------------------------------------------------
2394
2395fn uuid_column(
2396    row: &Row<'_>,
2397    table: &'static str,
2398    column: &'static str,
2399) -> Result<Uuid, StoreError> {
2400    let raw: String = row.get(column)?;
2401    Uuid::parse_str(&raw).map_err(|_| StoreError::CorruptColumn {
2402        table,
2403        column,
2404        // The unparseable id is the only handle on this row there is, so it is
2405        // both the id and the value here. Clipped in both places: a row whose
2406        // primary key is a megabyte of text is exactly the row an error message
2407        // must not repeat.
2408        id: clip(&raw),
2409        value: clip(&raw),
2410        expected: "a hyphenated UUID",
2411    })
2412}
2413
2414fn token_column<T: DeserializeOwned>(
2415    row: &Row<'_>,
2416    table: &'static str,
2417    column: &'static str,
2418    id: &str,
2419) -> Result<T, StoreError> {
2420    let raw: String = row.get(column)?;
2421    serde_json::from_value(serde_json::Value::String(raw.clone())).map_err(|_| {
2422        StoreError::CorruptColumn {
2423            table,
2424            column,
2425            id: id.to_string(),
2426            value: clip(&raw),
2427            expected: "one of this column's recognised tokens",
2428        }
2429    })
2430}
2431
2432/// The one decoder that reads a column which may carry text the agent captured
2433/// from a failure, and therefore the one that has to ask which column it is
2434/// looking at before it says anything about the payload. See
2435/// [`FREE_FORM_COLUMNS`].
2436fn json_column<T: DeserializeOwned>(
2437    row: &Row<'_>,
2438    table: &'static str,
2439    column: &'static str,
2440    id: &str,
2441    expected: &'static str,
2442) -> Result<Option<T>, StoreError> {
2443    match row.get::<_, Option<String>>(column)? {
2444        None => Ok(None),
2445        Some(raw) => {
2446            serde_json::from_str(&raw)
2447                .map(Some)
2448                .map_err(|error| StoreError::CorruptColumn {
2449                    table,
2450                    column,
2451                    id: id.to_string(),
2452                    value: if carries_free_form_text(table, column) {
2453                        position_only(&raw, &error)
2454                    } else {
2455                        clip(&raw)
2456                    },
2457                    expected,
2458                })
2459        }
2460    }
2461}
2462
2463fn timestamp_column(
2464    row: &Row<'_>,
2465    table: &'static str,
2466    column: &'static str,
2467    id: &str,
2468) -> Result<Timestamp, StoreError> {
2469    let raw: String = row.get(column)?;
2470    parse_timestamp(&raw, table, column, id)
2471}
2472
2473fn timestamp_option_column(
2474    row: &Row<'_>,
2475    table: &'static str,
2476    column: &'static str,
2477    id: &str,
2478) -> Result<Option<Timestamp>, StoreError> {
2479    match row.get::<_, Option<String>>(column)? {
2480        None => Ok(None),
2481        Some(raw) => parse_timestamp(&raw, table, column, id).map(Some),
2482    }
2483}
2484
2485fn parse_timestamp(
2486    raw: &str,
2487    table: &'static str,
2488    column: &'static str,
2489    id: &str,
2490) -> Result<Timestamp, StoreError> {
2491    chrono::DateTime::parse_from_rfc3339(raw)
2492        .map(|value| value.with_timezone(&chrono::Utc))
2493        .map_err(|_| StoreError::CorruptColumn {
2494            table,
2495            column,
2496            id: id.to_string(),
2497            value: clip(raw),
2498            expected: "an RFC 3339 timestamp",
2499        })
2500}
2501
2502fn bool_column(
2503    row: &Row<'_>,
2504    table: &'static str,
2505    column: &'static str,
2506    id: &str,
2507) -> Result<bool, StoreError> {
2508    // Read as an integer and require 0 or 1 rather than accepting rusqlite's
2509    // "any non-zero is true": a column that records operator intent should not
2510    // have several spellings of yes, and a hand-edited `7` is a corrupted row
2511    // rather than an enthusiastic one.
2512    match row.get::<_, i64>(column)? {
2513        0 => Ok(false),
2514        1 => Ok(true),
2515        other => Err(StoreError::CorruptColumn {
2516            table,
2517            column,
2518            id: id.to_string(),
2519            value: other.to_string(),
2520            expected: "0 or 1",
2521        }),
2522    }
2523}
2524
2525macro_rules! integer_column {
2526    ($name:ident, $ty:ty, $expected:literal) => {
2527        fn $name(
2528            row: &Row<'_>,
2529            table: &'static str,
2530            column: &'static str,
2531            id: &str,
2532        ) -> Result<$ty, StoreError> {
2533            let raw: i64 = row.get(column)?;
2534            <$ty>::try_from(raw).map_err(|_| StoreError::CorruptColumn {
2535                table,
2536                column,
2537                id: id.to_string(),
2538                value: raw.to_string(),
2539                expected: $expected,
2540            })
2541        }
2542    };
2543}
2544
2545macro_rules! integer_option_column {
2546    ($name:ident, $ty:ty, $expected:literal) => {
2547        fn $name(
2548            row: &Row<'_>,
2549            table: &'static str,
2550            column: &'static str,
2551            id: &str,
2552        ) -> Result<Option<$ty>, StoreError> {
2553            match row.get::<_, Option<i64>>(column)? {
2554                None => Ok(None),
2555                Some(raw) => {
2556                    <$ty>::try_from(raw)
2557                        .map(Some)
2558                        .map_err(|_| StoreError::CorruptColumn {
2559                            table,
2560                            column,
2561                            id: id.to_string(),
2562                            value: raw.to_string(),
2563                            expected: $expected,
2564                        })
2565                }
2566            }
2567        }
2568    };
2569}
2570
2571integer_column!(u16_column, u16, "a value in 0..=65535");
2572integer_column!(u64_column, u64, "a non-negative integer");
2573integer_option_column!(u16_option_column, u16, "a value in 0..=65535");
2574integer_option_column!(u32_option_column, u32, "a value in 0..=4294967295");
2575integer_option_column!(u64_option_column, u64, "a non-negative integer");
2576
2577#[cfg(test)]
2578mod tests {
2579    use super::*;
2580
2581    use std::sync::atomic::{AtomicBool, Ordering};
2582    use std::sync::{Arc, mpsc};
2583    use std::time::{Duration, Instant};
2584
2585    use crate::attempt::FailureReason;
2586    use crate::model::Label;
2587    use crate::policy::PolicyMode;
2588    use crate::workspace::{AttemptWorkspace, WorkspacePolicy};
2589
2590    // `b1`'s fixture ids, spelled as the UUID text a row holds, so a row written
2591    // by hand here and a `testkit` fixture in `tests/` describe the same objects.
2592    const HOST_UUID: &str = "00000000-0000-0000-0000-000000000001";
2593    const POLICY_UUID: &str = "00000000-0000-0000-0000-000000000010";
2594    const ATTEMPT_UUID: &str = "00000000-0000-0000-0000-000000000100";
2595    /// The `runtime_path` every pre-migration attempt fixture is written with,
2596    /// and the one migration 3 must leave exactly as it found it.
2597    const HISTORICAL_PATH: &str = "runtime/policy/pre-upgrade-attempt";
2598    const LABELS_JSON: &str = r#"{"host_label":"rm-home-win-x64","additional":[]}"#;
2599    const COMPLETED_JOB: &str = r#"{"outcome":"completed_job"}"#;
2600    static ATTEMPT_WRITE_BLOCKED: AtomicBool = AtomicBool::new(false);
2601
2602    fn mark_attempt_write_blocked(_: i32) -> bool {
2603        ATTEMPT_WRITE_BLOCKED.store(true, Ordering::Release);
2604        true
2605    }
2606
2607    fn host_id() -> HostId {
2608        HostId::from_u128(0x0000_0001)
2609    }
2610
2611    fn policy_id() -> PolicyId {
2612        PolicyId::from_u128(0x0000_0010)
2613    }
2614
2615    fn attempt_id() -> AttemptId {
2616        AttemptId::from_u128(0x0000_0100)
2617    }
2618
2619    fn ts(secs: i64) -> Timestamp {
2620        chrono::DateTime::from_timestamp(secs, 0).expect("a representable instant")
2621    }
2622
2623    fn store() -> SqliteStore {
2624        SqliteStore::open_in_memory().expect("an in-memory database always opens")
2625    }
2626
2627    // -- raw rows -----------------------------------------------------------
2628    //
2629    // Every corruption test below starts from a row the store itself would have
2630    // written and changes exactly one thing. A test that built its whole row by
2631    // hand would drift from the schema and start passing for the wrong reason.
2632
2633    #[derive(Debug, Clone)]
2634    struct RawHost {
2635        id: String,
2636        display_name: String,
2637        os: String,
2638        architecture: String,
2639        host_capacity: i64,
2640        service_start_mode: String,
2641        refresh_interval_secs: i64,
2642        runner_root_override: Option<String>,
2643        created_at: String,
2644    }
2645
2646    impl Default for RawHost {
2647        fn default() -> Self {
2648            Self {
2649                id: HOST_UUID.to_string(),
2650                display_name: "home-pc".to_string(),
2651                os: "windows".to_string(),
2652                architecture: "x64".to_string(),
2653                host_capacity: 2,
2654                service_start_mode: "boot".to_string(),
2655                refresh_interval_secs: 60,
2656                // D3: a host that has never been configured is on the platform
2657                // default, and the default is not stored.
2658                runner_root_override: None,
2659                created_at: timestamp_to_text(ts(1_000)),
2660            }
2661        }
2662    }
2663
2664    impl RawHost {
2665        fn insert(&self, store: &SqliteStore) {
2666            store
2667                .lock()
2668                .execute(
2669                    "INSERT OR REPLACE INTO hosts (
2670                         id, display_name, os, architecture, host_capacity,
2671                         service_start_mode, refresh_interval_secs,
2672                         runner_root_override, created_at
2673                     ) VALUES (
2674                         :id, :display_name, :os, :architecture, :host_capacity,
2675                         :service_start_mode, :refresh_interval_secs,
2676                         :runner_root_override, :created_at
2677                     )",
2678                    named_params! {
2679                        ":id": self.id,
2680                        ":display_name": self.display_name,
2681                        ":os": self.os,
2682                        ":architecture": self.architecture,
2683                        ":host_capacity": self.host_capacity,
2684                        ":service_start_mode": self.service_start_mode,
2685                        ":refresh_interval_secs": self.refresh_interval_secs,
2686                        ":runner_root_override": self.runner_root_override,
2687                        ":created_at": self.created_at,
2688                    },
2689                )
2690                .expect("the raw host row is writable");
2691        }
2692    }
2693
2694    #[derive(Debug, Clone)]
2695    struct RawPolicy {
2696        id: String,
2697        target_scope: String,
2698        target_slug: String,
2699        installation_id: i64,
2700        host_id: String,
2701        routing_labels: Option<String>,
2702        min_capacity: i64,
2703        max_capacity: Option<i64>,
2704        enabled: i64,
2705        state: String,
2706        cache_policy: String,
2707        workspace_mode: String,
2708        workspace_path: Option<String>,
2709        revision: i64,
2710    }
2711
2712    impl Default for RawPolicy {
2713        fn default() -> Self {
2714            Self {
2715                id: POLICY_UUID.to_string(),
2716                target_scope: "repository".to_string(),
2717                target_slug: "o/r".to_string(),
2718                installation_id: 1,
2719                host_id: HOST_UUID.to_string(),
2720                routing_labels: Some(LABELS_JSON.to_string()),
2721                min_capacity: 0,
2722                max_capacity: Some(2),
2723                enabled: 1,
2724                state: "active".to_string(),
2725                cache_policy: "retain_runner_package".to_string(),
2726                // D3, and what migration 3 gives every historical row.
2727                workspace_mode: "ephemeral".to_string(),
2728                workspace_path: None,
2729                revision: 1,
2730            }
2731        }
2732    }
2733
2734    impl RawPolicy {
2735        fn insert(&self, store: &SqliteStore) {
2736            store
2737                .lock()
2738                .execute(
2739                    "INSERT OR REPLACE INTO policies (
2740                         id, target_scope, target_slug, installation_id, host_id,
2741                         routing_labels, min_capacity, max_capacity, enabled,
2742                         state, cache_policy, workspace_mode, workspace_path, revision
2743                     ) VALUES (
2744                         :id, :target_scope, :target_slug, :installation_id, :host_id,
2745                         :routing_labels, :min_capacity, :max_capacity, :enabled,
2746                         :state, :cache_policy, :workspace_mode, :workspace_path, :revision
2747                     )",
2748                    named_params! {
2749                        ":id": self.id,
2750                        ":target_scope": self.target_scope,
2751                        ":target_slug": self.target_slug,
2752                        ":installation_id": self.installation_id,
2753                        ":host_id": self.host_id,
2754                        ":routing_labels": self.routing_labels,
2755                        ":min_capacity": self.min_capacity,
2756                        ":max_capacity": self.max_capacity,
2757                        ":enabled": self.enabled,
2758                        ":state": self.state,
2759                        ":cache_policy": self.cache_policy,
2760                        ":workspace_mode": self.workspace_mode,
2761                        ":workspace_path": self.workspace_path,
2762                        ":revision": self.revision,
2763                    },
2764                )
2765                .expect("the raw policy row is writable");
2766        }
2767    }
2768
2769    #[derive(Debug, Clone)]
2770    struct RawAttempt {
2771        id: String,
2772        policy_id: String,
2773        github_runner_id: Option<i64>,
2774        state: String,
2775        outcome: Option<String>,
2776        process_id: Option<i64>,
2777        runtime_path: String,
2778        workspace_mode: String,
2779        workspace_slot: Option<i64>,
2780        created_at: String,
2781        terminal_at: Option<String>,
2782        last_state_change_at: String,
2783    }
2784
2785    impl Default for RawAttempt {
2786        fn default() -> Self {
2787            Self {
2788                id: ATTEMPT_UUID.to_string(),
2789                policy_id: POLICY_UUID.to_string(),
2790                github_runner_id: None,
2791                state: "allocated".to_string(),
2792                outcome: None,
2793                process_id: None,
2794                runtime_path: "runtime/policy/attempt".to_string(),
2795                workspace_mode: "ephemeral".to_string(),
2796                workspace_slot: None,
2797                created_at: timestamp_to_text(ts(1_000)),
2798                terminal_at: None,
2799                last_state_change_at: timestamp_to_text(ts(1_000)),
2800            }
2801        }
2802    }
2803
2804    impl RawAttempt {
2805        fn insert(&self, store: &SqliteStore) {
2806            store
2807                .lock()
2808                .execute(
2809                    "INSERT OR REPLACE INTO attempts (
2810                         id, policy_id, github_runner_id, state, outcome, process_id,
2811                         runtime_path, workspace_mode, workspace_slot,
2812                         created_at, terminal_at, last_state_change_at
2813                     ) VALUES (
2814                         :id, :policy_id, :github_runner_id, :state, :outcome, :process_id,
2815                         :runtime_path, :workspace_mode, :workspace_slot,
2816                         :created_at, :terminal_at, :last_state_change_at
2817                     )",
2818                    named_params! {
2819                        ":id": self.id,
2820                        ":policy_id": self.policy_id,
2821                        ":github_runner_id": self.github_runner_id,
2822                        ":state": self.state,
2823                        ":outcome": self.outcome,
2824                        ":process_id": self.process_id,
2825                        ":runtime_path": self.runtime_path,
2826                        ":workspace_mode": self.workspace_mode,
2827                        ":workspace_slot": self.workspace_slot,
2828                        ":created_at": self.created_at,
2829                        ":terminal_at": self.terminal_at,
2830                        ":last_state_change_at": self.last_state_change_at,
2831                    },
2832                )
2833                .expect("the raw attempt row is writable");
2834        }
2835    }
2836
2837    /// A native absolute path with this leaf, for the columns that hold one.
2838    ///
2839    /// Built per platform rather than written as a literal, because
2840    /// `LocalAbsolutePath::new` judges against `PathPlatform::NATIVE`: a Unix
2841    /// literal is corrupt state on Windows and vice versa, which is the rule
2842    /// under test elsewhere and would be an accident here. The drive letter is
2843    /// deliberately not `C:` — nothing in these tests may assume the system
2844    /// drive exists or is named (`02-target-architecture.md`, "Windows root
2845    /// discovery ... never assumes `C:` in tests or code").
2846    fn a_root(leaf: &str) -> LocalAbsolutePath {
2847        let raw = if cfg!(windows) {
2848            format!("X:\\{leaf}")
2849        } else {
2850            format!("/{leaf}")
2851        };
2852        LocalAbsolutePath::new(raw).expect("a fixture root is a storable local path")
2853    }
2854
2855    // -- the on-disk format -------------------------------------------------
2856
2857    #[test]
2858    fn the_on_disk_tokens_are_pinned() {
2859        // A column token comes from the domain's own serde naming, which means a
2860        // rename in `b1` would silently change the on-disk format of every
2861        // existing database. These assertions turn that into a failing test here
2862        // instead. The three enums without an `ALL` constant are covered by an
2863        // exhaustive `match`, so a new variant is a compile error rather than an
2864        // unpinned token.
2865        for (state, expected) in [
2866            (AttemptState::Allocated, "allocated"),
2867            (AttemptState::JitReceived, "jit_received"),
2868            (AttemptState::Starting, "starting"),
2869            (AttemptState::Idle, "idle"),
2870            (AttemptState::Busy, "busy"),
2871            (AttemptState::Finished, "finished"),
2872            (AttemptState::Failed, "failed"),
2873            (AttemptState::Orphaned, "orphaned"),
2874            (AttemptState::Cleaned, "cleaned"),
2875        ] {
2876            assert_eq!(token(&state), expected);
2877        }
2878        assert_eq!(
2879            AttemptState::ALL.len(),
2880            9,
2881            "a new AttemptState needs a pinned token above"
2882        );
2883
2884        for (state, expected) in [
2885            (PolicyState::Pending, "pending"),
2886            (PolicyState::Active, "active"),
2887            (PolicyState::Draining, "draining"),
2888            (PolicyState::Disabled, "disabled"),
2889            (PolicyState::RepairRequired, "repair_required"),
2890            (PolicyState::AuthenticationFailed, "authentication_failed"),
2891        ] {
2892            assert_eq!(token(&state), expected);
2893        }
2894        assert_eq!(
2895            PolicyState::ALL.len(),
2896            6,
2897            "a new PolicyState needs a pinned token above"
2898        );
2899
2900        for os in Os::ALL {
2901            assert_eq!(
2902                token(&os),
2903                match os {
2904                    Os::Windows => "windows",
2905                    Os::MacOs => "mac_os",
2906                    Os::Linux => "linux",
2907                }
2908            );
2909        }
2910        for arch in Arch::ALL {
2911            assert_eq!(
2912                token(&arch),
2913                match arch {
2914                    Arch::X64 => "x64",
2915                    Arch::Arm64 => "arm64",
2916                    Arch::Arm32 => "arm32",
2917                }
2918            );
2919        }
2920        for mode in [StartMode::Boot, StartMode::Login] {
2921            assert_eq!(
2922                token(&mode),
2923                match mode {
2924                    StartMode::Boot => "boot",
2925                    StartMode::Login => "login",
2926                }
2927            );
2928        }
2929        for cache in [
2930            CachePolicy::RetainRunnerPackage,
2931            CachePolicy::DiscardRunnerPackage,
2932        ] {
2933            assert_eq!(
2934                token(&cache),
2935                match cache {
2936                    CachePolicy::RetainRunnerPackage => "retain_runner_package",
2937                    CachePolicy::DiscardRunnerPackage => "discard_runner_package",
2938                }
2939            );
2940        }
2941        for scope in [TargetScope::Repository, TargetScope::Organization] {
2942            assert_eq!(
2943                token(&scope),
2944                match scope {
2945                    TargetScope::Repository => "repository",
2946                    TargetScope::Organization => "organization",
2947                }
2948            );
2949        }
2950        // `workspace_mode` on two tables, and the literal the partial unique
2951        // index in migration 3 filters on. A rename in the domain would leave
2952        // every existing lease outside the index that enforces it, so the token
2953        // is pinned exactly as the states above are.
2954        for kind in [WorkspaceKind::Ephemeral, WorkspaceKind::Persistent] {
2955            assert_eq!(
2956                token(&kind),
2957                match kind {
2958                    WorkspaceKind::Ephemeral => "ephemeral",
2959                    WorkspaceKind::Persistent => "persistent",
2960                }
2961            );
2962        }
2963
2964        // The stored token is not the runtime `Display` string, and for `Os` the
2965        // two genuinely differ: `Windows` displays as its GitHub label token
2966        // `win` and is stored as `windows`. Pinned so that "just use Display"
2967        // becomes a visibly breaking change rather than a silent format
2968        // migration.
2969        assert_eq!(Os::Windows.to_string(), "win");
2970        assert_eq!(token(&Os::Windows), "windows");
2971    }
2972
2973    #[test]
2974    fn a_timestamp_round_trips_to_the_nanosecond() {
2975        // The system clock produces sub-second precision, so a text format that
2976        // truncated it would make every round-trip assertion on a `Host` or an
2977        // attempt fail for values production actually writes.
2978        let precise = chrono::DateTime::from_timestamp(1_787_270_400, 123_456_789)
2979            .expect("a representable instant");
2980        let text = timestamp_to_text(precise);
2981        assert_eq!(text, "2026-08-21T00:00:00.123456789Z");
2982        assert_eq!(
2983            parse_timestamp(&text, "t", "c", "id").expect("round trips"),
2984            precise
2985        );
2986
2987        // Fixed width, so lexical order is instant order and `ORDER BY
2988        // created_at` means what it says.
2989        assert_eq!(timestamp_to_text(ts(0)).len(), text.len());
2990        assert!(timestamp_to_text(ts(0)) < timestamp_to_text(ts(1)));
2991    }
2992
2993    // -- migrations ---------------------------------------------------------
2994
2995    #[test]
2996    fn the_migration_chain_is_ordered_and_starts_at_one() {
2997        assert!(!MIGRATIONS.is_empty());
2998        assert_eq!(MIGRATIONS[0].version, 1);
2999        for pair in MIGRATIONS.windows(2) {
3000            assert!(
3001                pair[1].version > pair[0].version,
3002                "the chain must be strictly ascending; {} does not follow {}",
3003                pair[1].version,
3004                pair[0].version
3005            );
3006        }
3007        assert_eq!(
3008            MIGRATIONS.last().expect("non-empty").version,
3009            SCHEMA_VERSION,
3010            "SCHEMA_VERSION must be the last step in the chain, or a fresh \
3011             database reports a version it was never migrated to"
3012        );
3013    }
3014
3015    #[test]
3016    fn a_fresh_database_gets_the_whole_chain() {
3017        let store = store();
3018        assert_eq!(store.schema_version(), SCHEMA_VERSION);
3019
3020        let conn = store.lock();
3021        for table in TABLES {
3022            let count: i64 = conn
3023                .query_row(
3024                    "SELECT count(*) FROM sqlite_master WHERE type = 'table' AND name = ?1",
3025                    [table],
3026                    |row| row.get(0),
3027                )
3028                .expect("sqlite_master is readable");
3029            assert_eq!(count, 1, "{table} was not created");
3030        }
3031
3032        let mut stmt = conn
3033            .prepare("SELECT version FROM schema_migrations ORDER BY version")
3034            .expect("prepared");
3035        let applied: Vec<i64> = stmt
3036            .query_map([], |row| row.get(0))
3037            .expect("queried")
3038            .collect::<Result<_, _>>()
3039            .expect("collected");
3040        assert_eq!(
3041            applied,
3042            MIGRATIONS
3043                .iter()
3044                .map(|m| i64::from(m.version))
3045                .collect::<Vec<_>>()
3046        );
3047    }
3048
3049    #[test]
3050    fn a_database_one_version_behind_gets_only_the_missing_step() {
3051        // The production chain has one step (see `MIGRATIONS`), so "one version
3052        // behind" cannot yet be expressed against it. The runner is general, and
3053        // this exercises the two properties that matter about it: an applied step
3054        // is not re-run, and a missing one is.
3055        const CHAIN: &[Migration] = &[
3056            Migration {
3057                version: 1,
3058                name: "first",
3059                sql: "CREATE TABLE step_one (id INTEGER NOT NULL PRIMARY KEY, \
3060                      note TEXT NOT NULL) STRICT;",
3061            },
3062            Migration {
3063                version: 2,
3064                name: "second",
3065                sql: "CREATE TABLE step_two (id INTEGER NOT NULL PRIMARY KEY) STRICT;",
3066            },
3067        ];
3068
3069        let mut conn = Connection::open_in_memory().expect("in-memory");
3070        assert_eq!(
3071            apply_migrations(&mut conn, &CHAIN[..1], &SystemClock).expect("step one applies"),
3072            1
3073        );
3074        conn.execute(
3075            "INSERT INTO step_one (id, note) VALUES (1, 'written between the two steps')",
3076            [],
3077        )
3078        .expect("insertable");
3079
3080        // Forward-only: `CREATE TABLE step_one` would fail outright if step one
3081        // were re-run, so a successful return already proves it was skipped, and
3082        // the surviving row proves nothing was rebuilt underneath it.
3083        assert_eq!(
3084            apply_migrations(&mut conn, CHAIN, &SystemClock).expect("only step two applies"),
3085            2
3086        );
3087        let note: String = conn
3088            .query_row("SELECT note FROM step_one WHERE id = 1", [], |row| {
3089                row.get(0)
3090            })
3091            .expect("the row written before the second step survives it");
3092        assert_eq!(note, "written between the two steps");
3093        let two: i64 = conn
3094            .query_row("SELECT count(*) FROM step_two", [], |row| row.get(0))
3095            .expect("step two created its table");
3096        assert_eq!(two, 0);
3097        assert_eq!(current_version(&conn).expect("readable"), 2);
3098
3099        // Running the whole chain over an up-to-date database changes nothing,
3100        // which is what every ordinary open does.
3101        assert_eq!(
3102            apply_migrations(&mut conn, CHAIN, &SystemClock).expect("idempotent"),
3103            2
3104        );
3105        let applied: i64 = conn
3106            .query_row("SELECT count(*) FROM schema_migrations", [], |row| {
3107                row.get(0)
3108            })
3109            .expect("readable");
3110        assert_eq!(applied, 2, "a step must be recorded exactly once");
3111    }
3112
3113    /// A database stopped at `version`, holding one host, one policy and one
3114    /// attempt written at that schema's shape.
3115    ///
3116    /// The rows are written with plain SQL against the older tables rather than
3117    /// through the store, because the store only knows how to write the current
3118    /// shape — which is the whole thing these tests need not to be true.
3119    fn a_database_at_version(path: &Path, version: u32) {
3120        let mut conn = Connection::open(path).expect("openable");
3121        let applied = apply_migrations(&mut conn, &MIGRATIONS[..version as usize], &SystemClock)
3122            .expect("the older chain applies");
3123        assert_eq!(applied, version);
3124
3125        conn.execute(
3126            "INSERT INTO hosts (
3127                 id, display_name, os, architecture, host_capacity,
3128                 service_start_mode, refresh_interval_secs, created_at
3129             ) VALUES (?1, 'home-pc', 'windows', 'x64', 2, 'boot', 60, ?2)",
3130            rusqlite::params![HOST_UUID, timestamp_to_text(ts(1_000))],
3131        )
3132        .expect("a version-1 host row");
3133
3134        // `requested_host_label` arrived in migration 2, so a version-1 database
3135        // must not name it and a version-2 one may. Both spellings are exercised
3136        // rather than one, because "the column list an older build wrote" is
3137        // exactly what a migration has to cope with.
3138        let policy_sql = if version >= 2 {
3139            "INSERT INTO policies (
3140                 id, target_scope, target_slug, installation_id, host_id,
3141                 requested_host_label, routing_labels, min_capacity, max_capacity,
3142                 enabled, state, cache_policy, revision
3143             ) VALUES (?1, 'repository', 'o/r', 1, ?2, 'host', ?3, 0, 2, 1,
3144                       'active', 'retain_runner_package', 1)"
3145        } else {
3146            "INSERT INTO policies (
3147                 id, target_scope, target_slug, installation_id, host_id,
3148                 routing_labels, min_capacity, max_capacity,
3149                 enabled, state, cache_policy, revision
3150             ) VALUES (?1, 'repository', 'o/r', 1, ?2, ?3, 0, 2, 1,
3151                       'active', 'retain_runner_package', 1)"
3152        };
3153        conn.execute(
3154            policy_sql,
3155            rusqlite::params![POLICY_UUID, HOST_UUID, LABELS_JSON],
3156        )
3157        .expect("a historical policy row");
3158
3159        conn.execute(
3160            "INSERT INTO attempts (
3161                 id, policy_id, state, runtime_path, created_at, last_state_change_at
3162             ) VALUES (?1, ?2, 'idle', ?3, ?4, ?4)",
3163            rusqlite::params![
3164                ATTEMPT_UUID,
3165                POLICY_UUID,
3166                HISTORICAL_PATH,
3167                timestamp_to_text(ts(1_000))
3168            ],
3169        )
3170        .expect("a historical attempt row");
3171
3172        drop(conn);
3173    }
3174
3175    /// The assertions `03-migration-rollout.md` states for every migrated row,
3176    /// whichever version the database started at.
3177    fn assert_everything_migrated_to_ephemeral(store: &SqliteStore) {
3178        assert_eq!(store.schema_version(), SCHEMA_VERSION);
3179
3180        let host = store.host(host_id()).expect("loads").expect("present");
3181        assert_eq!(
3182            host.runner_root_override, None,
3183            "a migrated host is on the platform default; storing the effective \
3184             path instead would freeze today's default into every database"
3185        );
3186        assert!(!host.has_configured_runner_root());
3187        // The columns migration 3 did not touch are still what they were, which
3188        // is what makes this an upgrade rather than a rewrite.
3189        assert_eq!(host.host_capacity.get(), 2);
3190        assert_eq!(host.service_start_mode, StartMode::Boot);
3191
3192        let policy = store.policy(policy_id()).expect("loads").expect("present");
3193        assert_eq!(
3194            policy.workspace_policy(),
3195            &WorkspacePolicy::Ephemeral,
3196            "an upgrade must not retain a workspace the operator never selected"
3197        );
3198        assert_eq!(policy.requested_host_label.as_str(), "host");
3199
3200        let attempt = store
3201            .attempt(attempt_id())
3202            .expect("loads")
3203            .expect("present");
3204        assert_eq!(attempt.workspace(), AttemptWorkspace::Ephemeral);
3205        assert!(!attempt.holds_slot_lease());
3206        assert_eq!(
3207            attempt.runtime_path(),
3208            Path::new(HISTORICAL_PATH),
3209            "`No journal row is rewritten merely to adopt the new default`: \
3210             recovery removes the exact directory the attempt was created in, \
3211             so a rewritten path would point cleanup at one it never used"
3212        );
3213        assert_eq!(attempt.state(), AttemptState::Idle);
3214    }
3215
3216    #[test]
3217    fn a_version_one_database_migrates_through_the_whole_chain() {
3218        // The oldest shape that exists: no `requested_host_label`, no workspace
3219        // columns. Both later steps have to run, in order, on one open.
3220        let dir = tempfile::tempdir().expect("a temporary directory");
3221        let path = dir.path().join("runner-manager.sqlite3");
3222        a_database_at_version(&path, 1);
3223
3224        let store = SqliteStore::open(&path).expect("a version-1 database migrates");
3225        assert_everything_migrated_to_ephemeral(&store);
3226
3227        let conn = store.lock();
3228        let mut stmt = conn
3229            .prepare("SELECT version FROM schema_migrations ORDER BY version")
3230            .expect("prepared");
3231        let applied: Vec<i64> = stmt
3232            .query_map([], |row| row.get(0))
3233            .expect("queried")
3234            .collect::<Result<_, _>>()
3235            .expect("collected");
3236        assert_eq!(applied, vec![1, 2, 3], "the full chain, in order, once");
3237    }
3238
3239    #[test]
3240    fn a_version_two_database_migrates_every_row_to_ephemeral() {
3241        // The shape a shipped build actually leaves behind, and the one
3242        // `03-migration-rollout.md`'s Phase 0 gate is written against: "Prove
3243        // version-2 databases migrate every policy and attempt to ephemeral."
3244        let dir = tempfile::tempdir().expect("a temporary directory");
3245        let path = dir.path().join("runner-manager.sqlite3");
3246        a_database_at_version(&path, 2);
3247
3248        let store = SqliteStore::open(&path).expect("a version-2 database migrates");
3249        assert_everything_migrated_to_ephemeral(&store);
3250
3251        // Migrating twice is what every subsequent start does.
3252        drop(store);
3253        let reopened = SqliteStore::open(&path).expect("reopens");
3254        assert_everything_migrated_to_ephemeral(&reopened);
3255    }
3256
3257    #[test]
3258    fn a_database_from_a_newer_build_is_refused_rather_than_guessed_at() {
3259        let dir = tempfile::tempdir().expect("a temporary directory");
3260        let path = dir.path().join("runner-manager.sqlite3");
3261
3262        let store = SqliteStore::open(&path).expect("a fresh database opens");
3263        assert_eq!(store.schema_version(), SCHEMA_VERSION);
3264        drop(store);
3265
3266        // A future build migrated it further than this build understands.
3267        let future = SCHEMA_VERSION + 1;
3268        {
3269            let conn = Connection::open(&path).expect("reopenable");
3270            conn.execute(
3271                "INSERT INTO schema_migrations (version, name, applied_at) \
3272                 VALUES (?1, 'from_the_future', ?2)",
3273                rusqlite::params![i64::from(future), timestamp_to_text(ts(2_000))],
3274            )
3275            .expect("insertable");
3276        }
3277
3278        let error = SqliteStore::open(&path).expect_err("a newer database must be refused");
3279        assert!(
3280            matches!(
3281                error,
3282                StoreError::SchemaTooNew { found, supported }
3283                    if found == future && supported == SCHEMA_VERSION
3284            ),
3285            "expected SchemaTooNew, got {error:?}"
3286        );
3287        let message = error.to_string();
3288        assert!(
3289            message.contains(&future.to_string()) && message.contains(&SCHEMA_VERSION.to_string()),
3290            "the error must name both versions so an operator knows which way to \
3291             move: {message}"
3292        );
3293        assert!(
3294            !error.is_conflict(),
3295            "a schema refusal is not an optimistic-concurrency conflict"
3296        );
3297    }
3298
3299    #[test]
3300    fn a_corrupt_schema_version_is_named_rather_than_reported_as_four_billion() {
3301        // `current_version` used to coerce a negative recorded version with
3302        // `unwrap_or(u32::MAX)`. That failed closed, which is right, but it did
3303        // so by telling the operator "this database is at schema version
3304        // 4294967295" -- a number no database has been at, and one that reads as
3305        // a bug in this code rather than as a corrupt bookkeeping row.
3306        //
3307        // The bookkeeping table alone, holding one hand-edited row. `MAX` is
3308        // what `current_version` reads, so a corrupt row only decides the answer
3309        // when it is the highest one -- which for a negative value means it is
3310        // the only one, and that is exactly the state an aborted or edited first
3311        // migration leaves behind.
3312        let mut conn = Connection::open_in_memory().expect("in-memory");
3313        conn.execute_batch(BOOTSTRAP_SQL).expect("bootstrapped");
3314        conn.execute(
3315            "INSERT INTO schema_migrations (version, name, applied_at) \
3316             VALUES (-1, 'hand_edited', ?1)",
3317            rusqlite::params![timestamp_to_text(ts(2_000))],
3318        )
3319        .expect("insertable");
3320
3321        let error = current_version(&conn).expect_err("a negative version is not a version");
3322        assert!(
3323            matches!(
3324                &error,
3325                StoreError::CorruptColumn {
3326                    table: "schema_migrations",
3327                    column: "version",
3328                    ..
3329                }
3330            ),
3331            "expected a named corrupt column, got {error:?}"
3332        );
3333        let message = error.to_string();
3334        assert!(
3335            message.contains("-1") && !message.contains(&u32::MAX.to_string()),
3336            "the message must name the row's actual value: {message}"
3337        );
3338
3339        // And it still fails closed: nothing re-applies a migration over this.
3340        assert!(apply_migrations(&mut conn, MIGRATIONS, &SystemClock).is_err());
3341    }
3342
3343    #[test]
3344    fn a_negative_stored_revision_is_a_corrupt_column_and_not_revision_zero() {
3345        // `conflict_or_missing` reads `revision` raw, bypassing the check every
3346        // other column gets, and used to coerce with `unwrap_or(0)`. Against a
3347        // hand-edited `-1` that produced "written against revision 0, but the
3348        // stored revision is now 0" -- a sentence that contradicts itself and
3349        // sends the operator looking in the wrong place.
3350        let store = store();
3351        RawPolicy {
3352            revision: -1,
3353            ..RawPolicy::default()
3354        }
3355        .insert(&store);
3356
3357        let policy = ScalePolicy::new(
3358            policy_id(),
3359            ScaleTarget::repository("o/r").expect("valid"),
3360            1,
3361            host_id(),
3362            PolicyMode::autoscale(
3363                RoutingLabels::from_host_label(Label::new("rm-home-win-x64").expect("valid")),
3364                0,
3365                NonZeroU16::new(2).expect("non-zero"),
3366            )
3367            .expect("valid"),
3368            CachePolicy::default(),
3369        );
3370        let error = store
3371            .update_policy(&policy, 0)
3372            .expect_err("the row's revision is not 0, so this matches nothing");
3373        assert!(
3374            matches!(
3375                &error,
3376                StoreError::CorruptColumn {
3377                    table: "policies",
3378                    column: "revision",
3379                    ..
3380                }
3381            ),
3382            "expected a named corrupt column, got {error:?}"
3383        );
3384        assert!(
3385            !error.is_conflict(),
3386            "a corrupt row is not a lost race, and a caller told to re-read and \
3387             retry would loop for ever on it"
3388        );
3389        assert!(
3390            error.to_string().contains("-1"),
3391            "the message must name the value that needs fixing: {error}"
3392        );
3393    }
3394
3395    // -- what an error message repeats --------------------------------------
3396
3397    #[test]
3398    fn a_corrupt_column_error_clips_the_payload_it_echoes() {
3399        assert_eq!(clip("short"), "short");
3400
3401        let exact = "a".repeat(ECHO_LIMIT);
3402        assert_eq!(clip(&exact), exact, "the limit is an edge, not a target");
3403
3404        let over = "a".repeat(ECHO_LIMIT + 1);
3405        let clipped = clip(&over);
3406        assert!(clipped.starts_with(&exact));
3407        assert!(
3408            clipped.contains(&format!("{} bytes in total", over.len())),
3409            "a clipped echo must say it is one: {clipped}"
3410        );
3411
3412        // Multi-byte characters are cut on a character boundary, or this panics.
3413        let wide = "é".repeat(ECHO_LIMIT * 2);
3414        assert!(clip(&wide).starts_with(&"é".repeat(ECHO_LIMIT)));
3415    }
3416
3417    #[test]
3418    fn a_secret_in_a_free_form_column_is_not_echoed_whole_into_the_error() {
3419        // "No column carries a credential" is a claim about the schema, and
3420        // `attempts.outcome` is where it stops being true: it holds the JSON of
3421        // an `AttemptOutcome`, and `FailureReason::Other(String)` inside that is
3422        // free-form text the agent captured while a start went wrong.
3423        // `the_token_scanner_can_actually_fail` in `tests/store_journal.rs`
3424        // plants a `ghu_...` there on purpose, to prove the field is a real
3425        // carrier.
3426        //
3427        // A malformed value in that column produces a `CorruptColumn`, whose
3428        // message goes wherever the error goes. This test was written when the
3429        // answer was a sixty-character clip, and its rationale was that `d1`'s
3430        // shape-matching sink would probably catch a prefixed token downstream
3431        // even if one leaked. It no longer rests on that, and should not: the
3432        // sink is a control that a *truncated* token can walk straight past,
3433        // which is the argument in `FREE_FORM_COLUMNS`. Nothing from this column
3434        // is echoed at all now, so the long payload below is a case of the rule
3435        // rather than the reason for it, and
3436        // `a_short_secret_in_the_free_form_column_is_not_echoed_at_all` covers
3437        // the length a clip could never have protected.
3438        let store = store();
3439        let blob = "QUJDREVGR0hJSktMTU5PUFFSU1RVVldYWVphYmNkZWZnaGlqa2xtbm9wcXJzdHV2d3h5ejAxMjM0\
3440                    NTY3ODkrLwABAgMEBQYHCAkKCwwNDg8QERITFBUWFxgZGhscHR4fICEiIyQlJicoKSorLC0uLzA"
3441            .to_string();
3442        assert!(blob.len() > ECHO_LIMIT * 2);
3443
3444        RawAttempt {
3445            state: "finished".to_string(),
3446            outcome: Some(format!(r#"{{"outcome":"went_home","detail":"{blob}"}}"#)),
3447            terminal_at: Some(timestamp_to_text(ts(2_000))),
3448            ..RawAttempt::default()
3449        }
3450        .insert(&store);
3451
3452        let error = store
3453            .attempt(attempt_id())
3454            .expect_err("`went_home` is not an outcome");
3455        let StoreError::CorruptColumn { value, .. } = &error else {
3456            panic!("expected a corrupt column, got {error:?}");
3457        };
3458        assert!(
3459            !value.contains(&blob),
3460            "the whole payload must not be repeated: {value}"
3461        );
3462        assert!(
3463            value.chars().count() < blob.chars().count(),
3464            "the echo must be shorter than what it echoes"
3465        );
3466        // And not a prefix of it either. A base64 blob has no prefix for `d1`'s
3467        // sink to match on, so a leading fragment of one is exactly as
3468        // unredactable as the whole thing is.
3469        assert!(
3470            !value.contains(&blob[..8]),
3471            "no leading fragment of the payload may travel either: {value}"
3472        );
3473
3474        // The row id is what an operator actually needs, and it is still there.
3475        assert!(
3476            error.to_string().contains(&attempt_id().to_string()),
3477            "the error must name the row to fix: {error}"
3478        );
3479    }
3480
3481    #[test]
3482    fn a_short_secret_in_the_free_form_column_is_not_echoed_at_all() {
3483        // The case a fixed character budget cannot reach. `ECHO_LIMIT` is sixty
3484        // characters and does not know which column it is echoing; the serde
3485        // prefix that reaches `FailureReason::Other` is thirty-nine of them, so
3486        // a budget-based echo repeats about twenty-one characters of whatever
3487        // the caller put there. That is partial for a forty-character `ghu_`
3488        // token and complete for anything shorter -- and a *partial* token is
3489        // the worse of the two, because `d1`'s sink matches on shape and a
3490        // fragment may no longer have the shape it redacts.
3491        let planted = "ghs_9tokenish";
3492        let raw = format!(r#"{{"outcome":"failed","reason":{{"other":"{planted}"}}"#);
3493
3494        // The counterfactual, asserted rather than described: this whole row
3495        // fits inside the budget, so the echo it authorises is not a clip of
3496        // the value but the entire value, secret included.
3497        assert!(raw.chars().count() < ECHO_LIMIT);
3498        assert_eq!(
3499            clip(&raw),
3500            raw,
3501            "a sixty-character budget repeats this row in full, which is what \
3502             makes the budget the wrong instrument at this column"
3503        );
3504        assert!(carries_free_form_text("attempts", "outcome"));
3505
3506        // Malformed only in its final brace, so the failure is a real parse
3507        // failure and the secret sits where a caller would really have put it.
3508        let store = store();
3509        RawAttempt {
3510            state: "failed".to_string(),
3511            outcome: Some(raw.clone()),
3512            terminal_at: Some(timestamp_to_text(ts(2_000))),
3513            ..RawAttempt::default()
3514        }
3515        .insert(&store);
3516
3517        let error = store
3518            .attempt(attempt_id())
3519            .expect_err("an unterminated object is not an attempt outcome");
3520        let rendered = error.to_string();
3521
3522        assert!(
3523            !rendered.contains(planted),
3524            "the secret must not appear in the message: {rendered}"
3525        );
3526        assert!(
3527            !rendered.contains("ghs_"),
3528            "and neither must a prefixed fragment of it, which is what a \
3529             clipped echo would have produced and what `d1`'s shape-matching \
3530             sink would then have failed to redact: {rendered}"
3531        );
3532        for len in 4..=planted.len() {
3533            assert!(
3534                !rendered.contains(&planted[..len]),
3535                "no prefix of the secret may survive, and {:?} did: {rendered}",
3536                &planted[..len]
3537            );
3538        }
3539
3540        // What is left is still diagnosable: the size of the payload, where it
3541        // gave up, and above all the row to go and look at.
3542        assert!(
3543            rendered.contains(&format!("{}-byte", raw.len())),
3544            "the message must say how much is there: {rendered}"
3545        );
3546        assert!(
3547            rendered.contains("stops parsing at line"),
3548            "and where it stopped: {rendered}"
3549        );
3550        assert!(
3551            rendered.contains(&attempt_id().to_string()),
3552            "the error must name the row to fix: {rendered}"
3553        );
3554
3555        // -- the other path, which is the likelier one ---------------------
3556        //
3557        // Everything above is the *torn write*: the object stops mid-text, so
3558        // serde fails at a place in the original input and has a position to
3559        // report. The common corruption in practice has no position at all.
3560        // `AttemptOutcome` is internally tagged, so serde buffers the content
3561        // and re-reads it to dispatch on the tag, and anything that goes wrong
3562        // inside that buffer has lost its place in the text: `line() == 0`,
3563        // `column() == 0`. Schema drift, a variant name an older build wrote, a
3564        // hand-edited journal all land there.
3565        //
3566        // This half was untested, and while it was, `position_only` printed
3567        // "line 0, column 0" for it -- which reads as a position pointing at the
3568        // payload's first character and is not one; it is serde's sentinel for
3569        // "unknown".
3570        let drifted = "ghs_9tokenish";
3571        let raw = format!(r#"{{"outcome":"failed","reason":"{drifted}"}}"#);
3572
3573        // The row parses as JSON and is well-formed; only the *variant* is
3574        // unknown, which is what puts the failure inside the buffer.
3575        serde_json::from_str::<serde_json::Value>(&raw).expect("this row is valid JSON");
3576        let inner = serde_json::from_str::<AttemptOutcome>(&raw)
3577            .expect_err("`ghs_9tokenish` is not a FailureReason");
3578        assert_eq!(
3579            (inner.line(), inner.column()),
3580            (0, 0),
3581            "the premise of this half of the test: serde has no position here"
3582        );
3583
3584        // And the counterfactual that makes the payload-free rule load-bearing
3585        // rather than decorative: serde's own message *does* carry the value, so
3586        // a `position_only` written in terms of `error.to_string()` would have
3587        // published the secret on precisely this path.
3588        assert!(
3589            inner.to_string().contains(drifted),
3590            "serde names the offending variant, so the message is not safe to \
3591             forward: {inner}"
3592        );
3593
3594        // `self::store` rather than `store`: the binding above shadows the
3595        // fixture function for the rest of this body, and this half needs a
3596        // database that does not already hold the row it is about to write.
3597        let drift_store = self::store();
3598        RawAttempt {
3599            state: "failed".to_string(),
3600            outcome: Some(raw.clone()),
3601            terminal_at: Some(timestamp_to_text(ts(2_000))),
3602            ..RawAttempt::default()
3603        }
3604        .insert(&drift_store);
3605
3606        let rendered = drift_store
3607            .attempt(attempt_id())
3608            .expect_err("an unknown reason variant is not an attempt outcome")
3609            .to_string();
3610
3611        assert!(
3612            !rendered.contains(drifted),
3613            "the secret must not appear here either: {rendered}"
3614        );
3615        for len in 4..=drifted.len() {
3616            assert!(
3617                !rendered.contains(&drifted[..len]),
3618                "no prefix of the secret may survive, and {:?} did: {rendered}",
3619                &drifted[..len]
3620            );
3621        }
3622
3623        // The byte count still works, because it is computed from the column
3624        // rather than taken from the error.
3625        assert!(
3626            rendered.contains(&format!("{}-byte", raw.len())),
3627            "the message must say how much is there: {rendered}"
3628        );
3629        assert!(
3630            rendered.contains(&attempt_id().to_string()),
3631            "the error must name the row to fix: {rendered}"
3632        );
3633
3634        // The substance of this half: no fabricated position. "line 0, column 0"
3635        // is not a location an operator can act on, and printing it as though it
3636        // were sends them looking at the start of a payload that is very likely
3637        // fine.
3638        assert!(
3639            !rendered.contains("line 0")
3640                && !rendered.contains("column 0")
3641                && !rendered.contains("stops parsing at"),
3642            "a position serde did not record must not be printed as one: \
3643             {rendered}"
3644        );
3645        assert!(
3646            rendered.contains("no position"),
3647            "and the message has to say so, or the absence is indistinguishable \
3648             from an omission: {rendered}"
3649        );
3650    }
3651
3652    #[test]
3653    fn an_unknown_outcome_tag_keeps_the_position_serde_did_record() {
3654        // The one row of `position_only`'s table that separates its
3655        // discriminator from the obvious alternative -- and, until this test,
3656        // the only row nothing planted.
3657        //
3658        // `position_only` branches on `error.line() == 0`, and its documentation
3659        // says so in bold, because `classify()` is the reading a later editor
3660        // reaches for: five of the seven measured rows are `Data` *and*
3661        // positionless, so "positionless means `Data`" fits almost every row.
3662        // It is wrong at exactly one -- an unknown *outcome* tag. `AttemptOutcome`
3663        // is internally tagged, so serde reads the tag straight out of the input
3664        // stream to decide what to deserialise into and only *then* buffers the
3665        // rest of the object; a tag it does not recognise fails before the
3666        // buffering that loses the position. That error is `Data` and carries a
3667        // real 1-based position, so a `classify()`-based branch would throw the
3668        // position away and tell an operator none was recorded when one was.
3669        //
3670        // The suite did not notice. Swapping the discriminator to
3671        // `error.classify() == serde_json::error::Category::Data` left every
3672        // test green -- 129 lib, 9 integration -- because the positionless half
3673        // of `a_short_secret_in_the_free_form_column_is_not_echoed_at_all`
3674        // plants an unknown *reason* variant, which is positionless under both
3675        // readings and so cannot tell them apart. By this crate's own standard a
3676        // measurement recorded once and never re-run is a gap; this closes the
3677        // one under that table.
3678        let raw = r#"{"outcome":"vanished"}"#;
3679
3680        // The premise, asserted rather than described. Both halves carry weight:
3681        // the first is what makes `classify()` look correct, the second is what
3682        // makes it wrong.
3683        let inner = serde_json::from_str::<AttemptOutcome>(raw)
3684            .expect_err("`vanished` is not an attempt outcome");
3685        assert_eq!(
3686            inner.classify(),
3687            serde_json::error::Category::Data,
3688            "a `classify()`-based discriminator would send this down the \
3689             positionless branch: {inner}"
3690        );
3691        assert_ne!(
3692            inner.line(),
3693            0,
3694            "and it has a real position to lose, which is the whole finding: \
3695             {inner}"
3696        );
3697
3698        let store = store();
3699        RawAttempt {
3700            state: "failed".to_string(),
3701            outcome: Some(raw.to_string()),
3702            terminal_at: Some(timestamp_to_text(ts(2_000))),
3703            ..RawAttempt::default()
3704        }
3705        .insert(&store);
3706
3707        let rendered = store
3708            .attempt(attempt_id())
3709            .expect_err("an unknown outcome tag is not an attempt outcome")
3710            .to_string();
3711
3712        // Read off the error rather than written as a literal: the column figure
3713        // is a property of this probe string, not a constant of the format.
3714        assert!(
3715            rendered.contains(&format!(
3716                "stops parsing at line {}, column {}",
3717                inner.line(),
3718                inner.column()
3719            )),
3720            "the position serde did record must survive into the message: \
3721             {rendered}"
3722        );
3723        assert!(
3724            !rendered.contains("no position"),
3725            "and must not be reported as absent: {rendered}"
3726        );
3727
3728        // Same branch, same payload-free rule: the position travels and the tag
3729        // that produced it does not. `position_only` reads `line()` and
3730        // `column()`, never `to_string()`, on this path as much as on the other.
3731        assert!(
3732            rendered.contains(&format!("{}-byte", raw.len())),
3733            "the message must still say how much is there: {rendered}"
3734        );
3735        assert!(
3736            !rendered.contains("vanished"),
3737            "serde names the offending tag; this message must not: {rendered}"
3738        );
3739    }
3740
3741    #[test]
3742    fn a_constrained_column_still_echoes_what_it_holds() {
3743        // The other half of the per-column rule: the echo is removed at the one
3744        // column that carries text the agent captured from a failure, and
3745        // nowhere else. A malformed timestamp has the shape the schema fixes --
3746        // nothing free-form can reach it -- so an operator still gets to see the
3747        // value that needs fixing, which is the diagnosability the clip was
3748        // argued for.
3749        assert!(!carries_free_form_text("attempts", "created_at"));
3750        assert!(!carries_free_form_text("policies", "routing_labels"));
3751
3752        // `routing_labels` is off the list on the *source* of its text, not on
3753        // its shape, and this is the assertion that keeps that honest: a token
3754        // is a perfectly legal `Label`, so "the schema constrains it" is not
3755        // available as the reason. What is available is that an operator types
3756        // a routing label into a scale policy as configuration, whereas
3757        // `FailureReason::Other` is filled from whatever the agent scraped off a
3758        // failure without reading it. If that ever stops being true of
3759        // `routing_labels`, this assertion still passes and the column still
3760        // belongs on `FREE_FORM_COLUMNS`.
3761        assert!(
3762            Label::new("ghu_16CharsOfPaddingAndThenSomeMore1234567").is_ok(),
3763            "a credential-shaped string is a valid Label, so the length and \
3764             character rules are not what keeps `routing_labels` off the list"
3765        );
3766
3767        let store = store();
3768        RawAttempt {
3769            created_at: "the third of never".to_string(),
3770            ..RawAttempt::default()
3771        }
3772        .insert(&store);
3773
3774        let error = store
3775            .attempt(attempt_id())
3776            .expect_err("`the third of never` is not RFC 3339");
3777        assert!(
3778            error.to_string().contains("the third of never"),
3779            "a constrained column keeps its echo: {error}"
3780        );
3781    }
3782
3783    #[test]
3784    fn the_journal_mode_is_read_back_rather_than_assumed() {
3785        // The pragma answers with the mode the database *ended up in*. Where WAL
3786        // is unavailable -- no shared-memory support, as on some network mounts
3787        // -- SQLite silently leaves the database in `delete` and says so in that
3788        // row, which this store used to discard.
3789        let memory = store();
3790        assert_eq!(
3791            memory.journal_mode(),
3792            "memory",
3793            "an in-memory database is exempt by construction"
3794        );
3795        assert!(
3796            !memory.readers_do_not_block_writers(),
3797            "there is no second reader of a private in-memory database, so the \
3798             question does not arise for it"
3799        );
3800
3801        let dir = tempfile::tempdir().expect("a temporary directory");
3802        let path = dir.path().join("runner-manager.sqlite3");
3803        let file = SqliteStore::open(&path).expect("opens");
3804        let mode = file.journal_mode();
3805
3806        // Not `assert_eq!(mode, "wal")`. That is the assumption this test is
3807        // named for refusing, and the same commit that named it removed exactly
3808        // this assertion from `tests/store_journal.rs` on the stated grounds
3809        // that it failed on a healthy build wherever WAL is unavailable. A
3810        // container whose `TMPDIR` is a tmpfs or a network mount gives
3811        // `tempfile::tempdir()` a directory that cannot host WAL's
3812        // shared-memory file, and SQLite then leaves the database in `delete`
3813        // and says so -- a correct answer, and a false red here.
3814        //
3815        // Both are legal; a third value would be a real finding, so the set is
3816        // closed rather than dropped.
3817        assert!(
3818            matches!(mode, "wal" | "delete"),
3819            "a file database is in WAL where the directory can host it and \
3820             `delete` where it cannot; {mode} is neither and is a finding"
3821        );
3822
3823        // The agreement check is the substance, and it reads a second source to
3824        // make it: the presence of the write-ahead log *on disk*. Comparing
3825        // `readers_do_not_block_writers()` against `mode == "wal"` would not,
3826        // because the method is defined as that comparison -- it could only
3827        // ever have failed on a difference in letter case, which is not what a
3828        // message about readers and writers claims to be checking.
3829        //
3830        // `file` is still open here, which is what makes the file check sound:
3831        // SQLite creates the `-wal` beside the database on the first write (the
3832        // migrations above are one) and removes it only on a clean close of the
3833        // last connection.
3834        assert_eq!(
3835            file.readers_do_not_block_writers(),
3836            path.with_extension("sqlite3-wal").exists(),
3837            "in {mode} mode the claim about readers and the write-ahead log on \
3838             disk must be the same fact told twice"
3839        );
3840        assert!(
3841            format!("{file:?}").contains(mode),
3842            "an operator reading a support bundle should see the mode"
3843        );
3844    }
3845
3846    // -- the column/field mapping ------------------------------------------
3847
3848    #[test]
3849    fn every_column_lands_in_the_field_of_the_same_name() {
3850        // `PersistedAttempt` and `PersistedPolicy` make the *field* names
3851        // compile-checked and say plainly that the *column* names are not. This
3852        // is the test they ask for: every column holds a value that appears
3853        // nowhere else in its row, so a transposition cannot survive it.
3854        //
3855        // The pairs this is really about are `created_at`/`last_state_change_at`
3856        // (both `Timestamp`) and `installation_id`/`revision` (both `u64`): each
3857        // transposes without a compile error, and each was a real defect in an
3858        // earlier positional signature.
3859        //
3860        // **This covers the read direction only**, and the name does not say so.
3861        // Every row here is written by hand through a `Raw*::insert`, so a
3862        // transposition in `policy_params` or `attempt_params` -- the write half
3863        // of the same crossing -- is invisible to it. The write direction is
3864        // pinned transitively instead, by the round-trip tests in
3865        // `tests/store_journal.rs`: a read proven correct here plus a domain
3866        // value that survives a store-and-load unchanged leaves no room for the
3867        // write to be transposed. That inference holds, but it is an inference,
3868        // and a reader of this test should know which half they are looking at.
3869        let store = store();
3870
3871        let configured_root = a_root("distinguishable-root");
3872        RawHost {
3873            host_capacity: 7,
3874            refresh_interval_secs: 45,
3875            created_at: timestamp_to_text(ts(1_234)),
3876            display_name: "distinguishable-name".to_string(),
3877            runner_root_override: Some(configured_root.as_str().to_string()),
3878            ..RawHost::default()
3879        }
3880        .insert(&store);
3881
3882        let host = store.host(host_id()).expect("loads").expect("present");
3883        assert_eq!(host.id, host_id(), "hosts.id");
3884        assert_eq!(host.display_name, "distinguishable-name");
3885        assert_eq!(host.host_capacity.get(), 7, "hosts.host_capacity");
3886        assert_eq!(
3887            host.refresh_interval.as_secs(),
3888            45,
3889            "hosts.refresh_interval_secs"
3890        );
3891        assert_eq!(host.created_at, ts(1_234), "hosts.created_at");
3892        assert_eq!(host.os, Os::Windows, "hosts.os");
3893        assert_eq!(host.architecture, Arch::X64, "hosts.architecture");
3894        assert_eq!(
3895            host.service_start_mode,
3896            StartMode::Boot,
3897            "hosts.service_start_mode"
3898        );
3899        assert_eq!(
3900            host.runner_root_override,
3901            Some(configured_root),
3902            "hosts.runner_root_override"
3903        );
3904
3905        let persistent_root = a_root("distinguishable-workspace");
3906        RawPolicy {
3907            installation_id: 111,
3908            revision: 222,
3909            min_capacity: 3,
3910            max_capacity: Some(9),
3911            enabled: 0,
3912            state: "pending".to_string(),
3913            cache_policy: "discard_runner_package".to_string(),
3914            target_slug: "owner/repo".to_string(),
3915            workspace_mode: "persistent".to_string(),
3916            workspace_path: Some(persistent_root.as_str().to_string()),
3917            ..RawPolicy::default()
3918        }
3919        .insert(&store);
3920
3921        let policy = store.policy(policy_id()).expect("loads").expect("present");
3922        assert_eq!(policy.id, policy_id(), "policies.id");
3923        assert_eq!(policy.host_id, host_id(), "policies.host_id");
3924        assert_eq!(
3925            policy.installation_id, 111,
3926            "policies.installation_id must not come from policies.revision"
3927        );
3928        assert_eq!(
3929            policy.revision(),
3930            222,
3931            "policies.revision must not come from policies.installation_id"
3932        );
3933        assert_eq!(policy.min_capacity(), 3, "policies.min_capacity");
3934        assert_eq!(
3935            policy.max_capacity().expect("autoscale").get(),
3936            9,
3937            "policies.max_capacity"
3938        );
3939        assert!(!policy.enabled(), "policies.enabled");
3940        assert_eq!(policy.state(), PolicyState::Pending, "policies.state");
3941        assert_eq!(
3942            policy.cache_policy,
3943            CachePolicy::DiscardRunnerPackage,
3944            "policies.cache_policy"
3945        );
3946        assert_eq!(policy.target.slug(), "owner/repo", "policies.target_slug");
3947        assert_eq!(
3948            policy.target.scope(),
3949            TargetScope::Repository,
3950            "policies.target_scope"
3951        );
3952        assert_eq!(
3953            policy
3954                .routing_labels()
3955                .expect("autoscale")
3956                .host_label()
3957                .as_str(),
3958            "rm-home-win-x64",
3959            "policies.routing_labels"
3960        );
3961        assert_eq!(
3962            policy.workspace_policy().root(),
3963            Some(&persistent_root),
3964            "policies.workspace_path"
3965        );
3966        assert_eq!(
3967            policy.workspace_policy().kind(),
3968            WorkspaceKind::Persistent,
3969            "policies.workspace_mode"
3970        );
3971
3972        RawAttempt {
3973            github_runner_id: Some(73),
3974            process_id: Some(4_242),
3975            state: "finished".to_string(),
3976            outcome: Some(COMPLETED_JOB.to_string()),
3977            runtime_path: "runtime/distinguishable".to_string(),
3978            workspace_mode: "persistent".to_string(),
3979            workspace_slot: Some(37),
3980            created_at: timestamp_to_text(ts(1_000)),
3981            last_state_change_at: timestamp_to_text(ts(2_000)),
3982            terminal_at: Some(timestamp_to_text(ts(3_000))),
3983            ..RawAttempt::default()
3984        }
3985        .insert(&store);
3986
3987        let attempt = store
3988            .attempt(attempt_id())
3989            .expect("loads")
3990            .expect("present");
3991        assert_eq!(attempt.id, attempt_id(), "attempts.id");
3992        assert_eq!(attempt.policy_id, policy_id(), "attempts.policy_id");
3993        assert_eq!(
3994            attempt.github_runner_id(),
3995            Some(73),
3996            "attempts.github_runner_id"
3997        );
3998        assert_eq!(attempt.process_id(), Some(4_242), "attempts.process_id");
3999        assert_eq!(attempt.state(), AttemptState::Finished, "attempts.state");
4000        assert_eq!(
4001            attempt.outcome(),
4002            Some(&AttemptOutcome::CompletedJob),
4003            "attempts.outcome"
4004        );
4005        assert_eq!(
4006            attempt.runtime_path(),
4007            Path::new("runtime/distinguishable"),
4008            "attempts.runtime_path"
4009        );
4010        assert_eq!(
4011            attempt.created_at,
4012            ts(1_000),
4013            "attempts.created_at must not come from attempts.last_state_change_at"
4014        );
4015        assert_eq!(
4016            attempt.last_state_change_at(),
4017            ts(2_000),
4018            "attempts.last_state_change_at must not come from attempts.created_at; \
4019             every recovery timeout is measured from it"
4020        );
4021        assert_eq!(
4022            attempt.terminal_at(),
4023            Some(ts(3_000)),
4024            "attempts.terminal_at"
4025        );
4026        assert_eq!(
4027            attempt.workspace().slot_number(),
4028            Some(37),
4029            "attempts.workspace_slot"
4030        );
4031        assert_eq!(
4032            attempt.workspace().kind(),
4033            WorkspaceKind::Persistent,
4034            "attempts.workspace_mode"
4035        );
4036    }
4037
4038    // -- hand-corrupted rows ------------------------------------------------
4039
4040    #[test]
4041    fn a_hand_corrupted_policy_shape_is_rejected_on_load() {
4042        // D19 says `MonitorOnly` requires both `routing_labels` and
4043        // `max_capacity` to be NULL and `Autoscale` requires both to be present.
4044        // The columns admit four combinations; two are illegal, and each gets its
4045        // own error rather than being coerced into something plausible.
4046        let store = store();
4047
4048        RawPolicy {
4049            routing_labels: Some(LABELS_JSON.to_string()),
4050            max_capacity: None,
4051            ..RawPolicy::default()
4052        }
4053        .insert(&store);
4054        assert!(
4055            matches!(
4056                store.policy(policy_id()),
4057                Err(StoreError::CorruptPolicy {
4058                    source: PolicyError::AutoscaleWithoutMaxCapacity,
4059                    ..
4060                })
4061            ),
4062            "labels without a ceiling could oversubscribe the host"
4063        );
4064
4065        RawPolicy {
4066            routing_labels: None,
4067            max_capacity: Some(2),
4068            ..RawPolicy::default()
4069        }
4070        .insert(&store);
4071        assert!(matches!(
4072            store.policy(policy_id()),
4073            Err(StoreError::CorruptPolicy {
4074                source: PolicyError::AutoscaleWithoutRoutingLabels,
4075                ..
4076            })
4077        ));
4078
4079        RawPolicy {
4080            routing_labels: None,
4081            max_capacity: None,
4082            min_capacity: 1,
4083            ..RawPolicy::default()
4084        }
4085        .insert(&store);
4086        assert!(matches!(
4087            store.policy(policy_id()),
4088            Err(StoreError::CorruptPolicy {
4089                source: PolicyError::MonitorOnlyWithMinCapacity { min: 1 },
4090                ..
4091            })
4092        ));
4093
4094        RawPolicy {
4095            min_capacity: 3,
4096            max_capacity: Some(2),
4097            ..RawPolicy::default()
4098        }
4099        .insert(&store);
4100        assert!(
4101            matches!(
4102                store.policy(policy_id()),
4103                Err(StoreError::CorruptPolicy {
4104                    source: PolicyError::InvertedCapacityRange { min: 3, max: 2 },
4105                    ..
4106                })
4107            ),
4108            "an inverted range makes clamp(demand, min, max) panic, so it must \
4109             not survive a load"
4110        );
4111
4112        // A scope/slug pair that cannot exist: `Org::new` rejects a slash.
4113        RawPolicy {
4114            target_scope: "organization".to_string(),
4115            target_slug: "o/r".to_string(),
4116            ..RawPolicy::default()
4117        }
4118        .insert(&store);
4119        assert!(
4120            matches!(
4121                store.policy(policy_id()),
4122                Err(StoreError::CorruptPolicy {
4123                    source: PolicyError::Invalid(ValidationError::IllegalCharacter {
4124                        found: '/',
4125                        ..
4126                    }),
4127                    ..
4128                })
4129            ),
4130            "the target is rebuilt through the real constructor, so GitHub's \
4131             naming rules run again on load"
4132        );
4133
4134        // And the columns that carry no domain type of their own.
4135        for (label, raw) in [
4136            (
4137                "a zero ceiling",
4138                RawPolicy {
4139                    max_capacity: Some(0),
4140                    ..RawPolicy::default()
4141                },
4142            ),
4143            (
4144                "a third spelling of enabled",
4145                RawPolicy {
4146                    enabled: 7,
4147                    ..RawPolicy::default()
4148                },
4149            ),
4150            (
4151                "an unrecognised state",
4152                RawPolicy {
4153                    state: "retired".to_string(),
4154                    ..RawPolicy::default()
4155                },
4156            ),
4157            (
4158                "a negative revision",
4159                RawPolicy {
4160                    revision: -1,
4161                    ..RawPolicy::default()
4162                },
4163            ),
4164            (
4165                "a routing label carrying the separator the runner splits on",
4166                RawPolicy {
4167                    routing_labels: Some(r#"{"host_label":"bad,label"}"#.to_string()),
4168                    ..RawPolicy::default()
4169                },
4170            ),
4171        ] {
4172            raw.insert(&store);
4173            assert!(
4174                matches!(
4175                    store.policy(policy_id()),
4176                    Err(StoreError::CorruptColumn { .. })
4177                ),
4178                "{label} must be reported as a corrupt column"
4179            );
4180        }
4181    }
4182
4183    #[test]
4184    fn a_hand_corrupted_host_row_is_rejected_on_load() {
4185        let store = store();
4186
4187        RawHost {
4188            refresh_interval_secs: 1,
4189            ..RawHost::default()
4190        }
4191        .insert(&store);
4192        assert!(
4193            matches!(
4194                store.host(host_id()),
4195                Err(StoreError::CorruptHost {
4196                    source: ValidationError::BelowFloor {
4197                        min: 30,
4198                        actual: 1,
4199                        ..
4200                    },
4201                    ..
4202                })
4203            ),
4204            "a hand-edited row must not make this host poll every second; the \
4205             floor is a rate-budget constraint"
4206        );
4207
4208        RawHost {
4209            display_name: "   ".to_string(),
4210            ..RawHost::default()
4211        }
4212        .insert(&store);
4213        assert!(matches!(
4214            store.host(host_id()),
4215            Err(StoreError::CorruptHost {
4216                source: ValidationError::Empty { .. },
4217                ..
4218            })
4219        ));
4220
4221        for (label, raw) in [
4222            (
4223                "zero capacity",
4224                RawHost {
4225                    host_capacity: 0,
4226                    ..RawHost::default()
4227                },
4228            ),
4229            (
4230                "an unsupported operating system",
4231                RawHost {
4232                    os: "plan9".to_string(),
4233                    ..RawHost::default()
4234                },
4235            ),
4236            (
4237                "a malformed created_at",
4238                RawHost {
4239                    created_at: "yesterday".to_string(),
4240                    ..RawHost::default()
4241                },
4242            ),
4243        ] {
4244            raw.insert(&store);
4245            assert!(
4246                matches!(store.host(host_id()), Err(StoreError::CorruptColumn { .. })),
4247                "{label} must be reported as a corrupt column"
4248            );
4249        }
4250    }
4251
4252    #[test]
4253    fn a_hand_corrupted_attempt_row_is_rejected_on_load() {
4254        let store = store();
4255
4256        for (label, raw) in [
4257            (
4258                "terminal with no outcome",
4259                RawAttempt {
4260                    state: "finished".to_string(),
4261                    outcome: None,
4262                    terminal_at: Some(timestamp_to_text(ts(2_000))),
4263                    ..RawAttempt::default()
4264                },
4265            ),
4266            (
4267                "non-terminal carrying an outcome",
4268                RawAttempt {
4269                    state: "busy".to_string(),
4270                    outcome: Some(COMPLETED_JOB.to_string()),
4271                    ..RawAttempt::default()
4272                },
4273            ),
4274            (
4275                "a failed attempt claiming it ran a job",
4276                RawAttempt {
4277                    state: "failed".to_string(),
4278                    outcome: Some(COMPLETED_JOB.to_string()),
4279                    terminal_at: Some(timestamp_to_text(ts(2_000))),
4280                    ..RawAttempt::default()
4281                },
4282            ),
4283            (
4284                "terminal with no terminal_at",
4285                RawAttempt {
4286                    state: "finished".to_string(),
4287                    outcome: Some(COMPLETED_JOB.to_string()),
4288                    terminal_at: None,
4289                    ..RawAttempt::default()
4290                },
4291            ),
4292        ] {
4293            raw.insert(&store);
4294            assert!(
4295                matches!(
4296                    store.attempt(attempt_id()),
4297                    Err(StoreError::CorruptAttempt { .. })
4298                ),
4299                "{label} must not load"
4300            );
4301        }
4302
4303        for (label, raw) in [
4304            (
4305                "a negative process id",
4306                RawAttempt {
4307                    process_id: Some(-1),
4308                    ..RawAttempt::default()
4309                },
4310            ),
4311            (
4312                "an unrecognised state",
4313                RawAttempt {
4314                    state: "wedged".to_string(),
4315                    ..RawAttempt::default()
4316                },
4317            ),
4318            (
4319                "an outcome that is not one",
4320                RawAttempt {
4321                    state: "finished".to_string(),
4322                    outcome: Some(r#"{"outcome":"went_home"}"#.to_string()),
4323                    terminal_at: Some(timestamp_to_text(ts(2_000))),
4324                    ..RawAttempt::default()
4325                },
4326            ),
4327            (
4328                "a malformed created_at",
4329                RawAttempt {
4330                    created_at: "yesterday".to_string(),
4331                    ..RawAttempt::default()
4332                },
4333            ),
4334        ] {
4335            raw.insert(&store);
4336            assert!(
4337                matches!(
4338                    store.attempt(attempt_id()),
4339                    Err(StoreError::CorruptColumn { .. })
4340                ),
4341                "{label} must be reported as a corrupt column"
4342            );
4343        }
4344    }
4345
4346    // -- hand-corrupted workspace columns -----------------------------------
4347
4348    #[test]
4349    fn a_hand_corrupted_host_runner_root_is_rejected_on_load() {
4350        // D10 and `02-target-architecture.md`'s "Path validation": the pure
4351        // stored-shape rules run at database load, with no filesystem probe, so
4352        // a hand-edited row cannot make this product place a runner on a network
4353        // share or above a filesystem root. The refusal names the reason,
4354        // because a path is not a credential and an operator has to be able to
4355        // fix the row.
4356        let store = store();
4357
4358        for (label, raw) in [
4359            ("a UNC share", r"\\nas\builds"),
4360            ("a relative path", "runners"),
4361            (
4362                "a traversal component",
4363                if cfg!(windows) {
4364                    r"X:\rman\..\elsewhere"
4365                } else {
4366                    "/srv/rman/../elsewhere"
4367                },
4368            ),
4369            (
4370                "a bare filesystem root",
4371                if cfg!(windows) { r"X:\" } else { "/" },
4372            ),
4373            // The one rule that is about *this* host rather than about syntax:
4374            // "an absolute path native to the current host". The foreign
4375            // spelling is a shape this build will not place a runner under.
4376            (
4377                "a path from the other platform",
4378                if cfg!(windows) {
4379                    "/srv/rman"
4380                } else {
4381                    r"X:\rman"
4382                },
4383            ),
4384        ] {
4385            RawHost {
4386                runner_root_override: Some(raw.to_string()),
4387                ..RawHost::default()
4388            }
4389            .insert(&store);
4390
4391            let error = store
4392                .host(host_id())
4393                .expect_err(&format!("{label} must not load"));
4394            assert!(
4395                matches!(error, StoreError::CorruptHostWorkspace { id, .. } if id == host_id()),
4396                "{label} must be reported against the host row, got {error:?}"
4397            );
4398            assert!(
4399                !error.is_conflict(),
4400                "{label} is corrupt state, not a concurrency conflict"
4401            );
4402        }
4403
4404        // And the shape that *is* legal still loads, so the check above is not
4405        // refusing everything.
4406        let configured = a_root("rman");
4407        RawHost {
4408            runner_root_override: Some(configured.as_str().to_string()),
4409            ..RawHost::default()
4410        }
4411        .insert(&store);
4412        let host = store.host(host_id()).expect("loads").expect("present");
4413        assert_eq!(host.runner_root_override, Some(configured));
4414        assert!(host.has_configured_runner_root());
4415    }
4416
4417    #[test]
4418    fn a_hand_corrupted_policy_workspace_is_rejected_on_load() {
4419        // The four illegal combinations of `workspace_mode` and
4420        // `workspace_path`, plus D7. `03-migration-rollout.md` states the shape
4421        // rule the loader enforces: "policy ephemeral -> workspace_path IS NULL;
4422        // policy persistent -> repository scope and workspace_path IS NOT NULL".
4423        let store = store();
4424        let root = a_root("workspaces");
4425
4426        for (label, raw) in [
4427            (
4428                "persistent without a path",
4429                RawPolicy {
4430                    workspace_mode: "persistent".to_string(),
4431                    workspace_path: None,
4432                    ..RawPolicy::default()
4433                },
4434            ),
4435            (
4436                "ephemeral with a stale path",
4437                RawPolicy {
4438                    workspace_mode: "ephemeral".to_string(),
4439                    workspace_path: Some(root.as_str().to_string()),
4440                    ..RawPolicy::default()
4441                },
4442            ),
4443            (
4444                "an organization policy claiming to retain a workspace",
4445                RawPolicy {
4446                    target_scope: "organization".to_string(),
4447                    target_slug: "tap-top-fun".to_string(),
4448                    workspace_mode: "persistent".to_string(),
4449                    workspace_path: Some(root.as_str().to_string()),
4450                    ..RawPolicy::default()
4451                },
4452            ),
4453            (
4454                "a persistent root that is not a storable local path",
4455                RawPolicy {
4456                    workspace_mode: "persistent".to_string(),
4457                    workspace_path: Some(r"\\nas\builds".to_string()),
4458                    ..RawPolicy::default()
4459                },
4460            ),
4461        ] {
4462            raw.insert(&store);
4463            let error = store
4464                .policy(policy_id())
4465                .expect_err(&format!("{label} must not load"));
4466            assert!(
4467                matches!(error, StoreError::CorruptPolicy { id, .. } if id == policy_id()),
4468                "{label} must be reported against the policy row, got {error:?}"
4469            );
4470        }
4471
4472        // An unrecognised mode is a column failure rather than a shape failure,
4473        // and it must not fall back to the default: an upgrade that read
4474        // `persistent-ish` as `ephemeral` would delete a retained workspace.
4475        RawPolicy {
4476            workspace_mode: "sticky".to_string(),
4477            ..RawPolicy::default()
4478        }
4479        .insert(&store);
4480        assert!(
4481            matches!(
4482                store.policy(policy_id()),
4483                Err(StoreError::CorruptColumn {
4484                    table: "policies",
4485                    column: "workspace_mode",
4486                    ..
4487                })
4488            ),
4489            "an unknown workspace mode must fail closed"
4490        );
4491
4492        // The legal persistent shape still loads.
4493        RawPolicy {
4494            workspace_mode: "persistent".to_string(),
4495            workspace_path: Some(root.as_str().to_string()),
4496            ..RawPolicy::default()
4497        }
4498        .insert(&store);
4499        let policy = store.policy(policy_id()).expect("loads").expect("present");
4500        assert_eq!(policy.workspace_policy().root(), Some(&root));
4501        assert!(policy.workspace_policy().retains_job_workspace());
4502    }
4503
4504    #[test]
4505    fn a_hand_corrupted_attempt_workspace_is_rejected_on_load() {
4506        // `04-security-recovery.md`, "Corrupt SQLite changes cleanup mode for an
4507        // old path": the journalled kind decides which cleanup algorithm is
4508        // legal, so a pair that cannot describe one allocation must not load at
4509        // all rather than fall through to the destructive branch.
4510        let store = store();
4511
4512        for (label, raw) in [
4513            (
4514                "persistent without a slot",
4515                RawAttempt {
4516                    workspace_mode: "persistent".to_string(),
4517                    workspace_slot: None,
4518                    ..RawAttempt::default()
4519                },
4520            ),
4521            (
4522                "ephemeral holding a slot",
4523                RawAttempt {
4524                    workspace_mode: "ephemeral".to_string(),
4525                    workspace_slot: Some(1),
4526                    ..RawAttempt::default()
4527                },
4528            ),
4529            (
4530                "slot zero, which names no directory",
4531                RawAttempt {
4532                    workspace_mode: "persistent".to_string(),
4533                    workspace_slot: Some(0),
4534                    ..RawAttempt::default()
4535                },
4536            ),
4537        ] {
4538            raw.insert(&store);
4539            assert!(
4540                matches!(
4541                    store.attempt(attempt_id()),
4542                    Err(StoreError::CorruptAttempt { .. })
4543                ),
4544                "{label} must not load"
4545            );
4546        }
4547
4548        for (label, raw) in [
4549            (
4550                "an unrecognised mode",
4551                RawAttempt {
4552                    workspace_mode: "sticky".to_string(),
4553                    ..RawAttempt::default()
4554                },
4555            ),
4556            (
4557                "a slot above u16",
4558                RawAttempt {
4559                    workspace_mode: "persistent".to_string(),
4560                    workspace_slot: Some(70_000),
4561                    ..RawAttempt::default()
4562                },
4563            ),
4564        ] {
4565            raw.insert(&store);
4566            assert!(
4567                matches!(
4568                    store.attempt(attempt_id()),
4569                    Err(StoreError::CorruptColumn {
4570                        table: "attempts",
4571                        ..
4572                    })
4573                ),
4574                "{label} must be reported as a corrupt column"
4575            );
4576        }
4577
4578        RawAttempt {
4579            workspace_mode: "persistent".to_string(),
4580            workspace_slot: Some(2),
4581            ..RawAttempt::default()
4582        }
4583        .insert(&store);
4584        let attempt = store
4585            .attempt(attempt_id())
4586            .expect("loads")
4587            .expect("present");
4588        assert_eq!(
4589            attempt.workspace(),
4590            AttemptWorkspace::persistent_slot(NonZeroU16::new(2).expect("non-zero"))
4591        );
4592        assert!(attempt.holds_slot_lease());
4593    }
4594
4595    // -- durable slot leases ------------------------------------------------
4596
4597    /// The outcome a terminal state has to be paired with, and `None` for a
4598    /// non-terminal one.
4599    ///
4600    /// `RunnerAttempt::from_persisted` refuses a terminal row without an outcome
4601    /// and one whose outcome names a different terminal state, so a test that
4602    /// puts an attempt into an arbitrary state has to supply the matching value
4603    /// rather than one fixed one. `Cleaned` is the exception the loader itself
4604    /// makes -- it follows any of the three -- and takes the ordinary one.
4605    fn outcome_for(state: AttemptState) -> Option<AttemptOutcome> {
4606        match state {
4607            AttemptState::Failed => Some(AttemptOutcome::failed(
4608                FailureReason::ProcessExitedUnexpectedly,
4609            )),
4610            AttemptState::Orphaned => Some(AttemptOutcome::Orphaned),
4611            AttemptState::Finished | AttemptState::Cleaned => Some(AttemptOutcome::CompletedJob),
4612            _ => None,
4613        }
4614    }
4615
4616    /// The same attempt in a different state, rebuilt rather than transitioned.
4617    ///
4618    /// One helper covers every state without these tests knowing the
4619    /// lifecycle's edge list, which is `b1`'s to own and is tested there.
4620    fn attempt_in_state(attempt: &RunnerAttempt, state: AttemptState) -> RunnerAttempt {
4621        let mut fields = attempt.to_persisted();
4622        fields.state = state;
4623        fields.outcome = outcome_for(state);
4624        fields.terminal_at = state.is_terminal().then(|| ts(2_000));
4625        fields.last_state_change_at = ts(2_000);
4626        RunnerAttempt::from_persisted(fields).expect("a state the domain accepts")
4627    }
4628
4629    /// A journalled persistent attempt on `slot`, in `state`.
4630    fn persistent_attempt(id: u128, slot: u16, state: AttemptState) -> RunnerAttempt {
4631        let attempt = RunnerAttempt::allocate_in(
4632            AttemptId::from_u128(id),
4633            policy_id(),
4634            format!("root/s{slot}"),
4635            AttemptWorkspace::persistent_slot(NonZeroU16::new(slot).expect("positive")),
4636            ts(1_000),
4637        );
4638        if state == AttemptState::Allocated {
4639            return attempt;
4640        }
4641        attempt_in_state(&attempt, state)
4642    }
4643
4644    #[test]
4645    fn one_slot_is_leased_to_at_most_one_uncleaned_attempt() {
4646        // Invariant 5, enforced durably. The allocation lock coordinates
4647        // selection; this index is what catches the race the lock cannot see,
4648        // and `04-security-recovery.md` names it as the control for "two
4649        // attempts use one slot concurrently".
4650        let store = store();
4651        let first = persistent_attempt(0x501, 1, AttemptState::Idle);
4652        store.record_attempt(&first).expect("the first lease");
4653
4654        let second = persistent_attempt(0x502, 1, AttemptState::Allocated);
4655        let error = store
4656            .record_attempt(&second)
4657            .expect_err("a second uncleaned attempt must not take a leased slot");
4658        assert!(
4659            matches!(
4660                error,
4661                StoreError::SlotAlreadyLeased { policy, slot }
4662                    if policy == policy_id() && slot == 1
4663            ),
4664            "expected SlotAlreadyLeased, got {error:?}"
4665        );
4666        assert!(
4667            error.is_conflict(),
4668            "an allocator that lost a slot picks another one; this is not an \
4669             I/O failure"
4670        );
4671
4672        // Atomic: the refused write left nothing behind, and the existing lease
4673        // is untouched.
4674        assert!(
4675            store.attempt(second.id).expect("loads").is_none(),
4676            "the refused insert must not be half-applied"
4677        );
4678        let leases = store.slot_leases_for_policy(policy_id()).expect("loads");
4679        assert_eq!(leases.len(), 1);
4680        assert_eq!(leases[0].id, first.id);
4681
4682        // A *different* slot is free, and so is the same slot under a different
4683        // policy: the index is scoped to the pair.
4684        store
4685            .record_attempt(&persistent_attempt(0x503, 2, AttemptState::Allocated))
4686            .expect("slot 2 is not leased");
4687        let other_policy = RunnerAttempt::allocate_in(
4688            AttemptId::from_u128(0x504),
4689            PolicyId::from_u128(0x11),
4690            "root/s1",
4691            AttemptWorkspace::persistent_slot(NonZeroU16::new(1).expect("positive")),
4692            ts(1_000),
4693        );
4694        store
4695            .record_attempt(&other_policy)
4696            .expect("another policy's s1 is a different slot");
4697    }
4698
4699    #[test]
4700    fn a_cleaned_historical_attempt_releases_its_slot_for_reuse() {
4701        // "Persistent directories provide retained bytes but never lease truth."
4702        // The lease is the uncleaned row, so cleaning it is what frees `s1` --
4703        // and the historical row stays in the journal, because `g2`'s history
4704        // and `04`'s audit trail both read it.
4705        let store = store();
4706        let first = persistent_attempt(0x511, 1, AttemptState::Cleaned);
4707        store.record_attempt(&first).expect("a cleaned lease");
4708        assert!(!first.holds_slot_lease());
4709        assert!(
4710            store
4711                .slot_leases_for_policy(policy_id())
4712                .expect("loads")
4713                .is_empty(),
4714            "a cleaned attempt holds no lease"
4715        );
4716
4717        let second = persistent_attempt(0x512, 1, AttemptState::Allocated);
4718        store
4719            .record_attempt(&second)
4720            .expect("a cleaned row must not block reuse of its slot");
4721
4722        assert_eq!(
4723            store.attempts_for_policy(policy_id()).expect("loads").len(),
4724            2,
4725            "the historical row is kept, not deleted, when the slot is reused"
4726        );
4727
4728        // A third attempt now collides with the *live* lease rather than with
4729        // the historical row.
4730        assert!(matches!(
4731            store.record_attempt(&persistent_attempt(0x513, 1, AttemptState::Allocated)),
4732            Err(StoreError::SlotAlreadyLeased { slot: 1, .. })
4733        ));
4734    }
4735
4736    #[test]
4737    fn a_terminal_attempt_awaiting_cleanup_still_holds_its_slot() {
4738        // `04-security-recovery.md`, "Cleanup partly fails and the slot is
4739        // reused anyway": the attempt "remains not-cleaned and continues to hold
4740        // the slot through the unique lease index; it does not count as active
4741        // host capacity".
4742        let store = store();
4743        let failed_cleanup = persistent_attempt(0x521, 1, AttemptState::Finished);
4744        store.record_attempt(&failed_cleanup).expect("journalled");
4745
4746        assert!(
4747            store
4748                .active_attempts_for_policy(policy_id())
4749                .expect("loads")
4750                .is_empty(),
4751            "a terminal attempt occupies no capacity slot"
4752        );
4753        let leases = store.slot_leases_for_policy(policy_id()).expect("loads");
4754        assert_eq!(leases.len(), 1, "but it does still hold its lease");
4755        assert_eq!(leases[0].workspace().slot_number(), Some(1));
4756
4757        assert!(matches!(
4758            store.record_attempt(&persistent_attempt(0x522, 1, AttemptState::Allocated)),
4759            Err(StoreError::SlotAlreadyLeased { slot: 1, .. })
4760        ));
4761    }
4762
4763    #[test]
4764    fn journalling_the_same_attempt_again_cannot_move_its_lease() {
4765        // The allocation fact is immutable in the domain and write-once in the
4766        // journal: `record_attempt` is called repeatedly over one attempt's
4767        // life, and none of those calls may rewrite the kind or the slot that
4768        // decides which cleanup algorithm is legal on its directory.
4769        let store = store();
4770        let allocated = persistent_attempt(0x531, 1, AttemptState::Allocated);
4771        store.record_attempt(&allocated).expect("journalled");
4772
4773        let mut fields = allocated.to_persisted();
4774        fields.state = AttemptState::Idle;
4775        fields.last_state_change_at = ts(2_000);
4776        fields.workspace_slot = Some(9);
4777        let moved = RunnerAttempt::from_persisted(fields).expect("a legal attempt in isolation");
4778        store
4779            .record_attempt(&moved)
4780            .expect("the state change lands");
4781
4782        let stored = store
4783            .attempt(allocated.id)
4784            .expect("loads")
4785            .expect("present");
4786        assert_eq!(stored.state(), AttemptState::Idle, "the state did move");
4787        assert_eq!(
4788            stored.workspace().slot_number(),
4789            Some(1),
4790            "the slot journalled at allocation is the one that stands"
4791        );
4792    }
4793
4794    // -- attempt sets -------------------------------------------------------
4795
4796    #[test]
4797    fn the_attempt_set_predicates_follow_the_domain() {
4798        // The SQL fragments are derived from `AttemptState`'s own predicates and
4799        // tokens; this is the assertion that the derivation and the domain agree
4800        // state by state, so a tenth state cannot be counted by the capacity
4801        // formula and silently missed by the fence.
4802        let store = store();
4803        for (index, state) in AttemptState::ALL.into_iter().enumerate() {
4804            // Ephemeral, so that every state can coexist under one policy
4805            // without the lease index refusing the second one.
4806            let attempt = RunnerAttempt::allocate(
4807                AttemptId::from_u128(0x600 + index as u128),
4808                policy_id(),
4809                format!("runtime/{state}"),
4810                ts(1_000),
4811            );
4812            store
4813                .record_attempt(&attempt_in_state(&attempt, state))
4814                .expect("journalled");
4815        }
4816
4817        let expected_active = AttemptState::ALL
4818            .into_iter()
4819            .filter(|state| state.counts_against_capacity())
4820            .count();
4821        let expected_uncleaned = AttemptState::ALL
4822            .into_iter()
4823            .filter(|state| *state != AttemptState::Cleaned)
4824            .count();
4825        assert_ne!(
4826            expected_active, expected_uncleaned,
4827            "if these were equal the two fences would be the same fence and this \
4828             test would prove nothing"
4829        );
4830
4831        assert_eq!(
4832            store
4833                .active_attempts_for_policy(policy_id())
4834                .expect("loads")
4835                .len(),
4836            expected_active
4837        );
4838        assert_eq!(
4839            store
4840                .uncleaned_attempts_for_policy(policy_id())
4841                .expect("loads")
4842                .len(),
4843            expected_uncleaned
4844        );
4845        assert_eq!(
4846            store.uncleaned_ephemeral_attempts().expect("loads").len(),
4847            expected_uncleaned,
4848            "every attempt here is ephemeral, so the host-wide set is the same \
4849             size as the per-policy uncleaned one"
4850        );
4851        assert!(
4852            store
4853                .slot_leases_for_policy(policy_id())
4854                .expect("loads")
4855                .is_empty(),
4856            "and none of them is a slot lease"
4857        );
4858    }
4859
4860    #[test]
4861    fn the_host_wide_ephemeral_set_counts_an_attempt_that_outlived_its_policy() {
4862        // The note on `Store::uncleaned_ephemeral_attempts`: an attempt whose
4863        // policy row is gone still owns a directory under the host runner root,
4864        // and `04-security-recovery.md` requires unknown-policy attempts to keep
4865        // their fail-closed ownership behaviour. A count scoped to this host's
4866        // policies would drop exactly those.
4867        let store = store();
4868        let orphan = RunnerAttempt::allocate(
4869            AttemptId::from_u128(0x701),
4870            PolicyId::from_u128(0xdead),
4871            "runtime/orphan",
4872            ts(1_000),
4873        );
4874        store.record_attempt(&orphan).expect("journalled");
4875        store
4876            .record_attempt(&persistent_attempt(0x702, 1, AttemptState::Allocated))
4877            .expect("journalled");
4878
4879        let ephemeral = store.uncleaned_ephemeral_attempts().expect("loads");
4880        assert_eq!(ephemeral.len(), 1, "the persistent attempt is not counted");
4881        assert_eq!(ephemeral[0].id, orphan.id);
4882    }
4883
4884    // -- targeted host-root mutation ----------------------------------------
4885
4886    /// The store, its host, and a `Host` value equal to the stored row.
4887    fn a_stored_host(store: &SqliteStore) -> Host {
4888        let host = Host::new(
4889            host_id(),
4890            "home-pc",
4891            Os::Windows,
4892            Arch::X64,
4893            NonZeroU16::new(2).expect("non-zero"),
4894            ts(1_000),
4895        )
4896        .expect("valid");
4897        store.put_host(&host).expect("stored");
4898        host
4899    }
4900
4901    #[test]
4902    fn the_host_root_mutation_writes_only_its_own_column() {
4903        // `02-target-architecture.md`: the targeted mutation exists so that "a
4904        // simultaneous capacity or service-mode change" is not overwritten. This
4905        // is that property, stated as a test: the concurrent change is made
4906        // *between* the read and the write, and it survives.
4907        let store = store();
4908        let read = a_stored_host(&store);
4909
4910        let mut concurrent = read.clone();
4911        concurrent.host_capacity = NonZeroU16::new(7).expect("non-zero");
4912        concurrent.service_start_mode = StartMode::Login;
4913        store.put_host(&concurrent).expect("the other writer wins");
4914
4915        let configured = a_root("rman");
4916        store
4917            .set_runner_root_override(host_id(), None, Some(&configured), 0)
4918            .expect("the root moves");
4919
4920        let stored = store.host(host_id()).expect("loads").expect("present");
4921        assert_eq!(stored.runner_root_override, Some(configured));
4922        assert_eq!(
4923            stored.host_capacity.get(),
4924            7,
4925            "a whole-record write built from the stale read would have rolled \
4926             this back to 2"
4927        );
4928        assert_eq!(stored.service_start_mode, StartMode::Login);
4929    }
4930
4931    #[test]
4932    fn the_host_root_mutation_refuses_a_changed_expected_override() {
4933        let store = store();
4934        a_stored_host(&store);
4935        let first = a_root("rman");
4936        let second = a_root("elsewhere");
4937
4938        store
4939            .set_runner_root_override(host_id(), None, Some(&first), 0)
4940            .expect("the first mutation");
4941
4942        // A second operator read `None` before the first landed.
4943        let error = store
4944            .set_runner_root_override(host_id(), None, Some(&second), 0)
4945            .expect_err("a stale expected override must be refused");
4946        assert!(
4947            matches!(&error, StoreError::RunnerRootChanged { id, .. } if *id == host_id()),
4948            "expected RunnerRootChanged, got {error:?}"
4949        );
4950        assert!(error.is_conflict());
4951        let message = error.to_string();
4952        assert!(
4953            message.contains("the platform default") && message.contains(first.as_str()),
4954            "the message must name both sides so the operator can re-read: \
4955             {message}"
4956        );
4957        assert_eq!(
4958            store
4959                .host(host_id())
4960                .expect("loads")
4961                .expect("present")
4962                .runner_root_override,
4963            Some(first.clone()),
4964            "nothing was written"
4965        );
4966
4967        // Resetting to the default is the same mutation in the other direction,
4968        // and it works once the expected value is the one actually stored.
4969        store
4970            .set_runner_root_override(host_id(), Some(&first), None, 0)
4971            .expect("reset to the platform default");
4972        assert_eq!(
4973            store
4974                .host(host_id())
4975                .expect("loads")
4976                .expect("present")
4977                .runner_root_override,
4978            None
4979        );
4980    }
4981
4982    #[test]
4983    fn the_host_root_mutation_refuses_a_changed_uncleaned_ephemeral_count() {
4984        // `04-security-recovery.md`: "A host root setting cannot change while
4985        // any ephemeral attempt is active or unresolved." The count is confirmed
4986        // in the same transaction as the write, so an attempt journalled between
4987        // the operator's read and the write refuses it.
4988        let store = store();
4989        a_stored_host(&store);
4990        let configured = a_root("rman");
4991
4992        let mut attempt = RunnerAttempt::allocate(
4993            attempt_id(),
4994            policy_id(),
4995            "runtime/policy/attempt",
4996            ts(1_000),
4997        );
4998        store.record_attempt(&attempt).expect("journalled");
4999
5000        let error = store
5001            .set_runner_root_override(host_id(), None, Some(&configured), 0)
5002            .expect_err("an attempt appeared after the operator counted zero");
5003        assert!(
5004            matches!(
5005                &error,
5006                StoreError::UncleanedCountChanged { expected: 0, found: 1, subject }
5007                    if subject.contains(&host_id().to_string())
5008            ),
5009            "expected UncleanedCountChanged naming the host, got {error:?}"
5010        );
5011        assert!(error.is_conflict());
5012
5013        // Confirming the count that is actually there is a refusal at the
5014        // command layer, not here: this store call is the fence, and it commits
5015        // when the caller's observation was correct.
5016        store
5017            .set_runner_root_override(host_id(), None, Some(&configured), 1)
5018            .expect("the confirmed count matches");
5019
5020        // A *terminal but uncleaned* attempt still counts. This is the whole
5021        // difference between this fence and the active-count one.
5022        attempt = attempt_in_state(&attempt, AttemptState::Finished);
5023        store.record_attempt(&attempt).expect("journalled");
5024        assert!(
5025            store
5026                .active_attempts_for_policy(policy_id())
5027                .expect("loads")
5028                .is_empty()
5029        );
5030        assert!(matches!(
5031            store.set_runner_root_override(host_id(), Some(&configured), None, 0),
5032            Err(StoreError::UncleanedCountChanged { found: 1, .. })
5033        ));
5034    }
5035
5036    #[test]
5037    fn the_host_root_mutation_reports_a_missing_host_rather_than_a_conflict() {
5038        let store = store();
5039        let error = store
5040            .set_runner_root_override(host_id(), None, Some(&a_root("rman")), 0)
5041            .expect_err("there is no such host");
5042        assert!(
5043            matches!(error, StoreError::NotFound { what: "host", .. }),
5044            "expected NotFound, got {error:?}"
5045        );
5046        assert!(!error.is_conflict());
5047    }
5048
5049    // -- policy mutation fenced on the uncleaned count -----------------------
5050
5051    /// The autoscale policy these tests mutate, in the store.
5052    ///
5053    /// The sibling of [`a_stored_host`]; `testkit`'s `fixtures::policy()` builds
5054    /// exactly this object but is unreachable from here, for the
5055    /// dev-dependency-cycle reason `tests/store_journal.rs` opens with.
5056    fn a_stored_policy(store: &SqliteStore) -> ScalePolicy {
5057        let policy = ScalePolicy::new(
5058            policy_id(),
5059            ScaleTarget::repository("o/r").expect("valid"),
5060            1,
5061            host_id(),
5062            PolicyMode::autoscale(
5063                RoutingLabels::from_host_label(Label::new("rm-home-win-x64").expect("valid")),
5064                0,
5065                NonZeroU16::new(2).expect("non-zero"),
5066            )
5067            .expect("valid"),
5068            CachePolicy::default(),
5069        );
5070        store.insert_policy(&policy).expect("inserted");
5071        policy
5072    }
5073
5074    #[test]
5075    fn the_policy_workspace_mutation_confirms_revision_and_uncleaned_count() {
5076        // `03-migration-rollout.md`: "The policy store operation compares its
5077        // revision and confirms the uncleaned policy-attempt count. Both checks
5078        // happen inside the same SQLite write transaction as the mutation."
5079        let store = store();
5080        let mut policy = a_stored_policy(&store);
5081
5082        let read_revision = policy.revision();
5083        let root = a_root("workspaces");
5084        policy
5085            .set_workspace_policy(
5086                WorkspacePolicy::persistent(root.clone(), TargetScope::Repository)
5087                    .expect("a repository may"),
5088            )
5089            .expect("a repository may");
5090
5091        // A stale revision is refused even when the count is right.
5092        assert!(matches!(
5093            store.update_policy_confirming_uncleaned_count(&policy, read_revision + 7, 0),
5094            Err(StoreError::StaleRevision { .. })
5095        ));
5096
5097        // A terminal-but-uncleaned attempt refuses the path change, which is the
5098        // case `update_policy_confirming_active_count` would have let through.
5099        let attempt =
5100            RunnerAttempt::allocate(attempt_id(), policy_id(), "runtime/attempt", ts(1_000));
5101        store
5102            .record_attempt(&attempt_in_state(&attempt, AttemptState::Failed))
5103            .expect("journalled");
5104
5105        let error = store
5106            .update_policy_confirming_uncleaned_count(&policy, read_revision, 0)
5107            .expect_err("an unresolved attempt appeared");
5108        assert!(
5109            matches!(
5110                &error,
5111                StoreError::UncleanedCountChanged { expected: 0, found: 1, subject }
5112                    if subject.contains(&policy_id().to_string())
5113            ),
5114            "expected UncleanedCountChanged naming the policy, got {error:?}"
5115        );
5116        assert!(error.is_conflict());
5117        assert_eq!(
5118            store
5119                .policy(policy_id())
5120                .expect("loads")
5121                .expect("present")
5122                .workspace_policy(),
5123            &WorkspacePolicy::Ephemeral,
5124            "nothing was written"
5125        );
5126        // The same write through the active-count guard would have committed,
5127        // which is why the two guards are not one.
5128        assert!(
5129            store
5130                .update_policy_confirming_active_count(&policy, read_revision, 0)
5131                .is_ok(),
5132            "an unresolved attempt is invisible to the active-count guard"
5133        );
5134    }
5135
5136    #[test]
5137    fn a_workspace_policy_survives_a_guarded_write_and_a_reload() {
5138        let store = store();
5139        let mut policy = a_stored_policy(&store);
5140
5141        let root = a_root("workspaces");
5142        let read_revision = policy.revision();
5143        policy
5144            .set_workspace_policy(
5145                WorkspacePolicy::persistent(root.clone(), TargetScope::Repository)
5146                    .expect("a repository may"),
5147            )
5148            .expect("a repository may");
5149        store
5150            .update_policy_confirming_uncleaned_count(&policy, read_revision, 0)
5151            .expect("no attempt stands in the way");
5152
5153        let stored = store.policy(policy_id()).expect("loads").expect("present");
5154        assert_eq!(&stored, &policy);
5155        assert_eq!(stored.workspace_policy().root(), Some(&root));
5156
5157        // And back to ephemeral, which must clear the path rather than leave a
5158        // stale one the loader would refuse.
5159        let read_revision = stored.revision();
5160        let mut back = stored;
5161        back.set_workspace_policy(WorkspacePolicy::Ephemeral)
5162            .expect("always permitted");
5163        store
5164            .update_policy_confirming_uncleaned_count(&back, read_revision, 0)
5165            .expect("no attempt stands in the way");
5166        let stored = store.policy(policy_id()).expect("loads").expect("present");
5167        assert_eq!(stored.workspace_policy(), &WorkspacePolicy::Ephemeral);
5168        let raw: Option<String> = store
5169            .lock()
5170            .query_row(
5171                "SELECT workspace_path FROM policies WHERE id = :id",
5172                named_params! { ":id": POLICY_UUID },
5173                |row| row.get(0),
5174            )
5175            .expect("readable");
5176        assert_eq!(raw, None, "the column is cleared, not merely ignored");
5177    }
5178
5179    // -- optimistic concurrency ---------------------------------------------
5180
5181    #[test]
5182    fn active_count_guard_fences_zero_to_one_and_one_to_zero_attempt_writes() {
5183        for starts_active in [false, true] {
5184            ATTEMPT_WRITE_BLOCKED.store(false, Ordering::Release);
5185            let directory = tempfile::TempDir::new().expect("temporary database directory");
5186            let path = directory.path().join("state.sqlite3");
5187            let updater = SqliteStore::open(&path).expect("update connection");
5188            let writer = Arc::new(SqliteStore::open(&path).expect("attempt connection"));
5189            writer
5190                .lock()
5191                .busy_handler(Some(mark_attempt_write_blocked))
5192                .expect("test busy observer");
5193
5194            let mut policy = ScalePolicy::new(
5195                policy_id(),
5196                ScaleTarget::repository("o/r").expect("valid"),
5197                1,
5198                host_id(),
5199                PolicyMode::autoscale(
5200                    RoutingLabels::from_host_label(Label::new("rm-home-win-x64").expect("valid")),
5201                    0,
5202                    NonZeroU16::new(2).expect("non-zero"),
5203                )
5204                .expect("valid"),
5205                CachePolicy::default(),
5206            );
5207            policy.activate().expect("active policy");
5208            updater.insert_policy(&policy).expect("inserted");
5209            let attempt = RunnerAttempt::allocate(attempt_id(), policy.id, "runner", ts(1_000));
5210            if starts_active {
5211                updater.record_attempt(&attempt).expect("active attempt");
5212            }
5213
5214            let expected_active = u16::from(starts_active);
5215            let mut disabled = policy.clone();
5216            disabled.request_disable().expect("disable requested");
5217            if !starts_active {
5218                disabled
5219                    .drain_completed(0)
5220                    .expect("zero drains immediately");
5221            }
5222
5223            let (begin_tx, write_attempt) = mpsc::channel();
5224            let (finished_tx, finished_rx) = mpsc::channel();
5225            let writer_thread = Arc::clone(&writer);
5226            let attempt_for_thread = attempt.clone();
5227            let handle = std::thread::spawn(move || {
5228                write_attempt.recv().expect("transaction began");
5229                if starts_active {
5230                    writer_thread
5231                        .remove_attempt(attempt_for_thread.id)
5232                        .expect("completion write");
5233                } else {
5234                    writer_thread
5235                        .record_attempt(&attempt_for_thread)
5236                        .expect("allocation write");
5237                }
5238                finished_tx.send(()).expect("completion observed");
5239            });
5240
5241            updater
5242                .update_policy_confirming_active_count_with(
5243                    &disabled,
5244                    policy.revision(),
5245                    expected_active,
5246                    || {
5247                        begin_tx.send(()).expect("release attempt writer");
5248                        let deadline = Instant::now() + Duration::from_secs(5);
5249                        while !ATTEMPT_WRITE_BLOCKED.load(Ordering::Acquire) {
5250                            assert!(
5251                                Instant::now() < deadline,
5252                                "attempt writer never reached SQLite's allocation fence"
5253                            );
5254                            std::thread::yield_now();
5255                        }
5256                        assert!(
5257                            matches!(finished_rx.try_recv(), Err(mpsc::TryRecvError::Empty)),
5258                            "attempt mutation crossed the policy transaction"
5259                        );
5260                    },
5261                )
5262                .expect("confirmed count commits while the attempt writer is fenced");
5263            handle
5264                .join()
5265                .expect("attempt writer completed after commit");
5266            finished_rx
5267                .recv()
5268                .expect("attempt mutation eventually commits");
5269
5270            let stored = updater.policy(policy.id).unwrap().unwrap();
5271            assert!(!stored.enabled());
5272            assert_eq!(
5273                updater.attempt(attempt.id).unwrap().is_some(),
5274                !starts_active,
5275                "the attempt mutation must occur only after policy persistence"
5276            );
5277        }
5278    }
5279
5280    #[test]
5281    fn a_stale_revision_write_is_rejected_and_is_not_an_io_error() {
5282        let store = store();
5283        let mut policy = ScalePolicy::new(
5284            policy_id(),
5285            ScaleTarget::repository("o/r").expect("valid"),
5286            1,
5287            host_id(),
5288            PolicyMode::autoscale(
5289                RoutingLabels::from_host_label(Label::new("rm-home-win-x64").expect("valid")),
5290                0,
5291                NonZeroU16::new(2).expect("non-zero"),
5292            )
5293            .expect("valid"),
5294            CachePolicy::default(),
5295        );
5296        store.insert_policy(&policy).expect("inserted");
5297        assert_eq!(policy.revision(), 0);
5298
5299        // The TUI reads revision 0 and enables the policy.
5300        let mut tui_copy = store.policy(policy_id()).expect("loads").expect("present");
5301        tui_copy.activate().expect("a pending policy activates");
5302        store
5303            .update_policy(&tui_copy, 0)
5304            .expect("the first write wins");
5305        assert_eq!(
5306            store
5307                .policy(policy_id())
5308                .expect("loads")
5309                .expect("present")
5310                .revision(),
5311            1
5312        );
5313
5314        // A CLI invocation that read revision 0 before that write now tries its
5315        // own change.
5316        policy
5317            .set_max_capacity(NonZeroU16::new(5).expect("non-zero"))
5318            .expect("autoscale");
5319        let error = store
5320            .update_policy(&policy, 0)
5321            .expect_err("the second write must be rejected");
5322        assert!(
5323            matches!(
5324                error,
5325                StoreError::StaleRevision {
5326                    expected: 0,
5327                    found: 1,
5328                    ..
5329                }
5330            ),
5331            "expected a stale-revision rejection, got {error:?}"
5332        );
5333        assert!(
5334            error.is_conflict(),
5335            "the caller must be able to tell a lost race from an I/O failure"
5336        );
5337
5338        // And nothing was written: the loser's ceiling is not in the database and
5339        // the winner's change is intact.
5340        let stored = store.policy(policy_id()).expect("loads").expect("present");
5341        assert_eq!(stored.max_capacity().expect("autoscale").get(), 2);
5342        assert!(stored.enabled());
5343
5344        // Re-reading and re-applying succeeds, which is the documented recovery.
5345        let mut fresh = stored;
5346        fresh
5347            .set_max_capacity(NonZeroU16::new(5).expect("non-zero"))
5348            .expect("autoscale");
5349        store.update_policy(&fresh, 1).expect("the retry wins");
5350        assert_eq!(
5351            store
5352                .policy(policy_id())
5353                .expect("loads")
5354                .expect("present")
5355                .max_capacity()
5356                .expect("autoscale")
5357                .get(),
5358            5
5359        );
5360    }
5361
5362    #[test]
5363    fn removing_a_policy_takes_the_same_revision_check() {
5364        let store = store();
5365        RawPolicy {
5366            revision: 4,
5367            ..RawPolicy::default()
5368        }
5369        .insert(&store);
5370
5371        let error = store
5372            .remove_policy(policy_id(), 3)
5373            .expect_err("a stale delete must be rejected");
5374        assert!(error.is_conflict(), "got {error:?}");
5375        assert!(
5376            store.policy(policy_id()).expect("loads").is_some(),
5377            "a rejected delete must not delete"
5378        );
5379
5380        store
5381            .remove_policy(policy_id(), 4)
5382            .expect("the current revision deletes");
5383        assert!(store.policy(policy_id()).expect("loads").is_none());
5384    }
5385
5386    #[test]
5387    fn a_revision_guarded_write_to_a_missing_row_is_not_found_not_a_conflict() {
5388        let store = store();
5389        let error = store
5390            .remove_policy(policy_id(), 0)
5391            .expect_err("there is nothing to delete");
5392        assert!(
5393            matches!(error, StoreError::NotFound { what: "policy", .. }),
5394            "a missing row is a different problem from a lost race: {error:?}"
5395        );
5396        assert!(!error.is_conflict());
5397    }
5398
5399    #[test]
5400    fn inserting_a_policy_twice_is_reported_as_already_existing() {
5401        let store = store();
5402        RawPolicy::default().insert(&store);
5403        let policy = store.policy(policy_id()).expect("loads").expect("present");
5404        let error = store.insert_policy(&policy).expect_err("the id is taken");
5405        assert!(
5406            matches!(error, StoreError::AlreadyExists { what: "policy", .. }),
5407            "got {error:?}"
5408        );
5409    }
5410
5411    // -- the journal --------------------------------------------------------
5412
5413    #[test]
5414    fn created_at_is_never_rewritten_by_a_later_journal_write() {
5415        let store = store();
5416        let allocated = RunnerAttempt::allocate(attempt_id(), policy_id(), "runtime/x", ts(1_000));
5417        store.record_attempt(&allocated).expect("journalled");
5418
5419        // A later write claiming a different allocation instant. `created_at` is
5420        // absent from the upsert's DO UPDATE list precisely so this cannot move
5421        // it; every elapsed-time calculation downstream depends on it.
5422        let rewritten = RunnerAttempt::from_persisted(PersistedAttempt {
5423            created_at: ts(5_000),
5424            last_state_change_at: ts(5_000),
5425            ..allocated.to_persisted()
5426        })
5427        .expect("a legal attempt");
5428        store.record_attempt(&rewritten).expect("journalled");
5429
5430        let stored = store
5431            .attempt(attempt_id())
5432            .expect("loads")
5433            .expect("present");
5434        assert_eq!(stored.created_at, ts(1_000), "created_at never moves");
5435        assert_eq!(
5436            stored.last_state_change_at(),
5437            ts(5_000),
5438            "every other column is updated in place"
5439        );
5440    }
5441
5442    #[test]
5443    fn a_backwards_clock_is_clamped_on_write_and_on_load() {
5444        // The hazard `SqliteStore::normalise` exists for. `move_to` accepts any
5445        // `now`, so a wall clock stepping backwards between two transitions
5446        // builds an in-memory attempt that `from_persisted` refuses -- which
5447        // would leave a journal row nothing can ever load, holding a capacity
5448        // slot and an uncleaned runtime directory for ever.
5449        let store = store();
5450        let mut attempt =
5451            RunnerAttempt::allocate(attempt_id(), policy_id(), "runtime/x", ts(1_000));
5452        attempt
5453            .jit_received(ts(900))
5454            .expect("the domain accepts a backwards `now` with no ordering check");
5455
5456        // The load path really would refuse this, so the repair below is
5457        // load-bearing rather than decorative.
5458        assert!(
5459            matches!(
5460                RunnerAttempt::from_persisted(attempt.to_persisted()),
5461                Err(AttemptError::TimestampsOutOfOrder {
5462                    field: "last_state_change_at",
5463                    ..
5464                })
5465            ),
5466            "if the domain ever accepts this, `normalise` is dead code and should go"
5467        );
5468
5469        store.record_attempt(&attempt).expect("journalled");
5470        assert_eq!(store.clock_skew_repairs(), 1);
5471        {
5472            let conn = store.lock();
5473            let raw: String = conn
5474                .query_row(
5475                    "SELECT last_state_change_at FROM attempts WHERE id = ?1",
5476                    [ATTEMPT_UUID],
5477                    |row| row.get(0),
5478                )
5479                .expect("readable");
5480            assert_eq!(
5481                raw,
5482                timestamp_to_text(ts(1_000)),
5483                "the write path stores a row that can be read back, not one that \
5484                 poisons the journal"
5485            );
5486        }
5487        let back = store
5488            .attempt(attempt_id())
5489            .expect("loads")
5490            .expect("present");
5491        assert_eq!(back.last_state_change_at(), ts(1_000));
5492        assert_eq!(
5493            store.clock_skew_repairs(),
5494            1,
5495            "the stored row is already sound, so loading it repairs nothing"
5496        );
5497
5498        // A row this build did not write -- an older version, or a hand edit --
5499        // is repaired on the way out instead.
5500        RawAttempt {
5501            created_at: timestamp_to_text(ts(1_000)),
5502            last_state_change_at: timestamp_to_text(ts(400)),
5503            state: "finished".to_string(),
5504            outcome: Some(COMPLETED_JOB.to_string()),
5505            terminal_at: Some(timestamp_to_text(ts(500))),
5506            ..RawAttempt::default()
5507        }
5508        .insert(&store);
5509        let repaired = store
5510            .attempt(attempt_id())
5511            .expect("loads")
5512            .expect("present");
5513        assert_eq!(repaired.last_state_change_at(), ts(1_000));
5514        assert_eq!(repaired.terminal_at(), Some(ts(1_000)));
5515        assert_eq!(
5516            store.clock_skew_repairs(),
5517            3,
5518            "both out-of-order timestamps are counted, and each is logged at warn"
5519        );
5520    }
5521
5522    #[test]
5523    fn a_runtime_path_that_is_not_utf8_is_refused_rather_than_mangled() {
5524        // `e3` deletes the directory this path names. A lossy conversion would
5525        // either fail to delete or name a different directory, so the write is
5526        // refused instead.
5527        #[cfg(windows)]
5528        let bad: PathBuf = {
5529            use std::os::windows::ffi::OsStringExt as _;
5530            // An unpaired surrogate: a legal Windows path, not legal UTF-8.
5531            std::ffi::OsString::from_wide(&[0x0072, 0xD800]).into()
5532        };
5533        #[cfg(not(windows))]
5534        let bad: PathBuf = {
5535            use std::os::unix::ffi::OsStringExt as _;
5536            std::ffi::OsString::from_vec(vec![b'r', 0xFF]).into()
5537        };
5538        assert!(bad.to_str().is_none(), "the fixture path must be non-UTF-8");
5539
5540        let store = store();
5541        let attempt = RunnerAttempt::allocate(attempt_id(), policy_id(), bad, ts(1_000));
5542        let error = store
5543            .record_attempt(&attempt)
5544            .expect_err("a path that cannot round-trip must not be stored");
5545        assert!(
5546            matches!(error, StoreError::UnrepresentablePath { .. }),
5547            "got {error:?}"
5548        );
5549    }
5550
5551    #[test]
5552    fn an_integer_too_large_for_a_sqlite_integer_is_refused_rather_than_truncated() {
5553        // SQLite has no unsigned 64-bit type. Saturating at `i64::MAX` would
5554        // store one number and read a different one back with nothing to say so,
5555        // and both `u64` the domain carries -- the GitHub runner id and the
5556        // installation id -- come from GitHub, so this is a path a caller can
5557        // reach rather than a theoretical one.
5558        let store = store();
5559
5560        let mut attempt =
5561            RunnerAttempt::allocate(attempt_id(), policy_id(), "runtime/x", ts(1_000));
5562        attempt.jit_received(ts(1_001)).expect("a legal transition");
5563        attempt
5564            .started(4_242, ts(1_002))
5565            .expect("a legal transition");
5566        attempt
5567            .registered_idle(u64::MAX, ts(1_003))
5568            .expect("the domain accepts any u64 as a runner id");
5569        let error = store
5570            .record_attempt(&attempt)
5571            .expect_err("a value SQLite cannot hold must not be silently truncated");
5572        assert!(
5573            matches!(
5574                error,
5575                StoreError::UnrepresentableInteger {
5576                    what: "attempts.github_runner_id",
5577                    value: u64::MAX,
5578                }
5579            ),
5580            "got {error:?}"
5581        );
5582
5583        // The largest value that does fit still round-trips exactly, so the
5584        // refusal is an edge and not a blanket ceiling.
5585        let biggest = u64::try_from(i64::MAX).expect("i64::MAX is a valid u64");
5586        let mut ok = RunnerAttempt::allocate(
5587            AttemptId::from_u128(0x0000_0101),
5588            policy_id(),
5589            "runtime/y",
5590            ts(1_000),
5591        );
5592        ok.jit_received(ts(1_001)).expect("a legal transition");
5593        ok.started(1, ts(1_002)).expect("a legal transition");
5594        ok.registered_idle(biggest, ts(1_003))
5595            .expect("a legal transition");
5596        store.record_attempt(&ok).expect("journalled");
5597        assert_eq!(
5598            store
5599                .attempt(ok.id)
5600                .expect("loads")
5601                .expect("present")
5602                .github_runner_id(),
5603            Some(biggest)
5604        );
5605
5606        // And the same on the policy side.
5607        let policy = ScalePolicy::new(
5608            policy_id(),
5609            ScaleTarget::repository("o/r").expect("valid"),
5610            u64::MAX,
5611            host_id(),
5612            PolicyMode::monitor_only(),
5613            CachePolicy::default(),
5614        );
5615        let error = store
5616            .insert_policy(&policy)
5617            .expect_err("an installation id SQLite cannot hold must not be truncated");
5618        assert!(
5619            matches!(
5620                error,
5621                StoreError::UnrepresentableInteger {
5622                    what: "policies.installation_id",
5623                    ..
5624                }
5625            ),
5626            "got {error:?}"
5627        );
5628    }
5629
5630    #[test]
5631    fn attempts_are_listed_oldest_first_and_can_be_filtered_by_policy() {
5632        let store = store();
5633        let other_policy = PolicyId::from_u128(0x0000_0011);
5634
5635        let first = RunnerAttempt::allocate(
5636            AttemptId::from_u128(0xA1),
5637            policy_id(),
5638            "runtime/a1",
5639            ts(1_000),
5640        );
5641        let second = RunnerAttempt::allocate(
5642            AttemptId::from_u128(0xA2),
5643            other_policy,
5644            "runtime/a2",
5645            ts(2_000),
5646        );
5647        let third = RunnerAttempt::allocate(
5648            AttemptId::from_u128(0xA3),
5649            policy_id(),
5650            "runtime/a3",
5651            ts(3_000),
5652        );
5653        for attempt in [&third, &first, &second] {
5654            store.record_attempt(attempt).expect("journalled");
5655        }
5656
5657        assert_eq!(
5658            store
5659                .attempts()
5660                .expect("loads")
5661                .iter()
5662                .map(|a| a.created_at)
5663                .collect::<Vec<_>>(),
5664            vec![ts(1_000), ts(2_000), ts(3_000)],
5665            "insertion order must not decide read order"
5666        );
5667
5668        assert_eq!(
5669            store.attempts_for_policy(policy_id()).expect("loads"),
5670            vec![first.clone(), third]
5671        );
5672
5673        assert!(store.remove_attempt(first.id).expect("removable"));
5674        assert!(
5675            !store.remove_attempt(first.id).expect("idempotent"),
5676            "removing an absent attempt is not an error, it is a `false`"
5677        );
5678        assert_eq!(store.attempts().expect("loads").len(), 2);
5679    }
5680
5681    #[test]
5682    fn the_store_is_usable_as_a_shared_trait_object() {
5683        // The agent holds one handle across tasks while the TUI reads through
5684        // it. If this stops compiling, every caller has to change shape.
5685        let concrete = store();
5686        RawHost::default().insert(&concrete);
5687        let store: Arc<dyn Store> = Arc::new(concrete);
5688
5689        let handle = Arc::clone(&store);
5690        let seen = std::thread::spawn(move || handle.host(host_id()).expect("loads").is_some())
5691            .join()
5692            .expect("the reader thread did not panic");
5693        assert!(seen);
5694        assert!(store.policies().expect("loads").is_empty());
5695    }
5696
5697    #[test]
5698    fn the_dump_names_every_table_and_reaches_every_column() {
5699        let store = store();
5700        let root = a_root("rman");
5701        RawHost {
5702            runner_root_override: Some(root.as_str().to_string()),
5703            ..RawHost::default()
5704        }
5705        .insert(&store);
5706        RawPolicy {
5707            workspace_mode: "persistent".to_string(),
5708            workspace_path: Some(root.as_str().to_string()),
5709            ..RawPolicy::default()
5710        }
5711        .insert(&store);
5712        RawAttempt {
5713            workspace_mode: "persistent".to_string(),
5714            workspace_slot: Some(1),
5715            ..RawAttempt::default()
5716        }
5717        .insert(&store);
5718
5719        let dump = store.dump_text().expect("dumpable");
5720        for table in TABLES {
5721            assert!(
5722                dump.contains(&format!("-- table {table}")),
5723                "{table} is missing from the dump"
5724            );
5725        }
5726        assert!(dump.contains(&format!("-- schema version {SCHEMA_VERSION}")));
5727        assert!(dump.contains("hosts.display_name=home-pc"));
5728        assert!(dump.contains("policies.target_slug=o/r"));
5729        assert!(dump.contains("attempts.outcome=NULL"));
5730        assert!(
5731            dump.contains("attempts.runtime_path=runtime/policy/attempt"),
5732            "a dump that omitted a column would make the security scan vacuous"
5733        );
5734        // The schema-3 columns reach the dump too, which is what keeps the
5735        // security scan over a populated database honest about them: a
5736        // configured path is not a credential, and this is where that claim is
5737        // actually checkable rather than merely asserted.
5738        for column in [
5739            format!("hosts.runner_root_override={root}"),
5740            format!("policies.workspace_path={root}"),
5741            "policies.workspace_mode=persistent".to_string(),
5742            "attempts.workspace_mode=persistent".to_string(),
5743            "attempts.workspace_slot=1".to_string(),
5744        ] {
5745            assert!(dump.contains(&column), "{column} is missing from the dump");
5746        }
5747
5748        // A free-form string reaches the dump, which is why `07-security.md`'s
5749        // scan runs over the dump rather than over the schema: the schema alone
5750        // cannot say what a caller put in `FailureReason::Other`.
5751        let reason = FailureReason::Other("no credential here".to_string());
5752        assert!(json(&AttemptOutcome::failed(reason)).contains("no credential here"));
5753    }
5754}