internal/store/store.go

  1// Package store owns SQLite access and schema migrations.
  2package store
  3
  4import (
  5	"context"
  6	"database/sql"
  7	"embed"
  8	"errors"
  9	"fmt"
 10	"io/fs"
 11	"log/slog"
 12	"os"
 13	"sort"
 14	"strconv"
 15	"strings"
 16	"sync"
 17
 18	"modernc.org/sqlite"
 19)
 20
 21//go:embed migrations/*.sql
 22var migrationFS embed.FS
 23
 24type Store struct {
 25	DB *sql.DB
 26
 27	// logWait holds one channel per build someone is following, closed
 28	// by the next change to that build's row (BuildLogWait).
 29	logMu   sync.Mutex
 30	logWait map[int64]chan struct{}
 31
 32	// onRevoke runs after each key revocation this process commits.
 33	revokeMu sync.Mutex
 34	onRevoke []func(Revoked)
 35
 36	// AuditJournal, when set, receives a copy of every audit row. The
 37	// daemon sets it to its own logger, whose output the service
 38	// journal keeps outside the database.
 39	AuditJournal *slog.Logger
 40}
 41
 42// Open opens (creating if needed) the database at path with WAL mode and
 43// foreign keys enforced. Use ":memory:" in tests.
 44//
 45// _txlock=immediate is what serialises writers. Every transaction in this
 46// package writes, and a deferred one takes the write lock only when it
 47// reaches its first write — by which point another writer may hold it.
 48// SQLite answers that with SQLITE_BUSY and does not invoke the busy
 49// handler, because waiting would deadlock two transactions each holding a
 50// read lock the other needs; busy_timeout cannot help. Measured with
 51// eight concurrent read-then-write transactions, 44% of them failed.
 52// Beginning immediate takes the write lock up front, where busy_timeout
 53// does apply, so a second writer waits its turn: the same load runs with
 54// no failures, and readers, which WAL keeps out of the way, are
 55// unaffected (#121).
 56func Open(path string) (*Store, error) {
 57	dsn := path + "?_txlock=immediate&_pragma=journal_mode(WAL)&_pragma=foreign_keys(ON)&_pragma=busy_timeout(5000)"
 58	if path == ":memory:" {
 59		dsn = ":memory:?_txlock=immediate&_pragma=foreign_keys(ON)"
 60	}
 61	db, err := sql.Open("sqlite", dsn)
 62	if err != nil {
 63		return nil, err
 64	}
 65	if err := db.Ping(); err != nil {
 66		db.Close()
 67		return nil, err
 68	}
 69	// SQLite creates the file 0666&~umask, so it lands 0644 by default. The
 70	// directory above it is the real boundary, but the file holds token
 71	// hashes, addresses and private repo names and has no business being
 72	// world-readable on its own.
 73	if path != ":memory:" {
 74		if err := os.Chmod(path, 0o640); err != nil && !errors.Is(err, fs.ErrNotExist) {
 75			db.Close()
 76			return nil, err
 77		}
 78	}
 79	return &Store{DB: db}, nil
 80}
 81
 82func (s *Store) Close() error { return s.DB.Close() }
 83
 84type migration struct {
 85	version int
 86	name    string
 87	up      string
 88	down    string
 89	// upFKOff and downFKOff are true when the up/down script's first line
 90	// is the directive "-- foreign_keys: off".
 91	upFKOff   bool
 92	downFKOff bool
 93}
 94
 95// fkOffDirective, as the first line of a migration script, opts that
 96// direction out of foreign-key enforcement for its step.
 97const fkOffDirective = "-- foreign_keys: off"
 98
 99func loadMigrations() ([]migration, error) {
100	entries, err := fs.ReadDir(migrationFS, "migrations")
101	if err != nil {
102		return nil, err
103	}
104	byVersion := map[int]*migration{}
105	for _, e := range entries {
106		name := e.Name()
107		// <version>_<name>.<up|down>.sql
108		base, ok := strings.CutSuffix(name, ".sql")
109		if !ok {
110			return nil, fmt.Errorf("migration %q: not .sql", name)
111		}
112		var dir string
113		if b, ok := strings.CutSuffix(base, ".up"); ok {
114			base, dir = b, "up"
115		} else if b, ok := strings.CutSuffix(base, ".down"); ok {
116			base, dir = b, "down"
117		} else {
118			return nil, fmt.Errorf("migration %q: missing .up/.down", name)
119		}
120		verStr, rest, ok := strings.Cut(base, "_")
121		if !ok {
122			return nil, fmt.Errorf("migration %q: missing version prefix", name)
123		}
124		ver, err := strconv.Atoi(verStr)
125		if err != nil {
126			return nil, fmt.Errorf("migration %q: bad version: %w", name, err)
127		}
128		m := byVersion[ver]
129		if m == nil {
130			m = &migration{version: ver, name: rest}
131			byVersion[ver] = m
132		}
133		sqlBytes, err := migrationFS.ReadFile("migrations/" + name)
134		if err != nil {
135			return nil, err
136		}
137		text := string(sqlBytes)
138		firstLine, _, _ := strings.Cut(text, "\n")
139		fkOff := strings.TrimSpace(firstLine) == fkOffDirective
140		if dir == "up" {
141			m.up = text
142			m.upFKOff = fkOff
143		} else {
144			m.down = text
145			m.downFKOff = fkOff
146		}
147	}
148	var ms []migration
149	for _, m := range byVersion {
150		if m.up == "" || m.down == "" {
151			return nil, fmt.Errorf("migration %d %q: missing up or down file", m.version, m.name)
152		}
153		ms = append(ms, *m)
154	}
155	sort.Slice(ms, func(i, j int) bool { return ms[i].version < ms[j].version })
156	for i, m := range ms {
157		if m.version != i+1 {
158			return nil, fmt.Errorf("migration versions not contiguous at %d", m.version)
159		}
160	}
161	return ms, nil
162}
163
164// Version returns the current schema version (0 = empty database).
165func (s *Store) Version() (int, error) {
166	var v int
167	err := s.DB.QueryRow("PRAGMA user_version").Scan(&v)
168	return v, err
169}
170
171// MigrateUp applies all pending migrations.
172func (s *Store) MigrateUp() error { return s.migrateTo(-1) }
173
174// MigrateTo migrates up or down to the given version. 0 empties the schema.
175func (s *Store) MigrateTo(target int) error { return s.migrateTo(target) }
176
177func (s *Store) migrateTo(target int) error {
178	ms, err := loadMigrations()
179	if err != nil {
180		return err
181	}
182	if target < 0 {
183		target = len(ms)
184	}
185	if target > len(ms) {
186		return fmt.Errorf("no such schema version %d (max %d)", target, len(ms))
187	}
188	cur, err := s.Version()
189	if err != nil {
190		return err
191	}
192	for cur < target {
193		m := ms[cur]
194		if err := s.migrateStep(m.up, m.version, m.upFKOff); err != nil {
195			return fmt.Errorf("migration %d up: %w", m.version, err)
196		}
197		cur = m.version
198	}
199	for cur > target {
200		m := ms[cur-1]
201		if err := s.migrateStep(m.down, m.version-1, m.downFKOff); err != nil {
202			return fmt.Errorf("migration %d down: %w", m.version, err)
203		}
204		cur = m.version - 1
205	}
206	return nil
207}
208
209func (s *Store) migrateStep(sqlText string, newVersion int, fkOff bool) (retErr error) {
210	if !fkOff {
211		tx, err := s.DB.Begin()
212		if err != nil {
213			return err
214		}
215		defer tx.Rollback()
216		if _, err := tx.Exec(sqlText); err != nil {
217			return err
218		}
219		if _, err := tx.Exec(fmt.Sprintf("PRAGMA user_version = %d", newVersion)); err != nil {
220			return err
221		}
222		return tx.Commit()
223	}
224
225	// A script whose first line is "-- foreign_keys: off" rebuilds a
226	// table that other tables reference (labels, milestones): with
227	// foreign keys on, the rebuild-by-rename loses the children's
228	// rows. PRAGMA foreign_keys is a no-op inside a transaction, and
229	// the pool gives no guarantee that a pragma set on one connection
230	// is seen by the connection Begin() draws next, so the whole step
231	// — pragma off, transaction, foreign_key_check, commit, pragma on —
232	// runs on a single pinned connection. The check runs before commit:
233	// checking after would report a violation once the bad schema and
234	// user_version were already persisted.
235	ctx := context.Background()
236	conn, err := s.DB.Conn(ctx)
237	if err != nil {
238		return err
239	}
240	defer conn.Close()
241	if _, err := conn.ExecContext(ctx, "PRAGMA foreign_keys = OFF"); err != nil {
242		return err
243	}
244	// The connection goes back to the pool when this returns, so every
245	// path out of here has to put foreign keys back on first.
246	defer func() {
247		if _, err := conn.ExecContext(ctx, "PRAGMA foreign_keys = ON"); err != nil && retErr == nil {
248			retErr = err
249		}
250	}()
251	tx, err := conn.BeginTx(ctx, nil)
252	if err != nil {
253		return err
254	}
255	defer tx.Rollback()
256	if _, err := tx.Exec(sqlText); err != nil {
257		return err
258	}
259	if _, err := tx.Exec(fmt.Sprintf("PRAGMA user_version = %d", newVersion)); err != nil {
260		return err
261	}
262	// foreign_key_check works with enforcement off: it inspects the data
263	// directly rather than consulting the pragma. Running it here, inside
264	// the transaction, means a violation rolls back the whole rebuild
265	// (the deferred tx.Rollback fires) instead of leaving the bad schema
266	// and version committed.
267	rows, err := tx.QueryContext(ctx, "PRAGMA foreign_key_check")
268	if err != nil {
269		return err
270	}
271	if rows.Next() {
272		var table string
273		var rowid sql.NullInt64
274		var referredTable string
275		var fkid int
276		if err := rows.Scan(&table, &rowid, &referredTable, &fkid); err != nil {
277			rows.Close()
278			return err
279		}
280		rows.Close()
281		return fmt.Errorf("foreign_key_check failed after migration: %s row %v", table, rowid)
282	}
283	if err := rows.Err(); err != nil {
284		rows.Close()
285		return err
286	}
287	rows.Close()
288	return tx.Commit()
289}
290
291// IsInternal reports whether err is the database or the I/O beneath it
292// failing, as opposed to a sentinel or a message about the caller's
293// input. Callers map it to a failure exit rather than a usage error.
294func IsInternal(err error) bool {
295	var sqlErr *sqlite.Error
296	var pathErr *fs.PathError
297	return errors.As(err, &sqlErr) || errors.As(err, &pathErr) ||
298		errors.Is(err, sql.ErrTxDone) || errors.Is(err, sql.ErrConnDone) ||
299		errors.Is(err, context.DeadlineExceeded) || errors.Is(err, context.Canceled)
300}