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(¶ms)[..],
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(¶ms)[..],
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(¶ms)[..],
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(¶ms)[..],
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}