1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
|
package main
import (
"database/sql"
"fmt"
"log"
"time"
_ "modernc.org/sqlite"
)
// ============================================================================
// Database handles
// ============================================================================
// database is the pair of connection pools mailweb talks to SQLite through: one
// connection for writing, several for reading.
//
// The split exists because the original single pool answered a writers'
// question with an answer that bound the readers too. SetMaxOpenConns(1)
// arrived alongside the ten parallel IMAP backfill workers, to serialise their
// commits and avoid SQLITE_BUSY, which is real — SQLite admits one writer at a
// time. But WAL exists precisely so that readers never contend, and a pool
// capped at one hands every page render the same queue as the backfill.
// Measured on this archive, eight concurrent reads of the shape the contact
// page issues took ~1.07s through one connection and ~490ms through several:
// the cap cost roughly 2x on reads to solve something only writes had.
//
// The two handles are separate fields rather than one pool with a larger cap
// because that is what makes the rule enforceable instead of merely stated:
//
// - Write is capped at one connection, so writes are serialised exactly as
// before and the backfill's guarantee is unchanged.
// - Read opens the same file with mode=ro, so SQLite itself refuses a write
// issued through it ("attempt to write a readonly database"). A handler
// that reaches for the wrong handle fails immediately and locally, rather
// than corrupting the serialisation of something far away.
//
// This also dissolves a whole class of deadlock rather than documenting it. A
// *sql.Rows holds its connection until closed or drained; with a single shared
// pool, any query issued while one was open — including a helper as innocuous
// as loadPetnames — waited for a connection only that cursor could release, and
// the request hung outright instead of running slowly. That was introduced
// once, in the contact detail handler, and presented as a page that never
// returned. With several read connections the pattern costs a connection rather
// than the request. Loading before the loop is still better, but forgetting is
// no longer fatal.
type database struct {
// Read serves every query that renders a page. Read-only at the SQLite
// level, so it cannot be misused.
Read *sql.DB
// Write serves migrations, sync and every mutation. One connection.
Write *sql.DB
}
// openReadPool opens the read-only companion to a database already created and
// migrated by openDB.
//
// It is opened after migrations rather than beside them because mode=ro cannot
// create a file: pointing it at a database that does not exist yet fails, and
// on a fresh install the file does not exist until openDB has made it.
//
// No journal_mode pragma is set here. Setting it is a write to the database
// header, which a read-only connection may not perform; the mode is a property
// of the file that openDB has already established, and WAL is what allows these
// readers to run alongside the writer at all.
func openReadPool(path string, maxConns int) (*sql.DB, error) {
db, err := sql.Open("sqlite",
"file:"+path+"?mode=ro&_pragma=busy_timeout(30000)")
if err != nil {
return nil, fmt.Errorf("open read pool: %w", err)
}
db.SetMaxOpenConns(maxConns)
// sql.Open is lazy, so a bad path or an unreadable file would otherwise
// surface on the first page render rather than at startup.
if err := db.Ping(); err != nil {
db.Close()
return nil, fmt.Errorf("ping read pool: %w", err)
}
return db, nil
}
// readPoolSize is how many concurrent reads may be in flight. This is a
// single-user application behind a loopback address, so the figure only has to
// cover one browser's parallel iframe loads: the message index issues one
// request per framed message, ten at a time by default.
const readPoolSize = 8
func openDB(path string) (*sql.DB, error) {
// SQLite only supports one writer at a time. With parallel backfill workers
// all trying to commit simultaneously we'd get SQLITE_BUSY errors without a
// retry timeout, so busy_timeout makes a blocked writer wait up to 30s, and
// WAL mode keeps readers from contending with it in the first place.
//
// The pragmas travel in the DSN because busy_timeout is per-connection
// state: the modernc driver applies _pragma parameters to every connection
// it opens, whereas a post-open db.Exec("PRAGMA …") only configures the one
// pooled connection that happens to run it. That distinction does not bite
// today — the pool is capped at a single connection below — but it would
// silently break the moment that cap were raised.
db, err := sql.Open("sqlite",
"file:"+path+"?_pragma=busy_timeout(30000)&_pragma=journal_mode(WAL)")
if err != nil {
return nil, fmt.Errorf("open db: %w", err)
}
// Writers are serialised to one connection; see the database type.
db.SetMaxOpenConns(1)
// Bootstrap: schema_version may not exist yet on a brand-new DB.
// We detect this by checking if the table exists before running migrations.
var tableCount int
err = db.QueryRow(
`SELECT COUNT(*) FROM sqlite_master WHERE type='table' AND name='schema_version'`,
).Scan(&tableCount)
if err != nil {
return nil, fmt.Errorf("check schema_version: %w", err)
}
// Find the current version (0 if schema_version doesn't exist yet).
currentVersion := 0
if tableCount > 0 {
var v sql.NullInt64
if err := db.QueryRow(`SELECT MAX(version) FROM schema_version`).Scan(&v); err != nil {
return nil, fmt.Errorf("read schema version: %w", err)
}
if v.Valid {
currentVersion = int(v.Int64)
}
}
// Apply any pending migrations in order.
for i, migration := range migrations {
version := i + 1
if version <= currentVersion {
continue
}
log.Printf("db: applying migration %d", version)
tx, err := db.Begin()
if err != nil {
return nil, fmt.Errorf("begin migration %d: %w", version, err)
}
if _, err := tx.Exec(migration); err != nil {
tx.Rollback()
return nil, fmt.Errorf("migration %d: %w", version, err)
}
if _, err := tx.Exec(
`INSERT INTO schema_version(version, applied_at) VALUES (?, ?)`,
version, time.Now().Unix(),
); err != nil {
tx.Rollback()
return nil, fmt.Errorf("record migration %d: %w", version, err)
}
if err := tx.Commit(); err != nil {
return nil, fmt.Errorf("commit migration %d: %w", version, err)
}
log.Printf("db: migration %d applied", version)
}
return db, nil
}
// backfillHeaders populates message_headers for any messages that have
// header_raw but no rows yet in message_headers. Run once at startup after
// migrations, before serving requests.
func backfillHeaders(db *sql.DB) error {
rows, err := db.Query(
`SELECT id, header_raw FROM messages
WHERE header_raw IS NOT NULL
AND id NOT IN (SELECT DISTINCT message_id FROM message_headers)`,
)
if err != nil {
return fmt.Errorf("backfill headers query: %w", err)
}
defer rows.Close()
type pending struct {
id int64
headerRaw []byte
}
var msgs []pending
for rows.Next() {
var p pending
if err := rows.Scan(&p.id, &p.headerRaw); err != nil {
return fmt.Errorf("scan: %w", err)
}
msgs = append(msgs, p)
}
if err := rows.Err(); err != nil {
return err
}
if len(msgs) == 0 {
return nil
}
log.Printf("db: backfilling headers for %d messages", len(msgs))
// Process in batches of 200 to keep transactions small.
const batchSize = 200
for i := 0; i < len(msgs); i += batchSize {
end := min(i+batchSize, len(msgs))
tx, err := db.Begin()
if err != nil {
return fmt.Errorf("begin: %w", err)
}
for _, p := range msgs[i:end] {
if err := parseAndInsertHeaders(tx, p.id, p.headerRaw); err != nil {
tx.Rollback()
return fmt.Errorf("msg %d: %w", p.id, err)
}
}
if err := tx.Commit(); err != nil {
return fmt.Errorf("commit: %w", err)
}
log.Printf("db: backfilled headers %d/%d", end, len(msgs))
}
log.Printf("db: header backfill complete")
return nil
}
|