diff --git a/internal/database/database.go b/internal/database/database.go index 3d4e89e..2a30e31 100644 --- a/internal/database/database.go +++ b/internal/database/database.go @@ -81,6 +81,23 @@ const ( sqliteHardHeapLimitBytes = 2147483648 ) +// Settings that bound the WAL file's size once no read is open (see +// Checkpoint). +const ( + // sqliteWALSizeLimitBytes (64 MiB) is the WAL file size past which + // Checkpoint runs a TRUNCATE checkpoint. It is also the + // journal_size_limit of the connection kept for writing: when SQLite + // starts the WAL again from the beginning, it cuts a larger WAL file down + // to this size. + sqliteWALSizeLimitBytes = 67108864 + // truncateBusyTimeoutMs is the busy timeout of the TRUNCATE checkpoint: + // the longest it waits for open reads, and so the longest it keeps batch + // writes waiting beyond the copy itself. + truncateBusyTimeoutMs = 500 + // walFileSuffix ends the name of the WAL file, next to the database file. + walFileSuffix = "-wal" +) + // Settings of the ANALYZE that Analyze runs. const ( // analysisLimit is about how many rows of each index ANALYZE reads @@ -112,6 +129,12 @@ var ( ErrNoRoute = errors.New("no route found") // ErrNoStaleASN is returned when no ASN needs WHOIS refresh ErrNoStaleASN = errors.New("no stale ASN found") + // ErrCheckpointIncomplete is returned by Checkpoint when it left part of + // the WAL uncopied, or left a WAL file over sqliteWALSizeLimitBytes + // untruncated, because reads still needed the WAL or SQLite reported the + // checkpoint busy. A later checkpoint that runs when no read is open + // finishes the work. + ErrCheckpointIncomplete = errors.New("WAL checkpoint incomplete") errInvalidASN = errors.New("invalid ASN") errSameASNInPeering = errors.New("cannot create peering with same ASN") @@ -194,6 +217,13 @@ func New(cfg *config.Config, logger *logger.Logger) (*Database, error) { err) } + _, err = writer.ExecContext(ctx, + fmt.Sprintf("PRAGMA journal_size_limit=%d", sqliteWALSizeLimitBytes)) + if err != nil { + return nil, fmt.Errorf( + "failed to set the writing connection's WAL size limit: %w", err) + } + database := &Database{ db: db, writer: writer, logger: logger, path: dbPath, counts: &liveCounts{}, @@ -2255,9 +2285,24 @@ func (d *Database) Analyze(ctx context.Context) error { return nil } -// Checkpoint runs a WAL checkpoint to transfer data from the WAL to the main database. -// Uses TRUNCATE mode which blocks writers briefly but ensures complete checkpoint, -// keeping the WAL small for fast read performance. It holds the database lock. +// Checkpoint copies the WAL into the database file. It holds the database +// lock, so batch writes wait while it runs. +// +// It runs a PASSIVE checkpoint, which never waits: it copies what no open read +// still needs, and leaves the file as large as it is. When the file is over +// sqliteWALSizeLimitBytes, Checkpoint runs one TRUNCATE checkpoint instead, +// which waits at most truncateBusyTimeoutMs for the open reads to end, then +// copies all of the WAL and empties the file, or gives up. It is logged with +// how long it took and whether it emptied the file. +// +// So a file over the limit is emptied by the first Checkpoint that finds no +// read open, at its start or at some moment of its wait. While reads follow +// one another with no pause between them, every TRUNCATE checkpoint can give +// up after its full wait, and the file can keep growing for as long as they +// do. +// +// It returns ErrCheckpointIncomplete when it leaves part of the WAL uncopied, +// or a file over the limit not emptied. func (d *Database) Checkpoint(ctx context.Context) error { d.lock("Checkpoint") defer d.unlock() @@ -2267,11 +2312,32 @@ func (d *Database) Checkpoint(ctx context.Context) error { return fmt.Errorf("failed to checkpoint WAL: %w", err) } - _, err = d.writer.ExecContext(ctx, "PRAGMA wal_checkpoint(TRUNCATE)") + wal, err := os.Stat(d.path + walFileSuffix) + if err != nil { + return fmt.Errorf("failed to read the WAL file's size: %w", err) + } + + if wal.Size() > sqliteWALSizeLimitBytes { + return d.truncateWAL(ctx, wal.Size()) + } + + var busy, walPages, copiedPages int + + err = d.writer.QueryRowContext(ctx, "PRAGMA wal_checkpoint(PASSIVE)"). + Scan(&busy, &walPages, &copiedPages) if err != nil { return fmt.Errorf("failed to checkpoint WAL: %w", err) } + if busy != 0 { + return fmt.Errorf("%w: SQLite reported it busy", ErrCheckpointIncomplete) + } + + if copiedPages < walPages { + return fmt.Errorf("%w: copied %d of %d WAL pages", + ErrCheckpointIncomplete, copiedPages, walPages) + } + return nil } @@ -2287,6 +2353,50 @@ func (d *Database) Ping(ctx context.Context) error { return nil } +// truncateWAL runs the TRUNCATE checkpoint of Checkpoint on a WAL file of +// walBytes, with the writing connection's busy timeout lowered to +// truncateBusyTimeoutMs for it, and logs how long it took. The caller holds the +// database lock. +func (d *Database) truncateWAL(ctx context.Context, walBytes int64) error { + _, err := d.writer.ExecContext(ctx, + fmt.Sprintf("PRAGMA busy_timeout=%d", truncateBusyTimeoutMs)) + if err != nil { + return fmt.Errorf("failed to lower the busy timeout to truncate WAL: %w", + err) + } + + started := time.Now() + + var busy, walPages, copiedPages int + + checkpointErr := d.writer.QueryRowContext(ctx, + "PRAGMA wal_checkpoint(TRUNCATE)").Scan(&busy, &walPages, &copiedPages) + took := time.Since(started) + + // Restored even when ctx has ended, so that no batch write runs with the + // short timeout. + _, restoreErr := d.writer.ExecContext(context.WithoutCancel(ctx), + fmt.Sprintf("PRAGMA busy_timeout=%d", sqliteBusyTimeoutMs)) + if checkpointErr != nil || restoreErr != nil { + return fmt.Errorf("failed to truncate WAL: %w", + errors.Join(checkpointErr, restoreErr)) + } + + if busy != 0 { + d.logger.Warn("WAL file over its size limit not truncated", + "wal_bytes", walBytes, "wal_pages", walPages, + "copied_pages", copiedPages, "duration", took) + + return fmt.Errorf("%w: WAL file of %d bytes not truncated", + ErrCheckpointIncomplete, walBytes) + } + + d.logger.Info("WAL file over its size limit truncated", + "wal_bytes", walBytes, "duration", took) + + return nil +} + // The debug messages lock logs, with the operation's name, when it starts to // wait for the database mutex and when it holds it. const ( diff --git a/internal/database/database_internal_test.go b/internal/database/database_internal_test.go index 7a48559..88030a1 100644 --- a/internal/database/database_internal_test.go +++ b/internal/database/database_internal_test.go @@ -8,6 +8,7 @@ import ( "log/slog" "math/rand/v2" "net/netip" + "os" "slices" "strconv" "sync" @@ -67,6 +68,15 @@ const ( lockWaitTimeout = time.Millisecond // lockTestASN is the AS the batch write in the lock tests stores. lockTestASN = 64500 + // readerCheckpointLimit is how long a checkpoint, and a batch write + // waiting for it, may take while a reader is open: well under the busy + // timeout, which a checkpoint that waits for the reader takes. + readerCheckpointLimit = time.Second + // truncateCheckpointLimit is how long a checkpoint of a WAL file over + // sqliteWALSizeLimitBytes may take to give up while a reader is open: its + // wait for the reader plus a second, well under the busy timeout. + truncateCheckpointLimit = truncateBusyTimeoutMs*time.Millisecond + + time.Second ) // Parameters for the prefix-length sample tests. @@ -457,6 +467,33 @@ func TestWritingConnectionCache(t *testing.T) { } } +// TestWritingConnectionWALSizeLimit checks that the connection kept for +// writing, the one that writes the WAL, has the WAL size limit, which cuts the +// WAL file back when SQLite starts the WAL again from the beginning. +func TestWritingConnectionWALSizeLimit(t *testing.T) { + t.Parallel() + + db, err := New(&config.Config{StateDir: t.TempDir()}, logger.New()) + if err != nil { + t.Fatalf("failed to create database: %v", err) + } + defer func() { _ = db.Close() }() + + var limit int64 + + err = db.writer.QueryRowContext(t.Context(), "PRAGMA journal_size_limit"). + Scan(&limit) + if err != nil { + t.Fatalf("failed to read the writing connection's journal_size_limit: %v", + err) + } + + if limit != sqliteWALSizeLimitBytes { + t.Errorf("writing connection: journal_size_limit = %d, want %d", + limit, sqliteWALSizeLimitBytes) + } +} + // readPragma reads the integer value of PRAGMA name on connection i, c. func readPragma( ctx context.Context, t *testing.T, c *sql.Conn, i int, name string, @@ -799,6 +836,218 @@ func TestCheckpointLockWaitEndsWithContext(t *testing.T) { db.mu.Unlock() } +// TestCheckpointDoesNotWaitForReader keeps a read transaction on +// live_routes_v4 open while a route is committed, then starts Checkpoint and a +// batch write together. It checks that both end within readerCheckpointLimit, +// whichever of them takes the database lock first, and that Checkpoint reports +// the part of the WAL the reader still needs as left uncopied. A TRUNCATE +// checkpoint waited the busy timeout for the reader while holding the lock +// (https://git.eeqj.de/sneak/routewatch/issues/102). +func TestCheckpointDoesNotWaitForReader(t *testing.T) { + t.Parallel() + + db, err := New(&config.Config{StateDir: t.TempDir()}, logger.New()) + if err != nil { + t.Fatalf("failed to create database: %v", err) + } + defer func() { _ = db.Close() }() + + ctx := context.Background() + ts := time.Now().UTC() + + err = db.UpsertLiveRouteBatch( + []*LiveRoute{mkV4Route(t, "198.51.100.0/24", lockTestASN, ts)}) + if err != nil { + t.Fatalf("failed to store the route the reader reads: %v", err) + } + + reader, err := db.db.Conn(ctx) + if err != nil { + t.Fatalf("failed to open the reading connection: %v", err) + } + defer func() { _ = reader.Close() }() + + // A plain BEGIN starts a transaction that only reads, unlike the BEGIN + // IMMEDIATE of _txlock. It keeps the snapshot its first read takes until + // it ends. + _, err = reader.ExecContext(ctx, "BEGIN") + if err != nil { + t.Fatalf("failed to begin the read transaction: %v", err) + } + defer func() { _, _ = reader.ExecContext(ctx, "ROLLBACK") }() + + var routes int + + err = reader.QueryRowContext(ctx, + "SELECT count(*) FROM live_routes_v4").Scan(&routes) + if err != nil { + t.Fatalf("failed to read the live routes: %v", err) + } + + err = db.UpsertLiveRouteBatch( + []*LiveRoute{mkV4Route(t, "203.0.113.0/24", lockTestASN, ts)}) + if err != nil { + t.Fatalf("failed to store a route after the read: %v", err) + } + + started := time.Now() + checkpointed := make(chan error, 1) + + go func() { checkpointed <- db.Checkpoint(ctx) }() + + // A route write, not an AS write: the first new AS loads the embedded AS + // names, which takes seconds under -race. + route := mkV4Route(t, "192.0.2.0/24", lockTestASN, ts) + written := make(chan error, 1) + + go func() { written <- db.UpsertLiveRouteBatch([]*LiveRoute{route}) }() + + err = <-checkpointed + if !errors.Is(err, ErrCheckpointIncomplete) { + t.Errorf("Checkpoint = %v with a reader open, want %v", + err, ErrCheckpointIncomplete) + } + + took := time.Since(started) + if took >= readerCheckpointLimit { + t.Errorf("Checkpoint took %v with a reader open, want under %v", + took, readerCheckpointLimit) + } + + err = <-written + if err != nil { + t.Errorf("batch write during Checkpoint: %v", err) + } + + took = time.Since(started) + if took >= readerCheckpointLimit { + t.Errorf("batch write during Checkpoint ended after %v, want under %v", + took, readerCheckpointLimit) + } +} + +// TestCheckpointEmptiesWALOnceNoReadIsOpen fills the WAL file past +// sqliteWALSizeLimitBytes while the test holds a read open. With the read +// open, it checks that Checkpoint gives up within truncateCheckpointLimit and +// reports the checkpoint incomplete, and that a batch write still succeeds. +// After the test closes the read, it checks that the next Checkpoint brings +// the WAL file under the limit and gives the writing connection its busy +// timeout back (https://git.eeqj.de/sneak/routewatch/issues/102). +func TestCheckpointEmptiesWALOnceNoReadIsOpen(t *testing.T) { + t.Parallel() + + db, err := New(&config.Config{StateDir: t.TempDir()}, logger.New()) + if err != nil { + t.Fatalf("failed to create database: %v", err) + } + defer func() { _ = db.Close() }() + + ctx := t.Context() + + reader, err := db.db.Conn(ctx) + if err != nil { + t.Fatalf("failed to open the reading connection: %v", err) + } + defer func() { _ = reader.Close() }() + + // A plain BEGIN starts a transaction that only reads, unlike the BEGIN + // IMMEDIATE of _txlock. It keeps the snapshot its first read takes until + // it ends, so no checkpoint can copy what is written after that read. + _, err = reader.ExecContext(ctx, "BEGIN") + if err != nil { + t.Fatalf("failed to begin the read transaction: %v", err) + } + + var routes int + + err = reader.QueryRowContext(ctx, + "SELECT count(*) FROM live_routes_v4").Scan(&routes) + if err != nil { + t.Fatalf("failed to read the live routes: %v", err) + } + + fillWALPastLimit(ctx, t, db) + + started := time.Now() + + err = db.Checkpoint(ctx) + if !errors.Is(err, ErrCheckpointIncomplete) { + t.Errorf("Checkpoint = %v with a read open, want %v", + err, ErrCheckpointIncomplete) + } + + took := time.Since(started) + if took >= truncateCheckpointLimit { + t.Errorf("Checkpoint took %v with a read open, want under %v", + took, truncateCheckpointLimit) + } + + err = db.UpsertLiveRouteBatch([]*LiveRoute{ + mkV4Route(t, "192.0.2.0/24", lockTestASN, time.Now().UTC()), + }) + if err != nil { + t.Errorf("batch write after Checkpoint gave up: %v", err) + } + + _, err = reader.ExecContext(ctx, "ROLLBACK") + if err != nil { + t.Fatalf("failed to end the read transaction: %v", err) + } + + err = db.Checkpoint(ctx) + if err != nil { + t.Errorf("Checkpoint with no read open: %v", err) + } + + size := walFileSize(t, db) + if size >= sqliteWALSizeLimitBytes { + t.Errorf("WAL file of %d bytes after Checkpoint with no read open, "+ + "want under %d", size, sqliteWALSizeLimitBytes) + } + + busyTimeout := readPragma(ctx, t, db.writer, 0, "busy_timeout") + if busyTimeout != sqliteBusyTimeoutMs { + t.Errorf("writing connection: busy_timeout = %d after the checkpoints, "+ + "want %d", busyTimeout, sqliteBusyTimeoutMs) + } +} + +// fillWALPastLimit writes sqliteWALSizeLimitBytes of random bytes to a table +// of the test's own, which with SQLite's page and frame headers leaves the WAL +// file of db over the limit, and checks that it does. +func fillWALPastLimit(ctx context.Context, t *testing.T, db *Database) { + t.Helper() + + _, err := db.writer.ExecContext(ctx, "CREATE TABLE wal_fill (data BLOB)") + if err != nil { + t.Fatalf("failed to create the table the test writes to: %v", err) + } + + _, err = db.writer.ExecContext(ctx, + "INSERT INTO wal_fill VALUES (randomblob(?))", sqliteWALSizeLimitBytes) + if err != nil { + t.Fatalf("failed to fill the WAL: %v", err) + } + + size := walFileSize(t, db) + if size <= sqliteWALSizeLimitBytes { + t.Fatalf("WAL file of %d bytes after the test's write, want over %d", + size, sqliteWALSizeLimitBytes) + } +} + +// walFileSize returns the size of the WAL file of db. +func walFileSize(t *testing.T, db *Database) int64 { + t.Helper() + + wal, err := os.Stat(db.path + walFileSuffix) + if err != nil { + t.Fatalf("failed to read the WAL file's size: %v", err) + } + + return wal.Size() +} + // TestGetRandomPrefixesByLengthListsPrefixOnce stores one prefix of each // family learned from two peers in the same second and checks that the // prefix-length sample lists each of them once. diff --git a/internal/routewatch/dbmaintainer.go b/internal/routewatch/dbmaintainer.go index 8859a46..6ecdf0b 100644 --- a/internal/routewatch/dbmaintainer.go +++ b/internal/routewatch/dbmaintainer.go @@ -2,6 +2,7 @@ package routewatch import ( "context" + "errors" "log/slog" "sync" "time" @@ -34,6 +35,12 @@ const ( analyzeTimeout = 5 * time.Minute ) +// checkpointIncompleteMessage is logged, at debug level like a completed +// checkpoint, when a checkpoint leaves part of the WAL for a later one: while +// reads overlap, most do. The database warns when the WAL file stays over its +// size limit. +const checkpointIncompleteMessage = "WAL checkpoint incomplete" + // DBMaintainer handles background database maintenance tasks. type DBMaintainer struct { db database.Store @@ -123,9 +130,13 @@ func (m *DBMaintainer) runCheckpoint(ctx context.Context) { duration, err := runWithTimeout(ctx, checkpointTimeout, m.db.Checkpoint) m.recordResult(&m.lastCheckpoint, &m.lastCheckpointError, &m.checkpointCount, err) - if err != nil { + switch { + case errors.Is(err, database.ErrCheckpointIncomplete): + m.logger.Debug(checkpointIncompleteMessage, + "error", err, "duration", duration) + case err != nil: m.logger.Error("WAL checkpoint failed", "error", err, "duration", duration) - } else { + default: m.logger.Debug("WAL checkpoint completed", "duration", duration) } } diff --git a/internal/routewatch/dbmaintainer_internal_test.go b/internal/routewatch/dbmaintainer_internal_test.go new file mode 100644 index 0000000..94d258a --- /dev/null +++ b/internal/routewatch/dbmaintainer_internal_test.go @@ -0,0 +1,54 @@ +package routewatch + +import ( + "bytes" + "context" + "encoding/json" + "log/slog" + "testing" + + "git.eeqj.de/sneak/routewatch/internal/database" +) + +// incompleteCheckpointStore is a store whose every checkpoint leaves part of +// the WAL uncopied. +type incompleteCheckpointStore struct{ *mockStore } + +// Checkpoint reports the checkpoint incomplete. +func (incompleteCheckpointStore) Checkpoint(context.Context) error { + return database.ErrCheckpointIncomplete +} + +// TestIncompleteCheckpointNotLoggedAsCompleted runs a checkpoint that leaves +// part of the WAL uncopied and checks that the maintainer logs it as +// incomplete, not as completed, and does not count it as a checkpoint that +// completed. +func TestIncompleteCheckpointNotLoggedAsCompleted(t *testing.T) { + t.Parallel() + + var logs bytes.Buffer + + logger := slog.New(slog.NewJSONHandler(&logs, + &slog.HandlerOptions{Level: slog.LevelDebug})) + m := NewDBMaintainer(incompleteCheckpointStore{newMockStore()}, logger) + + m.runCheckpoint(t.Context()) + + var logged struct { + Msg string `json:"msg"` + } + + err := json.Unmarshal(logs.Bytes(), &logged) + if err != nil { + t.Fatalf("want one log message, got %q: %v", logs.String(), err) + } + + if logged.Msg != checkpointIncompleteMessage { + t.Errorf("logged %q, want %q", logged.Msg, checkpointIncompleteMessage) + } + + if m.checkpointCount != 0 { + t.Errorf("checkpointCount = %d after an incomplete checkpoint, want 0", + m.checkpointCount) + } +}