The Runner's store behind an interface: SQLite as the shipped kind, an in-memory kind, a conformance suite (#173)

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,

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 {

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,45 +173,42 @@ 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
 }
 
 // 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.

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)
 		}
 	}