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