internal/store/store.go
296 lines · 8661 bytes
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 step := func(sqlText string, newVersion int, fkOff bool) (retErr error) {
193 if !fkOff {
194 tx, err := s.DB.Begin()
195 if err != nil {
196 return err
197 }
198 defer tx.Rollback()
199 if _, err := tx.Exec(sqlText); err != nil {
200 return err
201 }
202 if _, err := tx.Exec(fmt.Sprintf("PRAGMA user_version = %d", newVersion)); err != nil {
203 return err
204 }
205 return tx.Commit()
206 }
207
208 // A script whose first line is "-- foreign_keys: off" rebuilds a
209 // table that other tables reference (labels, milestones): with
210 // foreign keys on, the rebuild-by-rename loses the children's
211 // rows. PRAGMA foreign_keys is a no-op inside a transaction, and
212 // the pool gives no guarantee that a pragma set on one connection
213 // is seen by the connection Begin() draws next, so the whole step
214 // — pragma off, transaction, pragma on, foreign_key_check — runs
215 // on a single pinned connection.
216 ctx := context.Background()
217 conn, err := s.DB.Conn(ctx)
218 if err != nil {
219 return err
220 }
221 defer conn.Close()
222 if _, err := conn.ExecContext(ctx, "PRAGMA foreign_keys = OFF"); err != nil {
223 return err
224 }
225 // The connection goes back to the pool when this returns, so every
226 // path out of here has to put foreign keys back on first.
227 restoreFK := func() error {
228 _, err := conn.ExecContext(ctx, "PRAGMA foreign_keys = ON")
229 return err
230 }
231 defer func() {
232 if err := restoreFK(); err != nil && retErr == nil {
233 retErr = err
234 }
235 }()
236 tx, err := conn.BeginTx(ctx, nil)
237 if err != nil {
238 return err
239 }
240 defer tx.Rollback()
241 if _, err := tx.Exec(sqlText); err != nil {
242 return err
243 }
244 if _, err := tx.Exec(fmt.Sprintf("PRAGMA user_version = %d", newVersion)); err != nil {
245 return err
246 }
247 if err := tx.Commit(); err != nil {
248 return err
249 }
250 if err := restoreFK(); err != nil {
251 return err
252 }
253 rows, err := conn.QueryContext(ctx, "PRAGMA foreign_key_check")
254 if err != nil {
255 return err
256 }
257 defer rows.Close()
258 if rows.Next() {
259 var table string
260 var rowid sql.NullInt64
261 var referredTable string
262 var fkid int
263 if err := rows.Scan(&table, &rowid, &referredTable, &fkid); err != nil {
264 return err
265 }
266 return fmt.Errorf("foreign_key_check failed after migration: %s", table)
267 }
268 return rows.Err()
269 }
270 for cur < target {
271 m := ms[cur]
272 if err := step(m.up, m.version, m.upFKOff); err != nil {
273 return fmt.Errorf("migration %d up: %w", m.version, err)
274 }
275 cur = m.version
276 }
277 for cur > target {
278 m := ms[cur-1]
279 if err := step(m.down, m.version-1, m.downFKOff); err != nil {
280 return fmt.Errorf("migration %d down: %w", m.version, err)
281 }
282 cur = m.version - 1
283 }
284 return nil
285}
286
287// IsInternal reports whether err is the database or the I/O beneath it
288// failing, as opposed to a sentinel or a message about the caller's
289// input. Callers map it to a failure exit rather than a usage error.
290func IsInternal(err error) bool {
291 var sqlErr *sqlite.Error
292 var pathErr *fs.PathError
293 return errors.As(err, &sqlErr) || errors.As(err, &pathErr) ||
294 errors.Is(err, sql.ErrTxDone) || errors.Is(err, sql.ErrConnDone) ||
295 errors.Is(err, context.DeadlineExceeded) || errors.Is(err, context.Canceled)
296}