diff --git a/docs/design/store-interface.md b/docs/design/store-interface.md new file mode 100644 index 0000000..ef38222 --- /dev/null +++ b/docs/design/store-interface.md @@ -0,0 +1,232 @@ +# The Runner's store behind an interface + +Issue #173. The Runner keeps its durable state in one SQLite database, in the state directory +the boot line names, under an exclusive process lock. The Desk's operational jobs plan +(`docs/design/operational-jobs-plan.md` in judgment-pack-desk) names storage and dispatch as the +seams to define for use beyond a single host, without promising a free migration to distributed +execution. This note puts the queue, releases, inputs and run records behind one contract, with +SQLite as the kind that ships, its statements, transactions, pragmas and lock unchanged (section 8 +names what differs), an in-memory kind for tests, and a conformance suite both kinds run. +It states each operation's semantics as the SQLite code has them; it is not a new design. + +Code: `internal/runner/store_contract.go` (the contract, the faults, the kind registry), +`internal/runner/store_sqlite.go` (the sqlite kind), `internal/runner/store_memory_test.go` (the +in-memory kind) and `internal/runner/store_conformance_test.go` (the suite). + +## 1. What the contract covers + +| Area | Kept by the sqlite kind in | What it is | +|---|---|---| +| The queue | `runs` (`seq`, `state`) | A run's state is its place in the queue: `queued`, `running`, then `completed`, `failed` or `interrupted`. The queue is served by `seq`, the order runs were queued. | +| Inputs | `runs` (`job_id`, `caller`, `idem`, `request_digest`, `record`) | The input snapshot frozen into a run's record at admission, held under the request key (job, caller, idempotency key) with the digest of the request that made it. An occurrence's input, held until its run is queued, is kept with the occurrence, outside the contract (section 6). | +| Releases | `releases` | A release a preview made, kept once and never changed. | +| Run records | `runs` (`record`), `run_chain`, and the `events` entries written with them | A run's record, the installation's chain of runs, and the journal entries the Runner writes through the contract (a run's or a release's change, the Runner's start and stop), each in the transaction of what it records. | +| Jobs, read only | `jobs` | The job a run is queued for. Jobs are made and revised outside the contract (section 6). | + +## 2. Operations + +Every write is made inside `update`, one transaction. The reads in the first table are answered +both outside a transaction, each one statement, and inside one, where they see the transaction's +own writes. + +| Outside or inside a transaction | The sqlite kind's statement | Semantics | +|---|---|---| +| `release(id)` | `SELECT record FROM releases WHERE id=?` | The release, or `absent`. A record that does not decode is `unreadable`. | +| `job(id)` | `SELECT record FROM jobs WHERE id=?` | The job, or `absent`. | +| `requestedRun(job, caller, key)` | `SELECT request_digest,record FROM runs WHERE job_id=? AND caller=? AND idem=?` | The request's digest and the run's record, decoded only when asked, so a caller compares the digest first, as it did; the decoding's error is `decodeRun`'s own. `absent` when nothing is held under the key. | +| `entrySequence(run)` | `SELECT sequence FROM run_chain WHERE run=?` | The sequence of the run's entry, or `absent`. | + +| Outside a transaction | The sqlite kind's statement | Semantics | +|---|---|---| +| `run(id)` | `SELECT record FROM runs WHERE id=?` | The run, or `absent`. | +| `oldestQueued()` | `SELECT record FROM runs WHERE state='queued' ORDER BY seq LIMIT 1` | The run queued first among those still queued, or `absent`. The order is the order of queueing, never the ids'. | +| `entryLine(run)` | `SELECT line FROM run_chain WHERE run=?` | The entry's exact line, or `absent`. | +| `chainPage(after, limit)` | `SELECT sequence,line FROM run_chain WHERE sequence>? ORDER BY sequence LIMIT ?` | Up to `limit` lines after `after`, in order of sequence, and the last sequence read (`after` when none). Entries are only appended, so pages read in turn are a prefix of the chain as it stood at some moment of the reading. | +| `update(fn)` | `BEGIN` ... `COMMIT` | Runs `fn` in one transaction. What `fn` wrote is kept only if `fn` returns nil and the commit succeeds. `fn`'s error is returned as `fn` gave it. Transactions are serializable. | +| `close()` | | Closes the database, then gives the lock up. | + +| In a transaction | The sqlite kind's statement | Semantics | +|---|---|---| +| `activeRuns()` | `SELECT count(*) FROM runs WHERE state IN ('queued','running')` | The runs queued or running. | +| `createRun(r, caller, key, digest)` | `INSERT INTO runs(id,job_id,caller,idem,request_digest,state,record) VALUES (...)` | Queues a run, created once: `taken` when its id or its request key is held. | +| `putRun(r, expected)` | `SELECT state FROM runs WHERE id=?`, then `UPDATE runs SET state=?,record=? WHERE id=?` | Writes the run's state and record, and returns the state it held. With `expected` other than `anyState`, the run must be held in that state, or nothing is written and the fault is `conflict` (section 4). `absent` when no run is held under the id. | +| `runningRuns()` | `SELECT record FROM runs WHERE state='running' ORDER BY seq` | The runs left running, in the order they were queued. | +| `lastEntry()` | `SELECT sequence,line FROM run_chain ORDER BY sequence DESC LIMIT 1` | The chain's last entry, or `absent` when it holds none. | +| `createEntry(sequence, run, line)` | `INSERT INTO run_chain(sequence,run,line) VALUES (?,?,?)` | Appends an entry, created once: `taken` when the sequence or the run has one, `refused` for a sequence below 1. | +| `createRelease(r)` | `INSERT INTO releases(id,record) VALUES (?,?)` | Keeps a release, created once: `taken` when its id is held. | +| `appendJournal(c)` | `INSERT INTO events(...)` (`appendEvent`) | Appends a journal entry and returns its sequence. | + +### The Runner's operations over the contract + +Each is the transaction it was, with the same statements in the same order. + +- **Enqueue** (`submitInternalIntent`). One transaction: the request key; when a run is held under + it, the run is returned if the request's digest is the same (an identical retry), and the + transaction ends keeping nothing (`errKeepNothing`, the rollback it was), or the request is + refused with 409 `idempotency_conflict`. Otherwise the job, its release, the Runner's checks + (the manual-admission binding, the reviewed mapping), `activeRuns` against the limit of 100 + (429 `queue_full`), `createRun`, and the `run.queued` entry. A mapping-v2 request reads the key + once before, outside the transaction, as it did. +- **Claim** (`dispatchNext`, `changeRun`). `oldestQueued` outside a transaction, then one + transaction: `putRun` to `running` with `queued` expected, and the `run.started` entry. A run + that expired in the queue, or whose release names a profile set aside, ends `failed` the same + way, with `queued` expected. +- **Complete, fail, interrupt** (`finishRun`). One transaction: for a completed run, its entry in + the chain (`appendRunEntry`: `lastEntry`, the next entry, `createEntry`) and its sequence + (`entrySequence`); then `putRun` whatever the state held, and the entry of the run's end. A + completion without its record's exact bytes is refused with `errNoAuditBytes` and nothing is + written. +- **Retry.** None. A run is never queued again; a resubmission under the same key returns the + run, and one under a new key is a new run. +- **Restart** (`restart`). One transaction, under the lock: the `runner.started` entry, every run + `runningRuns` reads ended `interrupted` with an entry naming the start, and the occurrences the + sqlite kind keeps outside the contract (section 6), through `sqliteTxOf`. +- **Release** (`saveRelease`). One transaction: `createRelease` and the `release.previewed` entry. +- **Stop** (`recordStop`). The `runner.stopped` entry; a failure is not reported, as before. + +## 3. Faults + +Every failure a contract operation reports is a `storeError` carrying one fault. Its text stays +its cause's, so every message reads as it did, and `errors.Is` reaches the cause: the sqlite +kind's `absent` still answers `errors.Is(err, sql.ErrNoRows)`. `absent(err)` answers for both +kinds and for the SQLite statements outside the contract. + +| Fault | Meaning | sqlite kind | in-memory kind | +|---|---|---|---| +| `absent` | Nothing is held under the id or key. | `sql.ErrNoRows` | Nothing is held. | +| `taken` | A create-once id or key is held. Nothing is written. | `SQLITE_CONSTRAINT_PRIMARYKEY`, `SQLITE_CONSTRAINT_UNIQUE` | The id, key or sequence is held. | +| `conflict` | A versioned update found another state than expected. Nothing is written. | `putRun`'s comparison | The same comparison. | +| `durability` | A change was not made durable: a write statement, `BEGIN` or `COMMIT` failed. Never a success. | Any other failed write, a failed `BEGIN` or `COMMIT` | Its commit seam. | +| `unreadable` | A read failed, or a record does not decode. | Any other failed read; a decoding error | A decoding error. | +| `refused` | An argument outside the contract, refused before anything is written. | `SQLITE_CONSTRAINT_CHECK` (a chain sequence below 1); an unknown kind; `sqliteTxOf` for another kind | A chain sequence below 1. | + +An error the Runner's own function returns inside `update` (an `apiError`, `errNoAuditBytes`, +`errManualProtocol`) is returned unchanged. Opening a store is the kind's own: the sqlite kind's +opening errors (the lock, its files, the schema, the store's identity, the migration) are returned +as they were. + +## 4. Durability, order, exclusivity, versions + +**Durability.** The sqlite kind runs in WAL mode with `synchronous=FULL`: a transaction is on disk +when its `COMMIT` returns, and `update` reports success only after it. A failed statement keeps +nothing of its transaction; a failed `COMMIT` is `durability`, and the caller takes nothing as +kept. By operation: an enqueued run and its entry are durable together before the submission is +answered; a claim's `running` is durable before the Runtime is invoked; a run's end, its chain +entry and its journal entry are durable together, and a failure stops the dispatcher, after which +the run reads interrupted at the next start; a release and its entry are durable together before +the preview is answered; a restart's entries and interruptions are durable together before the +dispatcher starts. The in-memory kind keeps nothing past its process. + +**Order.** The queue and the runs left running are read by `seq`, SQLite's `AUTOINCREMENT` +column: the order of queueing. The chain is read by sequence. The journal's sequence is its own +`AUTOINCREMENT`. + +**Exclusivity.** Transactions are serializable. The sqlite kind holds one connection +(`SetMaxOpenConns(1)`), so one transaction runs at a time in the process, and the exclusive lock +keeps every other process out. The in-memory kind holds its own lock from a transaction's start to +its end and works on a copy, which replaces its state only at the commit. + +**The claim's fence.** A run leaves `queued` once and never returns to it, so its state is the +claim's fence: `putRun` with `queued` expected writes only if the run is still queued, and a +second writer that read it queued meets `conflict` and writes nothing. Before this change the +claim was exclusive because one dispatcher runs per store, under the lock; it still is, and the +comparison, made between the two statements the claim already made, never refuses it. A claim +does not expire, so there is no lease: a run left running is ended by the next process's restart, +whose lock proves the last dispatcher is gone. A kind for use beyond one host would need a lease +with fencing for the claim; this contract has none. + +**Versions.** The only record the contract updates is a run, and its state is its version. +Releases, chain entries and journal entries are created once and never changed. Jobs are revised +outside the contract. + +## 5. The kind registry and the boot line's directory + +`storeKinds` names the kinds a store can be opened as; it holds `sqlite`, which is +`defaultStoreKind`. The boot line names no kind, so every Runner opens `sqlite`, and no other kind +is selectable from the boot line. The in-memory kind is in a test file and is not registered. + +The boot line's directory reaches the store only through the sqlite kind: `openSQLiteStore`, +which takes the lock (`runner.lock`) and opens the database (`runs.sqlite`), and +`sqlitePreparedStore`, which checks that a manual-admission store was prepared. The rest of the +Runner keeps the queue, releases, inputs and run records through the contract. Opening keeps its +order: the lock; the Runner's working directories (`storeOpening.afterLock`); the database with +one connection; the manual-admission store's check (`storeOpening.opened`); the pragmas and the +schema; the store's identity (workspace, owner, input root); the migration. + +The directory is also the Runner's working directory for what is not the store: the files a +Runtime process runs on and writes (`releases/`, `attempts/`, `runtimes/`, +`workflow-attempts/`, `version-check-*`), the manual-admission marker and the executor lease, the +check that a signing key is kept outside it, and the health route's `stateDirectory`. A Runtime +launched as a process needs those as files on its host; they stay where they are. + +## 6. What stays SQLite-specific + +The kind itself: the exclusive process lock (`flock` on `runner.lock`), WAL, `synchronous=FULL`, +the busy timeout, one connection, the schema and its migration, and the store's identity in +`metadata`. + +The state the contract does not cover yet, kept in the same database and reached through the +database handle the kind gives the Runner at opening (and `sqliteTxOf` inside a contract +transaction): triggers and their revisions; occurrences and their keys; job creation and +revision (`jobs` writes, `job_creations`, `job_revisions`); briefs; workflow executions and their +artifacts; the manual-admission journal; `metadata` (beside the store's identity and schema: the profile +registry, the manual-admission demand, the relocation history); the journal's reads, and every +entry written with the state above. + +The statements left on `runs`, `releases` and `jobs` outside the kind, found by searching the +package for those tables: + +- the job and run listings (`list.go`): joins, SQLite's ASCII case folding and a window function; +- an occurrence's run by its request key (`triggers_dispatch.go`) and an occurrence's admission + counting a job's pending runs with its occurrences in one statement (`triggers_store.go`); +- job creation's and revision's checks over job revisions joined to releases + (`job_revisions.go`, `dialect_artifact.go`), and a revision's read and write of its job; +- manual admission (`manual_admission_activation.go`, `manual_execution_binding.go`, + `manual_store_prepare.go`). + +Programs and modes that hold the SQLite files themselves: manual store preparation, its own +process opening `runs.sqlite` through a held directory; offline workspace relocation +(`relocate.go`); and the source worker's own store (`sources.sqlite`). + +## 7. The in-memory kind and the conformance suite + +The in-memory kind keeps the contract's semantics in memory: records as their encoded bytes, +through the same encoding as the sqlite kind; runs in the order queued; a transaction on a copy +under the kind's lock, which replaces the state at the commit; and a commit seam that fails every +commit while set. + +`TestStoreKindsConform` runs every case against both kinds. A case uses only the contract and the +Runner's functions over it (`changeQueuedRun`, `finishRunIn`, `appendRunEntry`); what the +contract does not cover (a job made outside it, the journal's entries, a failed commit, a record +that does not decode) it asks of the kind under test. + +- A release is created once. +- A run is queued once under its id and once under its request key. +- The queue is served in the order runs were queued, with ids that sort against it. +- A run read queued is claimed once; an update expecting another state writes nothing. +- Concurrent claimers grant every run once and lose none, with producers queueing at the same + time; a claimer that meets `conflict` reads the queue again. The claimers share nothing but the + store. +- A transaction that fails keeps nothing, run, entry, release or chain entry. +- A commit that fails is `durability` with its cause's text, never a success, and keeps nothing. +- The chain gives a run one entry and is read in order; a completion without its record's bytes + is not recorded. +- Chain entries are read in order of sequence whatever order they were made in, under runs whose + ids sort against it. +- Nothing held reads `absent`. +- A record that does not read is `unreadable`. +- The sqlite kind only: a reopened store holds what it held (a claimed run still running, a + completed run with its entry, the queue's order, the release, the request, the journal), and one + process holds the store at a time. + +The suite runs in the `unit-tests` step, under `go test -race ./...`, and the step then requires +both kinds' cases to have run and passed. + +## 8. What changed in behaviour + +- A queued run's change (`changeRun`) now requires the store to hold the run queued. With one + dispatcher per store it always does. +- `Close` closes the attempts directory before the store, where it closed it between the database + and the lock. The lock is still given up last. +- Errors from the contract's operations carry a fault; their text and what `errors.Is` reaches are + unchanged. diff --git a/internal/runner/events.go b/internal/runner/events.go index 3f5d6c8..47b3f9b 100644 --- a/internal/runner/events.go +++ b/internal/runner/events.go @@ -322,33 +322,31 @@ func triggerBefore(tx *sql.Tx, id string, revision int) (string, error) { } // recordRun writes a run's record inside tx, and the entry of its change of -// state, if its state changed. -func recordRun(tx *sql.Tx, r Run, by EventActor, detail EventDetail) error { - var from string - if err := tx.QueryRow("SELECT state FROM runs WHERE id=?", r.ID).Scan(&from); err != nil { - return err - } - if _, err := tx.Exec("UPDATE runs SET state=?,record=? WHERE id=?", r.State, runText(r), r.ID); err != nil { +// state, if its state changed. With expected other than anyState, the store +// must hold the run in that state, or nothing is written (faultConflict). +func recordRun(tx runTx, r Run, expected string, by EventActor, detail EventDetail) error { + from, err := tx.putRun(r, expected) + if err != nil { return err } if from == r.State { return nil } - _, err := appendEvent(tx, runChange(r, from, by, detail)) + _, err = tx.appendJournal(runChange(r, from, by, detail)) return err } -// changeRun records a run's change of state and its entry in one transaction. -func (s *Service) changeRun(r Run) error { - tx, err := s.db.Begin() - if err != nil { - return err - } - defer tx.Rollback() - if err = recordRun(tx, r, runnerActor, EventDetail{}); err != nil { - return err - } - return tx.Commit() +// changeRun records a queued run's change of state, the dispatcher's claim +// of it or its end in the queue, and its entry in one transaction. +func (s *Service) changeRun(r Run) error { return changeQueuedRun(s.store, r) } + +// changeQueuedRun is changeRun in st. The store must still hold the run +// queued: a run is claimed, or ended in the queue, once, and a second writer +// that read it queued meets faultConflict. +func changeQueuedRun(st runStore, r Run) error { + return st.update(func(tx runTx) error { + return recordRun(tx, r, "queued", runnerActor, EventDetail{}) + }) } // recordOccurrence writes an occurrence's record inside tx, and the entry of @@ -389,89 +387,42 @@ func (s *Service) changeOccurrence(o Occurrence, guard string, by EventActor) er return tx.Commit() } -// migrateStore begins the journal of a store that has none, a new store or -// one an earlier Runner kept, in one transaction: it writes journal.began, -// with what the store held then and nothing back-filled, adds the column that -// names the revision a trigger's key was issued at, and raises the store's -// schema to storeSchema. -func (s *Service) migrateStore() error { - tx, err := s.db.Begin() - if err != nil { - return err - } - defer tx.Rollback() - var schema string - switch err = tx.QueryRow("SELECT value FROM metadata WHERE key='schema'").Scan(&schema); { - case errors.Is(err, sql.ErrNoRows): - case err != nil: - return err - } - switch schema { - case storeSchema: - return nil - case "", "1", "2": - default: - return errors.New("runner store belongs to another owner, workspace or schema") - } - if schema != "2" { - // SQLite's schema changes are transactional: a migration that fails - // leaves no table, column or entry behind. - if _, err = tx.Exec(eventsSchema); err != nil { +// restart records this process's start, and in the same transaction marks +// what the last process left running, which Runner did not see stop: a run in +// evaluation is interrupted, and a legacy preparation failed. Neither is +// repeated. Each entry gives the window its stop lies in. +func (s *Service) restart() error { + return s.store.update(func(contract runTx) error { + at := now() + started, err := contract.appendJournal(change{kind: "runner.started", at: at, by: runnerActor, detail: EventDetail{Version: buildinfo.Version(), RuntimeDigest: s.runtimeDigest}}) + if err != nil { + return err + } + runs, err := contract.runningRuns() + if err != nil { return err } - var held EventHeld - for table, count := range map[string]*int{"jobs": &held.Jobs, "triggers": &held.Triggers, "occurrences": &held.Occurrences, "runs": &held.Runs} { - if err = tx.QueryRow("SELECT count(*) FROM " + table).Scan(count); err != nil { + for _, r := range runs { + r.State = "interrupted" + r.FinishedAt = at + r.Problem = "The runner stopped during evaluation. Retained attempt files may contain an audit record. This run was not automatically repeated." + if err = recordRun(contract, r, anyState, runnerActor, EventDetail{Interruption: &EventInterruption{LastKnownRunning: r.StartedAt, Restart: started}}); err != nil { return err } } - if _, err = tx.Exec("ALTER TABLE triggers ADD COLUMN key_revision INTEGER"); err != nil { - return err - } - if _, err = appendEvent(tx, change{kind: "journal.began", by: runnerActor, detail: EventDetail{Held: &held}}); err != nil { + // Occurrences are kept outside the contract, by the sqlite kind, and + // recovered in the same transaction. + tx, err := sqliteTxOf(contract) + if err != nil { return err } - } - if _, err = tx.Exec(jobRevisionSchema); err != nil { - return err - } - if _, err = tx.Exec("INSERT OR REPLACE INTO metadata VALUES ('schema',?)", storeSchema); err != nil { - return err - } - return tx.Commit() + return restartOccurrences(tx, started) + }) } -// restart records this process's start, and in the same transaction marks -// what the last process left running, which Runner did not see stop: a run in -// evaluation is interrupted, and a legacy preparation failed. Neither is -// repeated. Each entry gives the window its stop lies in. -func (s *Service) restart() error { - tx, err := s.db.Begin() - if err != nil { - return err - } - defer tx.Rollback() - at := now() - started, err := appendEvent(tx, change{kind: "runner.started", at: at, by: runnerActor, detail: EventDetail{Version: buildinfo.Version(), RuntimeDigest: s.runtimeDigest}}) - if err != nil { - return err - } - var runs []Run - if err = eachRecord(tx, "SELECT record FROM runs WHERE state='running' ORDER BY seq", func(b []byte) error { - r, e := decodeRun(b) - runs = append(runs, r) - return e - }); err != nil { - return err - } - for _, r := range runs { - r.State = "interrupted" - r.FinishedAt = at - r.Problem = "The runner stopped during evaluation. Retained attempt files may contain an audit record. This run was not automatically repeated." - if err = recordRun(tx, r, runnerActor, EventDetail{Interruption: &EventInterruption{LastKnownRunning: r.StartedAt, Restart: started}}); err != nil { - return err - } - } +// restartOccurrences marks, inside tx, each legacy preparation the last +// process left preparing failed, with the window its stop lies in. +func restartOccurrences(tx *sql.Tx, started int64) (err error) { var occurrences []Occurrence if err = eachRecord(tx, "SELECT record FROM occurrences WHERE state='preparing' ORDER BY seq", func(b []byte) error { var o Occurrence @@ -493,7 +444,7 @@ func (s *Service) restart() error { return err } } - return tx.Commit() + return nil } // eachRecord reads the record column of a query's rows, closing them before @@ -519,30 +470,21 @@ func eachRecord(tx *sql.Tx, query string, each func([]byte) error) error { // recordStop records an orderly stop, after the dispatcher, automation and // background work have stopped. A process that is killed writes none. func (s *Service) recordStop() { - tx, err := s.db.Begin() - if err != nil { - return - } - defer tx.Rollback() - if _, err = appendEvent(tx, change{kind: "runner.stopped", by: runnerActor}); err == nil { - tx.Commit() - } + s.store.update(func(tx runTx) error { + _, err := tx.appendJournal(change{kind: "runner.stopped", by: runnerActor}) + return err + }) } // saveRelease stores a release that a preview made, with its entry. func (s *Service) saveRelease(r Release) error { - tx, err := s.db.Begin() - if err != nil { - return err - } - defer tx.Rollback() - if _, err = tx.Exec("INSERT INTO releases(id,record) VALUES (?,?)", r.ID, string(encode(r))); err != nil { - return err - } - if _, err = appendEvent(tx, change{kind: "release.previewed", by: s.installation(), concerns: EventConcerns{Release: r.ID}, detail: EventDetail{PackID: r.PackID, PackVersion: r.PackVersion, PackDigest: r.PackDigest, RuntimeDigest: r.RuntimeDigest, Tests: r.Tests}}); err != nil { + return s.store.update(func(tx runTx) error { + if err := tx.createRelease(r); err != nil { + return err + } + _, err := tx.appendJournal(change{kind: "release.previewed", by: s.installation(), concerns: EventConcerns{Release: r.ID}, detail: EventDetail{PackID: r.PackID, PackVersion: r.PackVersion, PackDigest: r.PackDigest, RuntimeDigest: r.RuntimeDigest, Tests: r.Tests}}) return err - } - return tx.Commit() + }) } // refused journals a refused request to start work: one answered 400, 401, diff --git a/internal/runner/http.go b/internal/runner/http.go index 05485cf..e3aca89 100644 --- a/internal/runner/http.go +++ b/internal/runner/http.go @@ -3,7 +3,6 @@ package runner import ( "bytes" "crypto/subtle" - "database/sql" "encoding/json" "errors" "fmt" @@ -25,7 +24,7 @@ func failure(w http.ResponseWriter, err error) { var e *apiError if errors.As(err, &e) { status, code, message = e.Status, e.Code, e.Message - } else if errors.Is(err, sql.ErrNoRows) { + } else if absent(err) { status, code, message = 404, "not_found", "This job, release or run was not found." } write(w, status, map[string]any{"error": map[string]any{"code": code, "message": message, "retryable": status == 429 || status == 503}}) @@ -117,7 +116,7 @@ func (s *Service) Handler(token string) http.Handler { return } run, err := s.run(r.PathValue("run")) - if errors.Is(err, sql.ErrNoRows) { + if absent(err) { // A run the store no longer holds keeps its entry in the chain, // which still names it: say so, rather than that it never existed. if sequence, chained, e := s.runEntrySequence(r.PathValue("run")); e != nil { diff --git a/internal/runner/run_chain.go b/internal/runner/run_chain.go index 888f80d..4b72b17 100644 --- a/internal/runner/run_chain.go +++ b/internal/runner/run_chain.go @@ -5,7 +5,6 @@ import ( "bytes" "crypto/rand" "crypto/sha256" - "database/sql" "encoding/hex" "encoding/json" "errors" @@ -143,15 +142,13 @@ var errEntryUnreadable = errors.New("the run's entry in the installation's chain // continues the last row's trail and sequence and links to that row's line, or // starts a chain at sequence 1, linked to the SHA-256 of nothing, under a new // identity. -func appendRunEntry(tx *sql.Tx, run string, auditBytes []byte) error { +func appendRunEntry(tx runTx, run string, auditBytes []byte) error { if !oneLine(auditBytes) { return errNoAuditBytes } next := ChainEntry{EntryVersion: ChainEntryVersion, Kind: chainEntryKind, Run: run, AuditDigest: digest(auditBytes)} - var last int64 - var line []byte - switch err := tx.QueryRow("SELECT sequence,line FROM run_chain ORDER BY sequence DESC LIMIT 1").Scan(&last, &line); { - case errors.Is(err, sql.ErrNoRows): + switch last, line, err := tx.lastEntry(); { + case absent(err): next.Trail, next.Sequence, next.Previous = newChainTrail(), 1, digest(nil) case err != nil: return err @@ -162,12 +159,11 @@ func appendRunEntry(tx *sql.Tx, run string, auditBytes []byte) error { } next.Trail, next.Sequence, next.Previous = held.Trail, last+1, digest(line) } - line = encode(next) + line := encode(next) if _, err := ParseChainEntry(line); err != nil { return errEntryUnreadable } - _, err := tx.Exec("INSERT INTO run_chain(sequence,run,line) VALUES (?,?,?)", next.Sequence, run, line) - return err + return tx.createEntry(next.Sequence, run, line) } // finishRun records a run's end. Every completed run is given its entry in the @@ -177,35 +173,33 @@ func appendRunEntry(tx *sql.Tx, run string, auditBytes []byte) error { // failed, is not recorded: the error is returned, and the dispatcher stops. // The run's entry in the journal of job activity (events.go) is written in the // same transaction too, and names the chain entry's sequence. -func (s *Service) finishRun(r Run) error { - tx, err := s.db.Begin() - if err != nil { - return err - } - defer tx.Rollback() - var detail EventDetail - if r.State == "completed" { - if err = appendRunEntry(tx, r.ID, r.AuditBytes); err != nil { - return err +func (s *Service) finishRun(r Run) error { return finishRunIn(s.store, r) } + +// finishRunIn is finishRun in st. +func finishRunIn(st runStore, r Run) error { + return st.update(func(tx runTx) error { + var detail EventDetail + if r.State == "completed" { + if err := appendRunEntry(tx, r.ID, r.AuditBytes); err != nil { + return err + } + sequence, err := tx.entrySequence(r.ID) + if err != nil { + return err + } + detail.ChainSequence = sequence } - if err = tx.QueryRow("SELECT sequence FROM run_chain WHERE run=?", r.ID).Scan(&detail.ChainSequence); err != nil { - return err + if r.State == "interrupted" && r.InterruptedAt != "" { + detail.Interruption = &EventInterruption{Seen: true, At: r.InterruptedAt} } - } - if r.State == "interrupted" && r.InterruptedAt != "" { - detail.Interruption = &EventInterruption{Seen: true, At: r.InterruptedAt} - } - if err = recordRun(tx, r, runnerActor, detail); err != nil { - return err - } - return tx.Commit() + return recordRun(tx, r, anyState, runnerActor, detail) + }) } // runEntry is a run's entry line, or nil when the chain holds none for it. func (s *Service) runEntry(run string) ([]byte, error) { - var line []byte - err := s.db.QueryRow("SELECT line FROM run_chain WHERE run=?", run).Scan(&line) - if errors.Is(err, sql.ErrNoRows) { + line, err := s.store.entryLine(run) + if absent(err) { return nil, nil } return line, err @@ -213,9 +207,8 @@ func (s *Service) runEntry(run string) ([]byte, error) { // runEntrySequence is the sequence of a run's entry, and whether there is one. func (s *Service) runEntrySequence(run string) (int64, bool, error) { - var sequence int64 - err := s.db.QueryRow("SELECT sequence FROM run_chain WHERE run=?", run).Scan(&sequence) - if errors.Is(err, sql.ErrNoRows) { + sequence, err := s.store.entrySequence(run) + if absent(err) { return 0, false, nil } return sequence, err == nil, err @@ -225,21 +218,7 @@ func (s *Service) runEntrySequence(run string) (int64, bool, error) { // sequence of the last. Rows are only ever appended, so pages read in turn are // a prefix of the chain as it stood at some moment during the reading. func (s *Service) chainPage(after int64, limit int) ([][]byte, int64, error) { - rows, err := s.db.Query("SELECT sequence,line FROM run_chain WHERE sequence>? ORDER BY sequence LIMIT ?", after, limit) - if err != nil { - return nil, 0, err - } - defer rows.Close() - var lines [][]byte - last := after - for rows.Next() { - var line []byte - if err = rows.Scan(&last, &line); err != nil { - return nil, 0, err - } - lines = append(lines, line) - } - return lines, last, rows.Err() + return s.store.chainPage(after, limit) } // checkpointOf is the checkpoint of an entry's line. diff --git a/internal/runner/run_chain_test.go b/internal/runner/run_chain_test.go index ec16761..af2ac83 100644 --- a/internal/runner/run_chain_test.go +++ b/internal/runner/run_chain_test.go @@ -287,7 +287,7 @@ func appendEntry(s *Service, run string, record []byte) error { return e } defer tx.Rollback() - if e = appendRunEntry(tx, run, record); e != nil { + if e = appendRunEntry(&sqliteTx{sqliteReads{tx}, tx}, run, record); e != nil { return e } return tx.Commit() @@ -345,7 +345,7 @@ func TestTheChainIsServedWhole(t *testing.T) { } for i := range 2345 { run, record := syntheticRun(i) - if e = appendRunEntry(tx, run, record); e != nil { + if e = appendRunEntry(&sqliteTx{sqliteReads{tx}, tx}, run, record); e != nil { t.Fatal(e) } } diff --git a/internal/runner/store.go b/internal/runner/store.go index a8e93a3..533f159 100644 --- a/internal/runner/store.go +++ b/internal/runner/store.go @@ -8,7 +8,6 @@ import ( "encoding/json" "errors" "fmt" - _ "modernc.org/sqlite" "os" "path/filepath" "sync" @@ -63,9 +62,12 @@ type Service struct { automationDone chan struct{} cfg Config db *sql.DB - lock *os.File runtime, runtimeDigest string + // store keeps the queue, releases, inputs and run records + // (store_contract.go). db is the sqlite kind's database handle, for the + // state the contract does not cover yet (store_sqlite.go). + store runStore // executor is fixed at Open and remains non-nil after Close. executor *ExecutorLease // authority is the immutable original inherited channel/session. It never @@ -166,7 +168,7 @@ func Open(cfg Config) (_ *Service, err error) { } } if cfg.ExecutorAdmission != nil { - if privateFile(filepath.Join(cfg.Dir, "runs.sqlite")) != nil { + if sqlitePreparedStore(cfg.Dir) != nil { return nil, errManualProtocol } } @@ -180,85 +182,41 @@ func Open(cfg Config) (_ *Service, err error) { s := &Service{cloudProblems: map[string]string{}, automationDone: make(chan struct{}), cfg: cfg, executor: executor, authority: authority, expectedKey: expectedKey, wake: make(chan struct{}, 1), done: make(chan struct{}), previewGate: make(chan struct{}, 1)} defer func() { if err != nil { - if s.db != nil { - s.db.Close() - } - if s.lock != nil { - s.lock.Close() + if s.store != nil { + s.store.close() } if s.attempts != nil { s.attempts.Close() } } }() - s.lock, err = instanceLock(filepath.Join(cfg.Dir, "runner.lock")) - if err != nil { - return nil, err - } - for _, n := range []string{"releases", "attempts", "runtimes"} { - if err = privateDir(filepath.Join(cfg.Dir, n)); err != nil { - return nil, err - } - } - if s.attempts, err = openHeldRoot(filepath.Join(cfg.Dir, "attempts")); err != nil { - return nil, err - } - dbpath := filepath.Join(cfg.Dir, "runs.sqlite") - f, e := os.OpenFile(dbpath, os.O_CREATE|os.O_EXCL|os.O_RDWR, 0600) - if e == nil { - f.Close() - } else if !os.IsExist(e) { - return nil, e - } - if err = privateFile(dbpath); err != nil { - return nil, err - } - s.db, err = sql.Open("sqlite", dbpath) - if err != nil { - return nil, err - } - s.db.SetMaxOpenConns(1) - if cfg.ExecutorAdmission != nil { - if err = s.validateManualStore(); err != nil { - return nil, err - } - } - _, err = s.db.Exec(`PRAGMA journal_mode=WAL; PRAGMA synchronous=FULL; PRAGMA busy_timeout=5000; - CREATE TABLE IF NOT EXISTS metadata (key TEXT PRIMARY KEY, value TEXT NOT NULL); - CREATE TABLE IF NOT EXISTS briefs (subject TEXT PRIMARY KEY, record TEXT NOT NULL); - CREATE TABLE IF NOT EXISTS releases (id TEXT PRIMARY KEY, record TEXT NOT NULL); - CREATE TABLE IF NOT EXISTS jobs (seq INTEGER PRIMARY KEY AUTOINCREMENT, id TEXT UNIQUE NOT NULL, release_id TEXT UNIQUE NOT NULL, record TEXT NOT NULL); - CREATE TABLE IF NOT EXISTS runs (seq INTEGER PRIMARY KEY AUTOINCREMENT, id TEXT UNIQUE NOT NULL, job_id TEXT NOT NULL, caller TEXT NOT NULL, idem TEXT NOT NULL, request_digest TEXT NOT NULL, state TEXT NOT NULL, record TEXT NOT NULL, UNIQUE(job_id,caller,idem)); - CREATE INDEX IF NOT EXISTS runs_state ON runs(state,seq);`) + // The boot line's directory reaches the store only through the store + // kind's constructor, which takes the store's lock, then runs afterLock, + // then opens the database, and applies the schema after opened. + s.store, err = openStore(defaultStoreKind, storeOpening{ + dir: cfg.Dir, workspace: cfg.Workspace, owner: cfg.Owner, inputRoot: cfg.InputRoot, + migrate: cfg.ExecutorAdmission == nil, + afterLock: func() error { + for _, n := range []string{"releases", "attempts", "runtimes"} { + if err := privateDir(filepath.Join(cfg.Dir, n)); err != nil { + return err + } + } + var err error + s.attempts, err = openHeldRoot(filepath.Join(cfg.Dir, "attempts")) + return err + }, + opened: func(db *sql.DB) error { + s.db = db + if cfg.ExecutorAdmission != nil { + return s.validateManualStore() + } + return nil + }, + }) if err != nil { return nil, err } - if _, err = s.db.Exec(triggerSchema); err != nil { - return nil, err - } - if _, err = s.db.Exec(runChainSchema); err != nil { - return nil, err - } - for _, k := range []struct{ key, value string }{{"workspace", cfg.Workspace}, {"owner", cfg.Owner}, {"inputRoot", cfg.InputRoot}} { - if _, err = s.db.Exec("INSERT OR IGNORE INTO metadata VALUES (?,?)", k.key, k.value); err != nil { - return nil, err - } - var held string - if err = s.db.QueryRow("SELECT value FROM metadata WHERE key=?", k.key).Scan(&held); err != nil { - return nil, err - } - if held != k.value { - return nil, errors.New("runner store belongs to another owner, workspace or schema") - } - } - // A store an earlier Runner kept, at schema "1", or a new one, begins its - // journal here and is raised to schema "2", which every earlier Runner - // refuses (docs/design/activity-journal.md, section 4). - if cfg.ExecutorAdmission == nil { - if err = s.migrateStore(); err != nil { - return nil, err - } - } if err = s.loadRegisteredProfiles(); err != nil { return nil, err } @@ -313,43 +271,18 @@ func (s *Service) Close() { <-s.automationDone s.background.Wait() s.recordStop() - s.db.Close() if s.attempts != nil { s.attempts.Close() } - s.lock.Close() + s.store.close() if s.executor != nil { s.executor.Close() } }) } -func (s *Service) release(id string) (Release, error) { - var r Release - var b []byte - e := s.db.QueryRow("SELECT record FROM releases WHERE id=?", id).Scan(&b) - if e == nil { - e = json.Unmarshal(b, &r) - } - return r, e -} -func (s *Service) job(id string) (Job, error) { - var j Job - var b []byte - e := s.db.QueryRow("SELECT record FROM jobs WHERE id=?", id).Scan(&b) - if e == nil { - e = json.Unmarshal(b, &j) - } - return j, e -} -func (s *Service) run(id string) (Run, error) { - var r Run - var b []byte - e := s.db.QueryRow("SELECT record FROM runs WHERE id=?", id).Scan(&b) - if e == nil { - r, e = decodeRun(b) - } - return r, e -} +func (s *Service) release(id string) (Release, error) { return s.store.release(id) } +func (s *Service) job(id string) (Job, error) { return s.store.job(id) } +func (s *Service) run(id string) (Run, error) { return s.store.run(id) } func (s *Service) createJob(name, releaseID string) (Job, error) { return s.createJobConfigured(name, releaseID, nil) } @@ -504,16 +437,15 @@ func (s *Service) submitInternalIntent(jobID, key string, input Input, origin *T } requestHash = digest(raw) // Identical retries return their original acceptance, even after freshness expires. - var held, before string - e = s.db.QueryRow("SELECT request_digest,record FROM runs WHERE job_id=? AND caller=? AND idem=?", jobID, caller, key).Scan(&held, &before) + held, e := s.store.requestedRun(jobID, caller, key) if e == nil { - if held != requestHash { + if held.digest != requestHash { return Run{}, false, &apiError{409, "idempotency_conflict", "This idempotency key was already used with different inputs."} } - r, e := decodeRun([]byte(before)) + r, e := held.run() return r, true, e } - if !errors.Is(e, sql.ErrNoRows) { + if !absent(e) { return Run{}, false, e } job, e := s.job(jobID) @@ -547,70 +479,63 @@ func (s *Service) submitInternalIntent(jobID, key string, input Input, origin *T if v2 { hash = requestHash } - tx, err := s.db.Begin() - if err != nil { - return Run{}, false, err - } - defer tx.Rollback() - var held, before string - err = tx.QueryRow("SELECT request_digest,record FROM runs WHERE job_id=? AND caller=? AND idem=?", jobID, caller, key).Scan(&held, &before) - if err == nil { - if held != hash { - return Run{}, false, &apiError{409, "idempotency_conflict", "This idempotency key was already used with different inputs."} + var r Run + var replayed bool + var replayErr error + err = s.store.update(func(tx runTx) error { + held, err := tx.requestedRun(jobID, caller, key) + if err == nil { + if held.digest != hash { + return &apiError{409, "idempotency_conflict", "This idempotency key was already used with different inputs."} + } + r, replayErr = held.run() + replayed = true + return errKeepNothing } - r, err := decodeRun([]byte(before)) - return r, true, err - } - if !errors.Is(err, sql.ErrNoRows) { - return Run{}, false, err - } - var jb []byte - if err = tx.QueryRow("SELECT record FROM jobs WHERE id=?", jobID).Scan(&jb); err != nil { - return Run{}, false, err - } - var j Job - if err = json.Unmarshal(jb, &j); err != nil { - return Run{}, false, err - } - var releaseBytes []byte - if err = tx.QueryRow("SELECT record FROM releases WHERE id=?", j.ReleaseID).Scan(&releaseBytes); err != nil { - return Run{}, false, err - } - var release Release - if err = json.Unmarshal(releaseBytes, &release); err != nil { - return Run{}, false, err - } - if manual != nil { - preimage, e := executionPayloadCanonical(release, maxExecutionRelease) - if e != nil || manual.resource == nil || manual.registration == nil || manual.resource.JobID != j.ID || manual.resource.JobRevision != decimalRevision(j.Revision) || manual.resource.ReleaseID != j.ReleaseID || manual.resource.ReleaseDigest != digest(preimage) || !manualRelease(release) { - return Run{}, false, errManualProtocol + if !absent(err) { + return err } + j, err := tx.job(jobID) + if err != nil { + return err + } + release, err := tx.release(j.ReleaseID) + if err != nil { + return err + } + if manual != nil { + preimage, e := executionPayloadCanonical(release, maxExecutionRelease) + if e != nil || manual.resource == nil || manual.registration == nil || manual.resource.JobID != j.ID || manual.resource.JobRevision != decimalRevision(j.Revision) || manual.resource.ReleaseID != j.ReleaseID || manual.resource.ReleaseDigest != digest(preimage) || !manualRelease(release) { + return errManualProtocol + } + } + if !mappingMatchesRelease(release, input) { + return bad("mapping_mismatch", "Use this job’s reviewed input mapping. Create a new job to change its input source or mapping.") + } + count, err := tx.activeRuns() + if err != nil { + return err + } + if count >= queueLimit { + return &apiError{429, "queue_full", "The local queue is full. Try again after pending runs finish."} + } + r = Run{manual: manual, Trigger: origin, SchemaVersion: "1", ID: id("run_"), JobID: j.ID, ReleaseID: j.ReleaseID, Revision: j.Revision, State: "queued", Input: input, CreatedAt: now(), RequestedBy: caller} + if err = tx.createRun(r, caller, key, hash); err != nil { + return err + } + by := s.installation() + if origin != nil { + by = triggerActor(origin.TriggerID, origin.TriggerRevision) + } + _, err = tx.appendJournal(runChange(r, "", by, EventDetail{})) + return err + }) + if replayed { + return r, true, replayErr } - if !mappingMatchesRelease(release, input) { - return Run{}, false, bad("mapping_mismatch", "Use this job’s reviewed input mapping. Create a new job to change its input source or mapping.") - } - var count int - if err = tx.QueryRow("SELECT count(*) FROM runs WHERE state IN ('queued','running')").Scan(&count); err != nil { - return Run{}, false, err - } - if count >= queueLimit { - return Run{}, false, &apiError{429, "queue_full", "The local queue is full. Try again after pending runs finish."} - } - r := Run{manual: manual, Trigger: origin, SchemaVersion: "1", ID: id("run_"), JobID: j.ID, ReleaseID: j.ReleaseID, Revision: j.Revision, State: "queued", Input: input, CreatedAt: now(), RequestedBy: caller} - _, err = tx.Exec("INSERT INTO runs(id,job_id,caller,idem,request_digest,state,record) VALUES (?,?,?,?,?,?,?)", r.ID, j.ID, caller, key, hash, r.State, runText(r)) if err != nil { return Run{}, false, err } - by := s.installation() - if origin != nil { - by = triggerActor(origin.TriggerID, origin.TriggerRevision) - } - if _, err = appendEvent(tx, runChange(r, "", by, EventDetail{})); err != nil { - return Run{}, false, err - } - if err = tx.Commit(); err != nil { - return Run{}, false, err - } select { case s.wake <- struct{}{}: default: @@ -649,18 +574,13 @@ func (s *Service) worker(ctx context.Context) { // end, and says whether there was one. An error is a storage error, which // stops the dispatcher. func (s *Service) dispatchNext(ctx context.Context) (bool, error) { - var b []byte - err := s.db.QueryRow("SELECT record FROM runs WHERE state='queued' ORDER BY seq LIMIT 1").Scan(&b) - if errors.Is(err, sql.ErrNoRows) { + r, err := s.store.oldestQueued() + if absent(err) { return false, nil } if err != nil { return false, err } - r, err := decodeRun(b) - if err != nil { - return false, err - } if r.manual != nil && s.cfg.ExecutorAdmission == nil || r.manual == nil && s.cfg.ExecutorAdmission != nil { return false, errManualInactive } diff --git a/internal/runner/store_conformance_test.go b/internal/runner/store_conformance_test.go new file mode 100644 index 0000000..81f4fe7 --- /dev/null +++ b/internal/runner/store_conformance_test.go @@ -0,0 +1,715 @@ +package runner + +// The store contract's conformance suite (docs/design/store-interface.md), +// run against every kind: the sqlite kind the Runner ships and the in-memory +// kind. A case uses only the contract and the Runner's own functions over it +// (changeQueuedRun, finishRunIn, appendRunEntry); what it needs that the +// contract does not cover, a job made outside it, the journal's entries, a +// commit made to fail and a record that does not read, it asks of the kind +// under test. The sqlite kind's own cases, a restart and the lock, follow. + +import ( + "database/sql" + "errors" + "fmt" + "os" + "path/filepath" + "slices" + "sync" + "sync/atomic" + "testing" + "time" +) + +// storeKindUnderTest is a kind and what the suite asks of it beside the +// contract. +type storeKindUnderTest struct { + name string + // open is a new, empty store of the kind. + open func(t *testing.T) runStore + // seedJob keeps a job, made outside the contract. + seedJob func(t *testing.T, st runStore, j Job) + // journal is the kinds of the journal's entries the contract wrote, in + // order. + journal func(t *testing.T, st runStore) []string + // failCommits makes every later commit fail with err, or succeed again + // for nil. + failCommits func(st runStore, err error) + // spoil makes a run's record not read as one. + spoil func(t *testing.T, st runStore, run string) + // own are the kind's own cases. + own []storeCase +} + +type storeCase struct { + name string + run func(t *testing.T, k storeKindUnderTest) +} + +// privateStateDir is a private directory, by its physical path, as a state +// directory must be. +func privateStateDir(t *testing.T) string { + t.Helper() + dir, err := filepath.EvalSymlinks(t.TempDir()) + if err != nil { + t.Fatal(err) + } + if err = os.Chmod(dir, 0700); err != nil { + t.Fatal(err) + } + return dir +} + +func openSQLiteForTest(t *testing.T, dir string) *sqliteStore { + t.Helper() + k, err := openSQLiteStore(storeOpening{dir: dir, workspace: "conformance", owner: "owner", migrate: true}) + if err != nil { + t.Fatal(err) + } + return k +} + +func storeKindsUnderTest() []storeKindUnderTest { + return []storeKindUnderTest{ + { + name: "sqlite", + open: func(t *testing.T) runStore { + k := openSQLiteForTest(t, privateStateDir(t)) + t.Cleanup(func() { k.close() }) + return k + }, + seedJob: func(t *testing.T, st runStore, j Job) { + // The statement job creation makes (store.go). + if _, err := st.(*sqliteStore).db.Exec("INSERT INTO jobs(id,release_id,record)VALUES(?,?,?)", j.ID, j.ReleaseID, string(encode(j))); err != nil { + t.Fatal(err) + } + }, + journal: func(t *testing.T, st runStore) []string { + rows, err := st.(*sqliteStore).db.Query("SELECT kind FROM events WHERE kind<>'journal.began' ORDER BY sequence") + if err != nil { + t.Fatal(err) + } + defer rows.Close() + var kinds []string + for rows.Next() { + var kind string + if err = rows.Scan(&kind); err != nil { + t.Fatal(err) + } + kinds = append(kinds, kind) + } + if err = rows.Err(); err != nil { + t.Fatal(err) + } + return kinds + }, + failCommits: func(st runStore, err error) { + k := st.(*sqliteStore) + if err == nil { + k.commit = (*sql.Tx).Commit + return + } + k.commit = func(*sql.Tx) error { return err } + }, + spoil: func(t *testing.T, st runStore, run string) { + if _, err := st.(*sqliteStore).db.Exec("UPDATE runs SET record='{' WHERE id=?", run); err != nil { + t.Fatal(err) + } + }, + own: sqliteOwnCases, + }, + { + name: "memory", + open: func(t *testing.T) runStore { return newMemoryStore() }, + seedJob: func(t *testing.T, st runStore, j Job) { st.(*memoryStore).seedJob(j) }, + journal: func(t *testing.T, st runStore) []string { return st.(*memoryStore).journalKinds() }, + failCommits: func(st runStore, err error) { st.(*memoryStore).failCommits(err) }, + spoil: func(t *testing.T, st runStore, run string) { st.(*memoryStore).spoilRun(run) }, + }, + } +} + +// TestStoreKindsConform runs the conformance suite against every kind. +func TestStoreKindsConform(t *testing.T) { + for _, k := range storeKindsUnderTest() { + t.Run(k.name, func(t *testing.T) { + for _, c := range append(slices.Clone(storeCases), k.own...) { + t.Run(c.name, func(t *testing.T) { c.run(t, k) }) + } + }) + } +} + +// storeWorld is a store with a release and a job to queue runs for. +type storeWorld struct { + st runStore + release Release + job Job +} + +func newStoreWorld(t *testing.T, k storeKindUnderTest) storeWorld { + t.Helper() + w := storeWorld{st: k.open(t)} + w.release = Release{SchemaVersion: "1", ID: "rel_conformance", PackID: "conformance", PackVersion: "1.0.0", Title: "Conformance", CreatedAt: now(), Tests: "passed"} + must(t, w.st.update(func(tx runTx) error { return tx.createRelease(w.release) })) + w.job = Job{SchemaVersion: "1", ID: "job_conformance", Name: "Conformance", ReleaseID: w.release.ID, Revision: 1, CreatedAt: now(), Workspace: "conformance", Owner: "owner"} + k.seedJob(t, w.st, w.job) + return w +} + +// queuedRun is a run of the world's job, as a submission makes one. +func (w storeWorld) queuedRun(id string) Run { + return Run{SchemaVersion: "1", ID: id, JobID: w.job.ID, ReleaseID: w.job.ReleaseID, Revision: w.job.Revision, State: "queued", Input: journalInput(), CreatedAt: now(), RequestedBy: "owner"} +} + +// enqueue queues a run under key in one transaction, as a submission does: +// the run, then its entry. +func (w storeWorld) enqueue(id, key string) (Run, error) { + r := w.queuedRun(id) + return r, w.st.update(func(tx runTx) error { + if err := tx.createRun(r, "owner", key, "sha256:"+key); err != nil { + return err + } + _, err := tx.appendJournal(runChange(r, "", runnerActor, EventDetail{})) + return err + }) +} + +// claim claims the oldest queued run as the dispatcher does, and retries +// when another claimer took the run it read: it returns the run it claimed, +// or false when none is queued. A claim that conflicts without end is an +// error: each conflict is another claimer's success, and they are bounded. +func claim(st runStore) (Run, bool, error) { + for range 10000 { + r, err := st.oldestQueued() + if absent(err) { + return Run{}, false, nil + } + if err != nil { + return Run{}, false, err + } + r.State, r.StartedAt, r.Attempt = "running", now(), 1 + switch err = changeQueuedRun(st, r); faultOf(err) { + case "": + return r, err == nil, err + case faultConflict: + continue + default: + return Run{}, false, err + } + } + return Run{}, false, errors.New("a claim conflicted ten thousand times") +} + +func wantFault(t *testing.T, err error, want storeFault, what string) { + t.Helper() + if got := faultOf(err); got != want { + t.Fatalf("%s: fault %q (%v), want %q", what, got, err, want) + } +} + +func kindCount(kinds []string, kind string) int { + n := 0 + for _, k := range kinds { + if k == kind { + n++ + } + } + return n +} + +var storeCases = []storeCase{ + {"a release is created once", func(t *testing.T, k storeKindUnderTest) { + w := newStoreWorld(t, k) + second := w.release + second.Title = "Another" + wantFault(t, w.st.update(func(tx runTx) error { return tx.createRelease(second) }), faultTaken, "a second release under one id") + held, err := w.st.release(w.release.ID) + if err != nil || held.Title != w.release.Title || held.PackID != w.release.PackID { + t.Fatal("the release held is not the first:", held.Title, err) + } + _, err = w.st.release("rel_none") + wantFault(t, err, faultAbsent, "a release never made") + if !absent(err) { + t.Fatal("absent does not say a release never made is absent") + } + }}, + {"a run is queued once under its id and once under its request key", func(t *testing.T, k storeKindUnderTest) { + w := newStoreWorld(t, k) + first, err := w.enqueue("run_a", "key-a") + must(t, err) + _, err = w.enqueue("run_a", "key-b") + wantFault(t, err, faultTaken, "a second run under one id") + _, err = w.enqueue("run_b", "key-a") + wantFault(t, err, faultTaken, "a second run under one request key") + held, err := w.st.requestedRun(w.job.ID, "owner", "key-a") + must(t, err) + if held.digest != "sha256:key-a" { + t.Fatal("the request's digest is not the first's:", held.digest) + } + if r, err := held.run(); err != nil || r.ID != first.ID { + t.Fatal("the run held under the key is not the first:", r.ID, err) + } + for _, other := range [][3]string{{w.job.ID, "owner", "key-b"}, {w.job.ID, "trigger:t", "key-a"}, {"job_other", "owner", "key-a"}} { + _, err = w.st.requestedRun(other[0], other[1], other[2]) + wantFault(t, err, faultAbsent, fmt.Sprint("another request key ", other)) + } + if got := kindCount(k.journal(t, w.st), "run.queued"); got != 1 { + t.Fatal("run.queued entries:", got) + } + }}, + {"the queue is served in the order runs were queued", func(t *testing.T, k storeKindUnderTest) { + w := newStoreWorld(t, k) + // Ids that sort against the order of queueing: the order is the + // queue's, never the ids'. + var order []string + for i := range 5 { + id := fmt.Sprintf("run_%d", 9-i) + _, err := w.enqueue(id, id) + must(t, err) + order = append(order, id) + } + must(t, w.st.update(func(tx runTx) error { + active, err := tx.activeRuns() + if err != nil || active != 5 { + t.Fatal("active runs:", active, err) + } + return nil + })) + var claimed []string + for range order { + r, ok, err := claim(w.st) + if err != nil || !ok { + t.Fatal("no run claimed:", ok, err) + } + claimed = append(claimed, r.ID) + } + if !slices.Equal(claimed, order) { + t.Fatal("claimed in the order", claimed, "not", order) + } + _, err := w.st.oldestQueued() + wantFault(t, err, faultAbsent, "an empty queue") + // The runs left running are read in the order they were queued, as a + // restart reads them, after one ended. + done := w.queuedRun(order[2]) + done.State = "failed" + must(t, finishRunIn(w.st, done)) + must(t, w.st.update(func(tx runTx) error { + running, err := tx.runningRuns() + if err != nil { + return err + } + var ids []string + for _, r := range running { + ids = append(ids, r.ID) + } + if want := slices.Delete(slices.Clone(order), 2, 3); !slices.Equal(ids, want) { + t.Fatal("running runs read in the order", ids, "not", want) + } + if active, err := tx.activeRuns(); err != nil || active != 4 { + t.Fatal("active runs:", active, err) + } + return nil + })) + }}, + {"a run read queued is claimed once", func(t *testing.T, k storeKindUnderTest) { + w := newStoreWorld(t, k) + _, err := w.enqueue("run_a", "key-a") + must(t, err) + first, err := w.st.oldestQueued() + must(t, err) + second, err := w.st.oldestQueued() + must(t, err) + first.State, second.State = "running", "running" + must(t, changeQueuedRun(w.st, first)) + second.Problem = "a second claim" + wantFault(t, changeQueuedRun(w.st, second), faultConflict, "a second claim of a run read queued") + held, err := w.st.run("run_a") + if err != nil || held.State != "running" || held.Problem != "" { + t.Fatal("the second claim wrote:", held.State, held.Problem, err) + } + // A versioned update at a state the run is not in writes nothing. + ended := held + ended.State, ended.Problem = "failed", "ended in the queue" + wantFault(t, changeQueuedRun(w.st, ended), faultConflict, "a running run ended as queued") + if held, err = w.st.run("run_a"); err != nil || held.State != "running" || held.Problem != "" { + t.Fatal("a conflicting update wrote:", held.State, held.Problem, err) + } + kinds := k.journal(t, w.st) + if kindCount(kinds, "run.started") != 1 || kindCount(kinds, "run.expired") != 0 { + t.Fatal("entries for one claim:", kinds) + } + }}, + {"concurrent claimers grant every run once and lose none", func(t *testing.T, k storeKindUnderTest) { + w := newStoreWorld(t, k) + const producers, perProducer, claimers = 4, 10, 6 + var produced atomic.Bool + var wg sync.WaitGroup + for p := range producers { + wg.Add(1) + go func() { + defer wg.Done() + for i := range perProducer { + id := fmt.Sprintf("run_%d_%d", p, i) + if _, err := w.enqueue(id, id); err != nil { + t.Error(err) + } + } + }() + } + // The claimers share nothing but the store: each keeps what it + // claimed, and they are compared after. + grants := make([][]string, claimers) + var cw sync.WaitGroup + for c := range claimers { + cw.Add(1) + go func() { + defer cw.Done() + for { + r, ok, err := claim(w.st) + if err != nil { + t.Error(err) + return + } + if ok { + grants[c] = append(grants[c], r.ID) + continue + } + if produced.Load() { + // Produced and nothing queued: confirm once more. + if _, err = w.st.oldestQueued(); absent(err) { + return + } + continue + } + time.Sleep(time.Millisecond) + } + }() + } + wg.Wait() + produced.Store(true) + cw.Wait() + seen := map[string]int{} + for _, g := range grants { + for _, id := range g { + seen[id]++ + } + } + if len(seen) != producers*perProducer { + t.Fatal("runs claimed:", len(seen), "of", producers*perProducer) + } + for id, n := range seen { + if n != 1 { + t.Fatal(id, "was granted", n, "times") + } + if held, err := w.st.run(id); err != nil || held.State != "running" { + t.Fatal(id, held.State, err) + } + } + kinds := k.journal(t, w.st) + if kindCount(kinds, "run.queued") != producers*perProducer || kindCount(kinds, "run.started") != producers*perProducer { + t.Fatal("entries:", kindCount(kinds, "run.queued"), kindCount(kinds, "run.started")) + } + }}, + {"a transaction that fails keeps nothing", func(t *testing.T, k storeKindUnderTest) { + w := newStoreWorld(t, k) + before := k.journal(t, w.st) + abandoned := errors.New("the transaction is abandoned") + for _, end := range []error{abandoned, errKeepNothing} { + err := w.st.update(func(tx runTx) error { + r := w.queuedRun("run_kept") + if err := tx.createRun(r, "owner", "key", "sha256:key"); err != nil { + return err + } + if _, err := tx.appendJournal(runChange(r, "", runnerActor, EventDetail{})); err != nil { + return err + } + // The transaction sees its own writes. + if _, err := tx.requestedRun(w.job.ID, "owner", "key"); err != nil { + return err + } + release := w.release + release.ID = "rel_kept" + if err := tx.createRelease(release); err != nil { + return err + } + if err := tx.createEntry(1, r.ID, []byte("{}")); err != nil { + return err + } + return end + }) + if err != end { + t.Fatal("update did not return its function's error as it was:", err) + } + if _, err = w.st.run("run_kept"); !absent(err) { + t.Fatal("a run was kept:", err) + } + if _, err = w.st.release("rel_kept"); !absent(err) { + t.Fatal("a release was kept:", err) + } + if _, err = w.st.entrySequence("run_kept"); !absent(err) { + t.Fatal("an entry was kept:", err) + } + if after := k.journal(t, w.st); !slices.Equal(after, before) { + t.Fatal("journal entries were kept:", after) + } + } + // What one failed transaction gave back is given again. + _, err := w.enqueue("run_kept", "key") + must(t, err) + }}, + {"a commit that fails is an error, never a success", func(t *testing.T, k storeKindUnderTest) { + w := newStoreWorld(t, k) + _, err := w.enqueue("run_a", "key-a") + must(t, err) + before := k.journal(t, w.st) + failed := errors.New("the disk refused the commit") + k.failCommits(w.st, failed) + _, err = w.enqueue("run_b", "key-b") + wantFault(t, err, faultDurability, "a queued run whose commit failed") + if !errors.Is(err, failed) || err.Error() != failed.Error() { + t.Fatal("the failure is not its cause's:", err) + } + r, err := w.st.oldestQueued() + must(t, err) + r.State = "running" + wantFault(t, changeQueuedRun(w.st, r), faultDurability, "a claim whose commit failed") + release := w.release + release.ID = "rel_b" + wantFault(t, w.st.update(func(tx runTx) error { return tx.createRelease(release) }), faultDurability, "a release whose commit failed") + if _, err = w.st.run("run_b"); !absent(err) { + t.Fatal("a run whose commit failed was kept:", err) + } + if held, err := w.st.run("run_a"); err != nil || held.State != "queued" { + t.Fatal("a claim whose commit failed was kept:", held.State, err) + } + if _, err = w.st.release("rel_b"); !absent(err) { + t.Fatal("a release whose commit failed was kept:", err) + } + if after := k.journal(t, w.st); !slices.Equal(after, before) { + t.Fatal("journal entries whose commit failed were kept:", after) + } + k.failCommits(w.st, nil) + _, err = w.enqueue("run_b", "key-b") + must(t, err) + must(t, changeQueuedRun(w.st, r)) + }}, + {"the chain gives a run one entry and is read in order", func(t *testing.T, k storeKindUnderTest) { + w := newStoreWorld(t, k) + var runs []string + records := map[string][]byte{} + for i := range 5 { + run, record := syntheticRun(i) + runs, records[run] = append(runs, run), record + must(t, w.st.update(func(tx runTx) error { return appendRunEntry(tx, run, record) })) + } + wantFault(t, w.st.update(func(tx runTx) error { return appendRunEntry(tx, runs[0], records[runs[0]]) }), faultTaken, "a second entry for a run") + wantFault(t, w.st.update(func(tx runTx) error { return tx.createEntry(3, "run_other", []byte("{}")) }), faultTaken, "a second entry at a sequence") + wantFault(t, w.st.update(func(tx runTx) error { return tx.createEntry(0, "run_other", []byte("{}")) }), faultRefused, "an entry before the first") + var lines [][]byte + for after := int64(0); ; { + page, last, err := w.st.chainPage(after, 2) + must(t, err) + if len(page) == 0 { + if last != after { + t.Fatal("an empty page moved the cursor:", last) + } + break + } + if len(page) > 2 || last != after+int64(len(page)) { + t.Fatal("a page of", len(page), "ending at", last, "after", after) + } + lines, after = append(lines, page...), last + } + if len(lines) != len(runs) { + t.Fatal("lines:", len(lines)) + } + for i, run := range runs { + e, err := ParseChainEntry(lines[i]) + if err != nil || e.Run != run || e.Sequence != int64(i+1) || e.AuditDigest != digest(records[run]) { + t.Fatal("entry", i+1, "is", e.Run, e.Sequence, err) + } + if i > 0 && e.Previous != digest(lines[i-1]) { + t.Fatal("entry", i+1, "does not link to the one before") + } + line, err := w.st.entryLine(run) + if err != nil || string(line) != string(lines[i]) { + t.Fatal("the line of", run, "is not the one read in order:", err) + } + if sequence, err := w.st.entrySequence(run); err != nil || sequence != int64(i+1) { + t.Fatal("the sequence of", run, "is", sequence, err) + } + } + // A completed run is recorded with its entry, or not at all. + _, err := w.enqueue("run_done", "key-done") + must(t, err) + r, ok, err := claim(w.st) + if err != nil || !ok { + t.Fatal(ok, err) + } + r.State = "completed" + if err = finishRunIn(w.st, r); err != errNoAuditBytes { + t.Fatal("a completion without its record's bytes:", err) + } + if held, err := w.st.run(r.ID); err != nil || held.State != "running" { + t.Fatal("a completion without its entry was recorded:", held.State, err) + } + r.AuditBytes = []byte(`{"recordVersion":"1"}`) + must(t, finishRunIn(w.st, r)) + if sequence, err := w.st.entrySequence(r.ID); err != nil || sequence != int64(len(runs)+1) { + t.Fatal("the completed run's entry is at", sequence, err) + } + if held, err := w.st.run(r.ID); err != nil || held.State != "completed" { + t.Fatal(held.State, err) + } + if got := kindCount(k.journal(t, w.st), "run.completed"); got != 1 { + t.Fatal("run.completed entries:", got) + } + }}, + {"chain entries are read in order of sequence, whatever order they were made in", func(t *testing.T, k storeKindUnderTest) { + st := k.open(t) + // Made last-first, under runs whose names sort against the order. + for _, e := range []struct { + sequence int64 + run string + }{{9, "run_a"}, {2, "run_c"}, {4, "run_b"}} { + must(t, st.update(func(tx runTx) error { + return tx.createEntry(e.sequence, e.run, fmt.Appendf(nil, "%d", e.sequence)) + })) + } + must(t, st.update(func(tx runTx) error { + last, line, err := tx.lastEntry() + if err != nil || last != 9 || string(line) != "9" { + t.Fatal("the last entry is", last, string(line), err) + } + return nil + })) + page, last, err := st.chainPage(2, -1) + if err != nil || last != 9 || len(page) != 2 || string(page[0]) != "4" || string(page[1]) != "9" { + t.Fatalf("the entries after 2 read %q to %d: %v", page, last, err) + } + if page, last, err = st.chainPage(0, 0); err != nil || len(page) != 0 || last != 0 { + t.Fatalf("a page of none read %q to %d: %v", page, last, err) + } + }}, + {"nothing held reads absent", func(t *testing.T, k storeKindUnderTest) { + st := k.open(t) + _, err := st.run("run_none") + wantFault(t, err, faultAbsent, "run") + _, err = st.job("job_none") + wantFault(t, err, faultAbsent, "job") + _, err = st.release("rel_none") + wantFault(t, err, faultAbsent, "release") + _, err = st.requestedRun("job_none", "owner", "key") + wantFault(t, err, faultAbsent, "requested run") + _, err = st.oldestQueued() + wantFault(t, err, faultAbsent, "oldest queued") + _, err = st.entrySequence("run_none") + wantFault(t, err, faultAbsent, "entry sequence") + _, err = st.entryLine("run_none") + wantFault(t, err, faultAbsent, "entry line") + if page, last, err := st.chainPage(0, 10); err != nil || len(page) != 0 || last != 0 { + t.Fatal("an empty chain's page:", len(page), last, err) + } + wantFault(t, st.update(func(tx runTx) error { + _, err := tx.putRun(Run{ID: "run_none", State: "running"}, anyState) + return err + }), faultAbsent, "a run never queued, changed") + wantFault(t, st.update(func(tx runTx) error { + _, _, err := tx.lastEntry() + return err + }), faultAbsent, "an empty chain's last entry") + }}, + {"a record that does not read is unreadable", func(t *testing.T, k storeKindUnderTest) { + w := newStoreWorld(t, k) + _, err := w.enqueue("run_a", "key-a") + must(t, err) + k.spoil(t, w.st, "run_a") + _, err = w.st.run("run_a") + wantFault(t, err, faultUnreadable, "a spoiled run") + _, err = w.st.oldestQueued() + wantFault(t, err, faultUnreadable, "a spoiled run at the head of the queue") + held, err := w.st.requestedRun(w.job.ID, "owner", "key-a") + must(t, err) + if _, err = held.run(); err == nil { + t.Fatal("a spoiled run read under its request key") + } + }}, +} + +// The sqlite kind's own cases: what it keeps across a restart, and the lock +// that holds the store for one process. +var sqliteOwnCases = []storeCase{ + {"a reopened store holds what it held", func(t *testing.T, k storeKindUnderTest) { + dir := privateStateDir(t) + st := openSQLiteForTest(t, dir) + w := storeWorld{st: st} + w.release = Release{SchemaVersion: "1", ID: "rel_restart", PackID: "restart", Title: "Restart", Tests: "passed"} + must(t, st.update(func(tx runTx) error { return tx.createRelease(w.release) })) + w.job = Job{SchemaVersion: "1", ID: "job_restart", ReleaseID: w.release.ID, Revision: 1} + k.seedJob(t, st, w.job) + for _, id := range []string{"run_c", "run_b", "run_a"} { + _, err := w.enqueue(id, id) + must(t, err) + } + claimed, ok, err := claim(st) + if err != nil || !ok || claimed.ID != "run_c" { + t.Fatal(claimed.ID, ok, err) + } + done, ok, err := claim(st) + if err != nil || !ok || done.ID != "run_b" { + t.Fatal(done.ID, ok, err) + } + done.State, done.AuditBytes = "completed", []byte(`{"recordVersion":"1"}`) + must(t, finishRunIn(st, done)) + line, err := st.entryLine(done.ID) + must(t, err) + journal := k.journal(t, st) + must(t, st.close()) + + st = openSQLiteForTest(t, dir) + defer st.close() + if held, err := st.run(claimed.ID); err != nil || held.State != "running" || held.StartedAt != claimed.StartedAt { + t.Fatal("the claimed run is not where it was:", held.State, err) + } + if held, err := st.run(done.ID); err != nil || held.State != "completed" { + t.Fatal("the completed run is not where it was:", held.State, err) + } + if next, err := st.oldestQueued(); err != nil || next.ID != "run_a" { + t.Fatal("the queue is not where it was:", next.ID, err) + } + if again, err := st.entryLine(done.ID); err != nil || string(again) != string(line) { + t.Fatal("the chain is not where it was:", err) + } + if held, err := st.release(w.release.ID); err != nil || held.Title != w.release.Title { + t.Fatal("the release is not where it was:", err) + } + if held, err := st.requestedRun(w.job.ID, "owner", "run_a"); err != nil || held.digest != "sha256:run_a" { + t.Fatal("the request is not where it was:", err) + } + if after := k.journal(t, st); !slices.Equal(after, journal) { + t.Fatal("the journal is not where it was:", after, journal) + } + must(t, st.update(func(tx runTx) error { + running, err := tx.runningRuns() + if err != nil || len(running) != 1 || running[0].ID != claimed.ID { + t.Fatal("runs left running:", len(running), err) + } + return nil + })) + }}, + {"one process holds the store", func(t *testing.T, k storeKindUnderTest) { + dir := privateStateDir(t) + first := openSQLiteForTest(t, dir) + if second, err := openSQLiteStore(storeOpening{dir: dir, workspace: "conformance", owner: "owner", migrate: true}); err == nil { + second.close() + first.close() + t.Fatal("a second opening took a store another holds") + } else if err.Error() != "another runner owns this workspace" { + t.Fatal(err) + } + must(t, first.close()) + again := openSQLiteForTest(t, dir) + must(t, again.close()) + }}, +} diff --git a/internal/runner/store_contract.go b/internal/runner/store_contract.go new file mode 100644 index 0000000..4c2568a --- /dev/null +++ b/internal/runner/store_contract.go @@ -0,0 +1,224 @@ +package runner + +// The store contract (docs/design/store-interface.md): the operations the +// Runner keeps its queue, releases, inputs and run records through, whatever +// kind of storage keeps them, and the faults a kind reports. The SQLite kind +// (store_sqlite.go) is the one that ships; every kind answers to the +// conformance suite (store_conformance_test.go), and an in-memory kind +// (store_memory_test.go) runs it beside SQLite. Nothing here names a path: +// where a kind keeps its state is the kind's. +// +// What a run, a release or a journal entry is, and when one may change, is +// the Runner's and not this file's: a kind keeps records and answers the +// questions the Runner asks of them, in one transaction when the Runner asks +// for one. + +import ( + "database/sql" + "errors" + "fmt" +) + +// storeFault is the kind of failure a storage operation met. A caller +// branches on it; the error's text stays its cause's own, so a message the +// Runner gave before a kind stood behind it reads as it did. +type storeFault string + +const ( + // faultAbsent: no record is held under the id or key asked for. + faultAbsent storeFault = "absent" + // faultTaken: a create-once id or key is held already. Nothing is + // written. + faultTaken storeFault = "taken" + // faultConflict: a versioned update found the record at another version + // than the one its writer read. Nothing is written. + faultConflict storeFault = "conflict" + // faultDurability: a change was not made durable. A statement of the + // change or its commit failed; the caller never takes it for a success. + // A failed statement keeps nothing of the transaction; a failed commit + // may or may not have kept it. + faultDurability storeFault = "durability" + // faultUnreadable: a read failed, or what is held does not read as the + // record it should be. + faultUnreadable storeFault = "unreadable" + // faultRefused: an argument outside the contract, refused before + // anything is written. + faultRefused storeFault = "refused" +) + +// storeError is every failure a kind reports. +type storeError struct { + fault storeFault + op string + err error +} + +func (e *storeError) Error() string { + if e.err != nil { + return e.err.Error() + } + return fmt.Sprintf("store: %s: %s", e.op, e.fault) +} + +func (e *storeError) Unwrap() error { return e.err } + +// storeFailure is a fault met by an operation, with its cause. +func storeFailure(fault storeFault, op string, err error) error { + return &storeError{fault: fault, op: op, err: err} +} + +// faultOf is the fault an error carries, or "" for an error no kind made. +func faultOf(err error) storeFault { + var e *storeError + if errors.As(err, &e) { + return e.fault + } + return "" +} + +// absent says whether err says nothing is held: a contract read's +// faultAbsent, or the no-rows error of a SQLite statement the Runner makes +// outside the contract (store_sqlite.go names that state). +func absent(err error) bool { + return faultOf(err) == faultAbsent || errors.Is(err, sql.ErrNoRows) +} + +// errKeepNothing ends a transaction without keeping anything, when its +// function found what it needed and has nothing to write: update returns it, +// as it returns every error of the function's, and the caller expects it. +var errKeepNothing = errors.New("store: the transaction keeps nothing") + +// anyState is putRun's expected state when a writer changes a run whatever +// state the store holds it in. +const anyState = "" + +// heldRequest is a run as the queue holds it under its request key: the +// digest of the request that made it, and its record, read when asked, so +// that a caller compares the digest before it reads the record. +type heldRequest struct { + digest string + record []byte +} + +// run reads the held record. +func (h heldRequest) run() (Run, error) { return decodeRun(h.record) } + +// storeReads are the reads a store answers both outside a transaction, each +// one statement as it stands, and inside one, where it sees the +// transaction's own writes. +type storeReads interface { + // release is the release held under id, or faultAbsent. + release(id string) (Release, error) + // job is the job held under id, or faultAbsent. Jobs are made and + // revised outside the contract (store_sqlite.go). + job(id string) (Job, error) + // requestedRun is the run held under its request key, the job, the + // caller and the caller's idempotency key, or faultAbsent. + requestedRun(jobID, caller, key string) (heldRequest, error) + // entrySequence is the sequence of a run's entry in the installation's + // chain of runs, or faultAbsent. + entrySequence(run string) (int64, error) +} + +// runStore is the Runner's durable state for the queue, releases, inputs and +// run records, behind one contract. +type runStore interface { + storeReads + // update runs fn in one transaction and keeps what it wrote only if fn + // returns nil and the commit succeeds. Transactions are serializable: one + // sees every write of each that committed before it began, and none of + // one still open. fn's error is returned as fn gave it, and nothing fn + // wrote is kept; a commit that fails is faultDurability. fn must not call + // the store outside its transaction: a kind may wait for the transaction + // to end first. + update(fn func(runTx) error) error + // run is the run held under id, or faultAbsent. + run(id string) (Run, error) + // oldestQueued is the queued run queued first, or faultAbsent when none + // is queued: the queue is served in the order runs were queued. + oldestQueued() (Run, error) + // entryLine is the exact line of a run's entry in the chain, or + // faultAbsent. + entryLine(run string) ([]byte, error) + // chainPage is up to limit entry lines with a sequence after after, in + // order of sequence, and the sequence of the last (after when none). + chainPage(after int64, limit int) ([][]byte, int64, error) + // close gives the store up. A kind that holds the store exclusively + // releases it last. + close() error +} + +// runTx is one transaction of a store. +type runTx interface { + storeReads + // activeRuns counts the runs queued or running. + activeRuns() (int, error) + // createRun queues r under its id and its request key (r.JobID, caller, + // key), with the request's digest: faultTaken when either is held. + createRun(r Run, caller, key, requestDigest string) error + // putRun writes r's state and record over the run held under r.ID, and + // returns the state it held before. With expected other than anyState, + // the run must be held in that state, or nothing is written and the + // fault is faultConflict: the versioned update by which a queued run is + // claimed once. faultAbsent when no run is held under r.ID. + putRun(r Run, expected string) (string, error) + // runningRuns are the runs held running, in the order they were queued. + runningRuns() ([]Run, error) + // lastEntry is the sequence and line of the chain's last entry, or + // faultAbsent when the chain holds none. + lastEntry() (int64, []byte, error) + // createEntry appends a run's entry to the chain at sequence: faultTaken + // when the sequence or the run has an entry, faultRefused for a sequence + // below 1. + createEntry(sequence int64, run string, line []byte) error + // createRelease keeps r under its id: faultTaken when one is held. + createRelease(r Release) error + // appendJournal appends a change's entry to the journal of job activity, + // and returns its sequence; entries are only ever appended. + appendJournal(c change) (int64, error) +} + +// storeOpening is what the Runner gives a kind's constructor. +type storeOpening struct { + // dir is the state directory the boot line names. Only the sqlite kind + // reads it. + dir string + // workspace, owner and inputRoot are the store's identity: a store + // another workspace, owner or input root holds is refused. + workspace, owner, inputRoot string + // migrate raises an earlier Runner's store to this Runner's schema and + // begins its journal. A manual-admission store is never migrated. + migrate bool + // afterLock runs once the kind holds the store exclusively and before + // it opens its database: the Runner prepares its working directories + // there, in the order it always has. + afterLock func() error + // opened runs on the sqlite kind's database handle before its schema is + // applied: the Runner keeps the handle for the state outside the + // contract, and checks a manual-admission store there. + opened func(*sql.DB) error +} + +// defaultStoreKind is the kind every Runner opens. The boot line names no +// kind, so no other is selectable yet. +const defaultStoreKind = "sqlite" + +// storeKinds are the kinds a store can be opened as, by name. +var storeKinds = map[string]func(storeOpening) (runStore, error){ + "sqlite": func(o storeOpening) (runStore, error) { + k, err := openSQLiteStore(o) + if err != nil { + return nil, err + } + return k, nil + }, +} + +// openStore opens a store of the named kind. +func openStore(kind string, o storeOpening) (runStore, error) { + open, ok := storeKinds[kind] + if !ok { + return nil, storeFailure(faultRefused, "open", fmt.Errorf("no store kind %q", kind)) + } + return open(o) +} diff --git a/internal/runner/store_memory_test.go b/internal/runner/store_memory_test.go new file mode 100644 index 0000000..fcbb4d9 --- /dev/null +++ b/internal/runner/store_memory_test.go @@ -0,0 +1,345 @@ +package runner + +// The in-memory kind, for tests (docs/design/store-interface.md): the store +// contract's semantics held in memory, run beside the sqlite kind by the +// conformance suite. It keeps records as their encoded bytes, as the sqlite +// kind keeps them as text, so a record is written and read through the same +// encoding. A transaction works on a copy of the state, under the kind's own +// lock held from its start to its end, and the copy replaces the state only +// when the transaction commits: transactions are serializable, and one that +// fails keeps nothing. + +import ( + "encoding/json" + "errors" + "maps" + "slices" + "sync" +) + +// errNotHeld is the cause of the in-memory kind's faultAbsent. +var errNotHeld = errors.New("nothing is held under that id or key") + +type memoryRun struct { + seq int64 + id, job, caller, idem, digest, state string + record []byte +} + +type memoryEntry struct { + sequence int64 + run string + line []byte +} + +// memoryState is everything the kind holds. Byte slices are never changed +// in place, so a copy shares them. +type memoryState struct { + releases map[string][]byte + jobs map[string][]byte + // runs are in the order they were queued; lastSeq is the last order + // given, which a transaction that fails gives back. + runs []memoryRun + lastSeq int64 + // chain is in order of sequence. + chain []memoryEntry + journal []change +} + +func (m memoryState) clone() memoryState { + return memoryState{releases: maps.Clone(m.releases), jobs: maps.Clone(m.jobs), runs: slices.Clone(m.runs), lastSeq: m.lastSeq, chain: slices.Clone(m.chain), journal: slices.Clone(m.journal)} +} + +// memoryStore is the in-memory kind. +type memoryStore struct { + mu sync.RWMutex + state memoryState + // failCommit, when set, is the cause every commit fails with: the + // kind's durability seam. + failCommit error +} + +func newMemoryStore() *memoryStore { + return &memoryStore{state: memoryState{releases: map[string][]byte{}, jobs: map[string][]byte{}}} +} + +// seedJob keeps a job, as one made outside the contract. +func (m *memoryStore) seedJob(j Job) { + m.mu.Lock() + defer m.mu.Unlock() + m.state.jobs[j.ID] = encode(j) +} + +// failCommits makes every later commit fail with err, or succeed again for +// nil. +func (m *memoryStore) failCommits(err error) { + m.mu.Lock() + defer m.mu.Unlock() + m.failCommit = err +} + +// journalKinds are the kinds of the journal's entries, in order. +func (m *memoryStore) journalKinds() []string { + m.mu.RLock() + defer m.mu.RUnlock() + var kinds []string + for _, c := range m.state.journal { + kinds = append(kinds, c.kind) + } + return kinds +} + +// spoilRun replaces a run's record with bytes that do not read as one. +func (m *memoryStore) spoilRun(id string) { + m.mu.Lock() + defer m.mu.Unlock() + for i := range m.state.runs { + if m.state.runs[i].id == id { + m.state.runs[i].record = []byte("{") + } + } +} + +func (m *memoryStore) update(fn func(runTx) error) error { + m.mu.Lock() + defer m.mu.Unlock() + next := m.state.clone() + if err := fn(&memoryTx{memoryReads{&next}}); err != nil { + return err + } + if m.failCommit != nil { + return storeFailure(faultDurability, "commit", m.failCommit) + } + m.state = next + return nil +} + +func (m *memoryStore) close() error { return nil } + +func (m *memoryStore) reads() memoryReads { return memoryReads{&m.state} } + +func (m *memoryStore) release(id string) (Release, error) { + m.mu.RLock() + defer m.mu.RUnlock() + return m.reads().release(id) +} + +func (m *memoryStore) job(id string) (Job, error) { + m.mu.RLock() + defer m.mu.RUnlock() + return m.reads().job(id) +} + +func (m *memoryStore) requestedRun(jobID, caller, key string) (heldRequest, error) { + m.mu.RLock() + defer m.mu.RUnlock() + return m.reads().requestedRun(jobID, caller, key) +} + +func (m *memoryStore) entrySequence(run string) (int64, error) { + m.mu.RLock() + defer m.mu.RUnlock() + return m.reads().entrySequence(run) +} + +func (m *memoryStore) run(id string) (Run, error) { + m.mu.RLock() + defer m.mu.RUnlock() + for _, r := range m.state.runs { + if r.id == id { + return decodeHeld("run", r.record) + } + } + return Run{}, storeFailure(faultAbsent, "run", errNotHeld) +} + +func (m *memoryStore) oldestQueued() (Run, error) { + m.mu.RLock() + defer m.mu.RUnlock() + for _, r := range m.state.runs { + if r.state == "queued" { + return decodeHeld("queue", r.record) + } + } + return Run{}, storeFailure(faultAbsent, "queue", errNotHeld) +} + +func (m *memoryStore) entryLine(run string) ([]byte, error) { + m.mu.RLock() + defer m.mu.RUnlock() + for _, e := range m.state.chain { + if e.run == run { + return e.line, nil + } + } + return nil, storeFailure(faultAbsent, "chain", errNotHeld) +} + +// chainPage reads as SQLite's LIMIT does: a negative limit reads every +// entry, and a limit of zero reads none. +func (m *memoryStore) chainPage(after int64, limit int) ([][]byte, int64, error) { + m.mu.RLock() + defer m.mu.RUnlock() + var lines [][]byte + last := after + for _, e := range m.state.chain { + if limit >= 0 && len(lines) == limit { + break + } + if e.sequence > after { + lines = append(lines, e.line) + last = e.sequence + } + } + return lines, last, nil +} + +func decodeHeld(op string, record []byte) (Run, error) { + r, err := decodeRun(record) + if err != nil { + return r, storeFailure(faultUnreadable, op, err) + } + return r, nil +} + +// memoryReads are the reads over one state: the kind's, under its read +// lock, or a transaction's copy. +type memoryReads struct{ s *memoryState } + +func (r memoryReads) release(id string) (Release, error) { + var rel Release + b, ok := r.s.releases[id] + if !ok { + return rel, storeFailure(faultAbsent, "release", errNotHeld) + } + if err := json.Unmarshal(b, &rel); err != nil { + return rel, storeFailure(faultUnreadable, "release", err) + } + return rel, nil +} + +func (r memoryReads) job(id string) (Job, error) { + var j Job + b, ok := r.s.jobs[id] + if !ok { + return j, storeFailure(faultAbsent, "job", errNotHeld) + } + if err := json.Unmarshal(b, &j); err != nil { + return j, storeFailure(faultUnreadable, "job", err) + } + return j, nil +} + +func (r memoryReads) requestedRun(jobID, caller, key string) (heldRequest, error) { + for _, held := range r.s.runs { + if held.job == jobID && held.caller == caller && held.idem == key { + return heldRequest{digest: held.digest, record: held.record}, nil + } + } + return heldRequest{}, storeFailure(faultAbsent, "run", errNotHeld) +} + +func (r memoryReads) entrySequence(run string) (int64, error) { + for _, e := range r.s.chain { + if e.run == run { + return e.sequence, nil + } + } + return 0, storeFailure(faultAbsent, "chain", errNotHeld) +} + +// memoryTx is one transaction of the in-memory kind, over its copy. +type memoryTx struct{ memoryReads } + +func (t *memoryTx) activeRuns() (int, error) { + count := 0 + for _, r := range t.s.runs { + if r.state == "queued" || r.state == "running" { + count++ + } + } + return count, nil +} + +func (t *memoryTx) createRun(r Run, caller, key, requestDigest string) error { + for _, held := range t.s.runs { + if held.id == r.ID || held.job == r.JobID && held.caller == caller && held.idem == key { + return storeFailure(faultTaken, "run", errors.New("a run is held under that id or request key")) + } + } + t.s.lastSeq++ + t.s.runs = append(t.s.runs, memoryRun{seq: t.s.lastSeq, id: r.ID, job: r.JobID, caller: caller, idem: key, digest: requestDigest, state: r.State, record: []byte(runText(r))}) + return nil +} + +func (t *memoryTx) putRun(r Run, expected string) (string, error) { + for i := range t.s.runs { + held := &t.s.runs[i] + if held.id != r.ID { + continue + } + from := held.state + if expected != anyState && from != expected { + return from, storeFailure(faultConflict, "run", errors.New("the run is held in another state")) + } + held.state, held.record = r.State, []byte(runText(r)) + return from, nil + } + return "", storeFailure(faultAbsent, "run", errNotHeld) +} + +func (t *memoryTx) runningRuns() ([]Run, error) { + var runs []Run + for _, held := range t.s.runs { + if held.state != "running" { + continue + } + r, err := decodeHeld("queue", held.record) + runs = append(runs, r) + if err != nil { + return runs, err + } + } + return runs, nil +} + +func (t *memoryTx) lastEntry() (int64, []byte, error) { + if len(t.s.chain) == 0 { + return 0, nil, storeFailure(faultAbsent, "chain", errNotHeld) + } + last := t.s.chain[len(t.s.chain)-1] + return last.sequence, last.line, nil +} + +func (t *memoryTx) createEntry(sequence int64, run string, line []byte) error { + if sequence < 1 { + return storeFailure(faultRefused, "chain", errors.New("an entry's sequence is 1 or more")) + } + at := len(t.s.chain) + for i, e := range t.s.chain { + if e.sequence == sequence || e.run == run { + return storeFailure(faultTaken, "chain", errors.New("the chain has an entry at that sequence or for that run")) + } + if e.sequence > sequence && at == len(t.s.chain) { + at = i + } + } + t.s.chain = slices.Insert(t.s.chain, at, memoryEntry{sequence: sequence, run: run, line: slices.Clone(line)}) + return nil +} + +func (t *memoryTx) createRelease(r Release) error { + if _, ok := t.s.releases[r.ID]; ok { + return storeFailure(faultTaken, "release", errors.New("a release is held under that id")) + } + t.s.releases[r.ID] = encode(r) + return nil +} + +func (t *memoryTx) appendJournal(c change) (int64, error) { + if c.at == "" { + c.at = now() + } + t.s.journal = append(t.s.journal, c) + return int64(len(t.s.journal)), nil +} diff --git a/internal/runner/store_sqlite.go b/internal/runner/store_sqlite.go new file mode 100644 index 0000000..64d1fb3 --- /dev/null +++ b/internal/runner/store_sqlite.go @@ -0,0 +1,426 @@ +package runner + +// The sqlite kind (docs/design/store-interface.md): the store the Runner +// ships with, one SQLite database in the state directory the boot line names, +// in WAL mode with synchronous FULL, held by one process at a time through an +// exclusive lock on a file beside it. Its statements, transactions and +// pragmas are the ones the Runner made before the contract stood in front of +// them, moved here. +// +// The kind also keeps the state the contract does not cover yet, in the same +// database: jobs as they are made and revised, triggers, occurrences, briefs, +// workflow executions, the manual-admission journal, the profile registry and +// the store's metadata, and the journal's own reads. The Runner reaches that +// state through the database handle the kind gives it at opening +// (storeOpening.opened), and, inside a contract transaction, through +// sqliteTxOf. + +import ( + "database/sql" + "encoding/json" + "errors" + "fmt" + "os" + "path/filepath" + + sqlite "modernc.org/sqlite" + sqlite3 "modernc.org/sqlite/lib" +) + +const ( + // sqliteStoreFile is the kind's database, in the state directory. + sqliteStoreFile = "runs.sqlite" + // sqliteLockFile is the file whose exclusive lock holds the store for + // one process. + sqliteLockFile = "runner.lock" +) + +// sqliteQuerier is what a read is made through: the database, or a +// transaction. +type sqliteQuerier interface { + QueryRow(query string, args ...any) *sql.Row + Query(query string, args ...any) (*sql.Rows, error) + Exec(query string, args ...any) (sql.Result, error) +} + +// sqliteReads are the reads both outside and inside a transaction: the same +// statement, made through the database or through the transaction. +type sqliteReads struct{ q sqliteQuerier } + +// sqliteStore is the sqlite kind. +type sqliteStore struct { + sqliteReads + db *sql.DB + lock *os.File + // commit makes a transaction durable: (*sql.Tx).Commit, which a test + // replaces to make a commit fail. + commit func(*sql.Tx) error +} + +// sqliteTx is one transaction of the sqlite kind. +type sqliteTx struct { + sqliteReads + tx *sql.Tx +} + +// sqliteFault types an error a statement of op returned: nothing held, a +// key held, a check refused, and otherwise a write that was not made or a +// read that failed. +func sqliteFault(op string, err error, write bool) error { + if err == nil { + return nil + } + if errors.Is(err, sql.ErrNoRows) { + return storeFailure(faultAbsent, op, err) + } + var e *sqlite.Error + if errors.As(err, &e) { + switch e.Code() { + case sqlite3.SQLITE_CONSTRAINT_PRIMARYKEY, sqlite3.SQLITE_CONSTRAINT_UNIQUE: + return storeFailure(faultTaken, op, err) + case sqlite3.SQLITE_CONSTRAINT_CHECK: + return storeFailure(faultRefused, op, err) + } + } + if write { + return storeFailure(faultDurability, op, err) + } + return storeFailure(faultUnreadable, op, err) +} + +// sqlitePreparedStore checks that a manual-admission store was prepared +// before the Runner starts on it: its database is there, private and the +// current user's. +func sqlitePreparedStore(dir string) error { + return privateFile(filepath.Join(dir, sqliteStoreFile)) +} + +// openSQLiteStore is the sqlite kind's constructor, and the one place the +// boot line's directory reaches the store. It takes the lock, then runs the +// Runner's afterLock, opens the database with one connection, gives it to +// the Runner's opened, applies the schema, holds the store to its identity +// and, when asked, migrates it. On a failure it gives up what it took. +func openSQLiteStore(o storeOpening) (_ *sqliteStore, err error) { + k := &sqliteStore{commit: (*sql.Tx).Commit} + defer func() { + if err != nil { + if k.db != nil { + k.db.Close() + } + if k.lock != nil { + k.lock.Close() + } + } + }() + k.lock, err = instanceLock(filepath.Join(o.dir, sqliteLockFile)) + if err != nil { + return nil, err + } + if o.afterLock != nil { + if err = o.afterLock(); err != nil { + return nil, err + } + } + dbpath := filepath.Join(o.dir, sqliteStoreFile) + f, e := os.OpenFile(dbpath, os.O_CREATE|os.O_EXCL|os.O_RDWR, 0600) + if e == nil { + f.Close() + } else if !os.IsExist(e) { + return nil, e + } + if err = privateFile(dbpath); err != nil { + return nil, err + } + k.db, err = sql.Open("sqlite", dbpath) + if err != nil { + return nil, err + } + k.db.SetMaxOpenConns(1) + if o.opened != nil { + if err = o.opened(k.db); err != nil { + return nil, err + } + } + _, err = k.db.Exec(`PRAGMA journal_mode=WAL; PRAGMA synchronous=FULL; PRAGMA busy_timeout=5000; + CREATE TABLE IF NOT EXISTS metadata (key TEXT PRIMARY KEY, value TEXT NOT NULL); + CREATE TABLE IF NOT EXISTS briefs (subject TEXT PRIMARY KEY, record TEXT NOT NULL); + CREATE TABLE IF NOT EXISTS releases (id TEXT PRIMARY KEY, record TEXT NOT NULL); + CREATE TABLE IF NOT EXISTS jobs (seq INTEGER PRIMARY KEY AUTOINCREMENT, id TEXT UNIQUE NOT NULL, release_id TEXT UNIQUE NOT NULL, record TEXT NOT NULL); + CREATE TABLE IF NOT EXISTS runs (seq INTEGER PRIMARY KEY AUTOINCREMENT, id TEXT UNIQUE NOT NULL, job_id TEXT NOT NULL, caller TEXT NOT NULL, idem TEXT NOT NULL, request_digest TEXT NOT NULL, state TEXT NOT NULL, record TEXT NOT NULL, UNIQUE(job_id,caller,idem)); + CREATE INDEX IF NOT EXISTS runs_state ON runs(state,seq);`) + if err != nil { + return nil, err + } + if _, err = k.db.Exec(triggerSchema); err != nil { + return nil, err + } + if _, err = k.db.Exec(runChainSchema); err != nil { + return nil, err + } + for _, m := range []struct{ key, value string }{{"workspace", o.workspace}, {"owner", o.owner}, {"inputRoot", o.inputRoot}} { + if _, err = k.db.Exec("INSERT OR IGNORE INTO metadata VALUES (?,?)", m.key, m.value); err != nil { + return nil, err + } + var held string + if err = k.db.QueryRow("SELECT value FROM metadata WHERE key=?", m.key).Scan(&held); err != nil { + return nil, err + } + if held != m.value { + return nil, errors.New("runner store belongs to another owner, workspace or schema") + } + } + // A store an earlier Runner kept, at schema "1", or a new one, begins its + // journal here and is raised to schema "2", which every earlier Runner + // refuses (docs/design/activity-journal.md, section 4). + if o.migrate { + if err = k.migrate(); err != nil { + return nil, err + } + } + k.sqliteReads = sqliteReads{k.db} + return k, nil +} + +// migrate begins the journal of a store that has none, a new store or one an +// earlier Runner kept, in one transaction: it writes journal.began, with what +// the store held then and nothing back-filled, adds the column that names the +// revision a trigger's key was issued at, and raises the store's schema to +// storeSchema. +func (k *sqliteStore) migrate() error { + tx, err := k.db.Begin() + if err != nil { + return err + } + defer tx.Rollback() + var schema string + switch err = tx.QueryRow("SELECT value FROM metadata WHERE key='schema'").Scan(&schema); { + case errors.Is(err, sql.ErrNoRows): + case err != nil: + return err + } + switch schema { + case storeSchema: + return nil + case "", "1", "2": + default: + return errors.New("runner store belongs to another owner, workspace or schema") + } + if schema != "2" { + // SQLite's schema changes are transactional: a migration that fails + // leaves no table, column or entry behind. + if _, err = tx.Exec(eventsSchema); err != nil { + return err + } + var held EventHeld + for table, count := range map[string]*int{"jobs": &held.Jobs, "triggers": &held.Triggers, "occurrences": &held.Occurrences, "runs": &held.Runs} { + if err = tx.QueryRow("SELECT count(*) FROM " + table).Scan(count); err != nil { + return err + } + } + if _, err = tx.Exec("ALTER TABLE triggers ADD COLUMN key_revision INTEGER"); err != nil { + return err + } + if _, err = appendEvent(tx, change{kind: "journal.began", by: runnerActor, detail: EventDetail{Held: &held}}); err != nil { + return err + } + } + if _, err = tx.Exec(jobRevisionSchema); err != nil { + return err + } + if _, err = tx.Exec("INSERT OR REPLACE INTO metadata VALUES ('schema',?)", storeSchema); err != nil { + return err + } + return tx.Commit() +} + +// close closes the database, then gives the lock up. +func (k *sqliteStore) close() error { + err := k.db.Close() + if e := k.lock.Close(); err == nil { + err = e + } + return err +} + +func (k *sqliteStore) update(fn func(runTx) error) error { + tx, err := k.db.Begin() + if err != nil { + return storeFailure(faultDurability, "begin", err) + } + defer tx.Rollback() + if err = fn(&sqliteTx{sqliteReads{tx}, tx}); err != nil { + return err + } + if err = k.commit(tx); err != nil { + return storeFailure(faultDurability, "commit", err) + } + return nil +} + +// sqliteTxOf is the SQL transaction under a sqlite kind's transaction, for +// the state outside the contract that the Runner changes in the same +// transaction as the contract's (a restart's occurrences). +func sqliteTxOf(tx runTx) (*sql.Tx, error) { + if t, ok := tx.(*sqliteTx); ok { + return t.tx, nil + } + return nil, storeFailure(faultRefused, "sqlite", errors.New("the state outside the store contract is kept only by the sqlite kind")) +} + +func (r sqliteReads) release(id string) (Release, error) { + var rel Release + var b []byte + if err := r.q.QueryRow("SELECT record FROM releases WHERE id=?", id).Scan(&b); err != nil { + return rel, sqliteFault("release", err, false) + } + if err := json.Unmarshal(b, &rel); err != nil { + return rel, storeFailure(faultUnreadable, "release", err) + } + return rel, nil +} + +func (r sqliteReads) job(id string) (Job, error) { + var j Job + var b []byte + if err := r.q.QueryRow("SELECT record FROM jobs WHERE id=?", id).Scan(&b); err != nil { + return j, sqliteFault("job", err, false) + } + if err := json.Unmarshal(b, &j); err != nil { + return j, storeFailure(faultUnreadable, "job", err) + } + return j, nil +} + +func (r sqliteReads) requestedRun(jobID, caller, key string) (heldRequest, error) { + var held heldRequest + var record string + if err := r.q.QueryRow("SELECT request_digest,record FROM runs WHERE job_id=? AND caller=? AND idem=?", jobID, caller, key).Scan(&held.digest, &record); err != nil { + return heldRequest{}, sqliteFault("run", err, false) + } + held.record = []byte(record) + return held, nil +} + +func (r sqliteReads) entrySequence(run string) (int64, error) { + var sequence int64 + if err := r.q.QueryRow("SELECT sequence FROM run_chain WHERE run=?", run).Scan(&sequence); err != nil { + return 0, sqliteFault("chain", err, false) + } + return sequence, nil +} + +func (k *sqliteStore) run(id string) (Run, error) { + var b []byte + if err := k.db.QueryRow("SELECT record FROM runs WHERE id=?", id).Scan(&b); err != nil { + return Run{}, sqliteFault("run", err, false) + } + r, err := decodeRun(b) + if err != nil { + return r, storeFailure(faultUnreadable, "run", err) + } + return r, nil +} + +func (k *sqliteStore) oldestQueued() (Run, error) { + var b []byte + if err := k.db.QueryRow("SELECT record FROM runs WHERE state='queued' ORDER BY seq LIMIT 1").Scan(&b); err != nil { + return Run{}, sqliteFault("queue", err, false) + } + r, err := decodeRun(b) + if err != nil { + return r, storeFailure(faultUnreadable, "queue", err) + } + return r, nil +} + +func (k *sqliteStore) entryLine(run string) ([]byte, error) { + var line []byte + if err := k.db.QueryRow("SELECT line FROM run_chain WHERE run=?", run).Scan(&line); err != nil { + return nil, sqliteFault("chain", err, false) + } + return line, nil +} + +// chainPage reads the chain's rows in order. Rows are only ever appended, so +// pages read in turn are a prefix of the chain as it stood at some moment +// during the reading. +func (k *sqliteStore) chainPage(after int64, limit int) ([][]byte, int64, error) { + rows, err := k.db.Query("SELECT sequence,line FROM run_chain WHERE sequence>? ORDER BY sequence LIMIT ?", after, limit) + if err != nil { + return nil, 0, sqliteFault("chain", err, false) + } + defer rows.Close() + var lines [][]byte + last := after + for rows.Next() { + var line []byte + if err = rows.Scan(&last, &line); err != nil { + return nil, 0, sqliteFault("chain", err, false) + } + lines = append(lines, line) + } + return lines, last, sqliteFault("chain", rows.Err(), false) +} + +func (t *sqliteTx) activeRuns() (int, error) { + var count int + if err := t.tx.QueryRow("SELECT count(*) FROM runs WHERE state IN ('queued','running')").Scan(&count); err != nil { + return 0, sqliteFault("queue", err, false) + } + return count, nil +} + +func (t *sqliteTx) createRun(r Run, caller, key, requestDigest string) error { + _, err := t.tx.Exec("INSERT INTO runs(id,job_id,caller,idem,request_digest,state,record) VALUES (?,?,?,?,?,?,?)", r.ID, r.JobID, caller, key, requestDigest, r.State, runText(r)) + return sqliteFault("run", err, true) +} + +func (t *sqliteTx) putRun(r Run, expected string) (string, error) { + var from string + if err := t.tx.QueryRow("SELECT state FROM runs WHERE id=?", r.ID).Scan(&from); err != nil { + return "", sqliteFault("run", err, false) + } + if expected != anyState && from != expected { + return from, storeFailure(faultConflict, "run", fmt.Errorf("the run %s is %s, not %s", r.ID, from, expected)) + } + if _, err := t.tx.Exec("UPDATE runs SET state=?,record=? WHERE id=?", r.State, runText(r), r.ID); err != nil { + return from, sqliteFault("run", err, true) + } + return from, nil +} + +func (t *sqliteTx) runningRuns() ([]Run, error) { + var runs []Run + if err := eachRecord(t.tx, "SELECT record FROM runs WHERE state='running' ORDER BY seq", func(b []byte) error { + r, e := decodeRun(b) + runs = append(runs, r) + return e + }); err != nil { + return runs, storeFailure(faultUnreadable, "queue", err) + } + return runs, nil +} + +func (t *sqliteTx) lastEntry() (int64, []byte, error) { + var last int64 + var line []byte + if err := t.tx.QueryRow("SELECT sequence,line FROM run_chain ORDER BY sequence DESC LIMIT 1").Scan(&last, &line); err != nil { + return 0, nil, sqliteFault("chain", err, false) + } + return last, line, nil +} + +func (t *sqliteTx) createEntry(sequence int64, run string, line []byte) error { + _, err := t.tx.Exec("INSERT INTO run_chain(sequence,run,line) VALUES (?,?,?)", sequence, run, line) + return sqliteFault("chain", err, true) +} + +func (t *sqliteTx) createRelease(r Release) error { + _, err := t.tx.Exec("INSERT INTO releases(id,record) VALUES (?,?)", r.ID, string(encode(r))) + return sqliteFault("release", err, true) +} + +func (t *sqliteTx) appendJournal(c change) (int64, error) { + sequence, err := appendEvent(t.tx, c) + return sequence, sqliteFault("journal", err, true) +} diff --git a/scripts/checks/unit-tests.sh b/scripts/checks/unit-tests.sh index 3045c5d..2c53e1d 100755 --- a/scripts/checks/unit-tests.sh +++ b/scripts/checks/unit-tests.sh @@ -9,3 +9,11 @@ JPACK_GATEWAY_TEST_BIN="$bin/gateway" \ JPACK_MCP_ADAPTER_TEST_BIN="$bin/adapter-mcp" \ JPACK_OLD_RUNNER_TEST_BIN="$bin/jpack-runner-old" \ go test -race ./... +# The store contract's conformance suite runs against every kind +# (docs/design/store-interface.md): a kind whose cases do not run and pass +# fails this step, as a skipped test would otherwise pass. +out="$(go test -race -count=1 -v -run '^TestStoreKindsConform$' ./internal/runner)" || { printf '%s\n' "$out"; exit 1; } +for kind in sqlite memory; do + printf '%s\n' "$out" | grep -q -- "^[[:space:]]*--- PASS: TestStoreKindsConform/$kind " || { echo "TestStoreKindsConform/$kind did not run and pass"; exit 1; } +done +echo "store conformance: the sqlite and memory kinds ran and passed"