From cdc0d33da768b4c35f532d5cc178b24c1b7b1e2e Mon Sep 17 00:00:00 2001 From: sneak Date: Tue, 20 Oct 2026 02:08:51 +0000 Subject: [PATCH] Log the start-up checkpoint's figures under their own names (closes #117) Initialize scanned the row of PRAGMA wal_checkpoint(TRUNCATE), which is the busy flag, the frames in the WAL and the frames copied, into wal_pages, checkpointed and moved; and a completed TRUNCATE reads 0 for both counts. The line now logs busy, wal_frames and checkpointed_frames, plus wal_bytes, the WAL file's size read before the checkpoint, and is a warning when busy is 1. Tests open a database a second time while the first stays open, with and without a read open on the first, and check the line's figures. Model: opus-5-5 --- internal/database/database.go | 51 ++++-- internal/database/database_internal_test.go | 169 ++++++++++++++++++++ 2 files changed, 210 insertions(+), 10 deletions(-) diff --git a/internal/database/database.go b/internal/database/database.go index ec6bded..04efd7a 100644 --- a/internal/database/database.go +++ b/internal/database/database.go @@ -245,6 +245,12 @@ func New(cfg *config.Config, logger *logger.Logger) (*Database, error) { return database, nil } +// The messages Initialize logs the start-up WAL checkpoint with. +const ( + startupCheckpointMessage = "WAL checkpoint on startup" + startupCheckpointIncompleteMessage = "WAL checkpoint on startup incomplete" +) + // Initialize creates the database schema if it doesn't exist. func (d *Database) Initialize() error { // Set SQLite pragmas for performance. Per-connection settings (cache_size, @@ -266,19 +272,44 @@ func (d *Database) Initialize() error { } } - // Run WAL checkpoint on startup to consolidate any existing WAL - var walPages, checkpointedPages, movedPages int + // The start-up checkpoint copies the WAL a previous run left into the + // database file and empties the WAL file; its frame counts then read 0, + // so the WAL file's size is read first. There is no WAL file when the + // database is new or the last run closed the database, which deletes it. + var walBytes int64 - err := d.writer.QueryRowContext(context.Background(), + wal, err := os.Stat(d.path + walFileSuffix) + if err == nil { + walBytes = wal.Size() + } else if !errors.Is(err, os.ErrNotExist) { + d.logger.Warn("Failed to read the WAL file's size on startup", + "error", err) + } + + var busy, walFrames, checkpointedFrames int + + err = d.writer.QueryRowContext(context.Background(), "PRAGMA wal_checkpoint(TRUNCATE)", - ).Scan(&walPages, &checkpointedPages, &movedPages) - if err != nil { + ).Scan(&busy, &walFrames, &checkpointedFrames) + + switch { + case err != nil: d.logger.Warn("Failed to checkpoint WAL on startup", "error", err) - } else { - d.logger.Info("WAL checkpoint on startup", - "wal_pages", walPages, - "checkpointed", checkpointedPages, - "moved", movedPages, + case busy != 0: + // Another process using the database kept the checkpoint from + // copying or emptying all of the WAL. + d.logger.Warn(startupCheckpointIncompleteMessage, + "wal_bytes", walBytes, + "busy", busy, + "wal_frames", walFrames, + "checkpointed_frames", checkpointedFrames, + ) + default: + d.logger.Info(startupCheckpointMessage, + "wal_bytes", walBytes, + "busy", busy, + "wal_frames", walFrames, + "checkpointed_frames", checkpointedFrames, ) } diff --git a/internal/database/database_internal_test.go b/internal/database/database_internal_test.go index a24f25e..1c526bd 100644 --- a/internal/database/database_internal_test.go +++ b/internal/database/database_internal_test.go @@ -1,8 +1,10 @@ package database import ( + "bytes" "context" "database/sql" + "encoding/json" "errors" "fmt" "log/slog" @@ -1052,6 +1054,173 @@ func walFileSize(t *testing.T, db *Database) int64 { return wal.Size() } +// TestStartupCheckpointLogsItsFigures opens a new database, writes a route, +// and opens the same file again while the first stays open, so that its WAL +// file stays. The first open finds no WAL file: it logs a size of 0 and no +// warning. The second logs the size of the WAL file it found, at least what +// the first left (opening the file writes to the WAL before the size is +// read), and the checkpoint's busy flag and frame counts, each under its own +// name; the counts read 0 once the checkpoint has emptied the file. +// Initialize logged the busy flag as wal_pages and the counts under each +// other's names (https://git.eeqj.de/sneak/routewatch/issues/117). +func TestStartupCheckpointLogsItsFigures(t *testing.T) { + t.Parallel() + + cfg := &config.Config{StateDir: t.TempDir()} + + db, lines := openLogged(t, cfg) + + for _, line := range lines { + if line[slog.LevelKey] == slog.LevelWarn.String() { + t.Errorf("opening a new database logged a warning: %v", line) + } + } + + line := findLine(t, lines, startupCheckpointMessage) + if line["wal_bytes"] != 0.0 { + t.Errorf("new database: start-up checkpoint logged wal_bytes = %v, "+ + "want 0", line["wal_bytes"]) + } + + err := db.UpsertLiveRouteBatch([]*LiveRoute{ + mkV4Route(t, "192.0.2.0/24", lockTestASN, time.Now().UTC()), + }) + if err != nil { + t.Fatalf("failed to store a route: %v", err) + } + + walBytes := walFileSize(t, db) + if walBytes == 0 { + t.Fatal("WAL file empty after the route write, want frames in it") + } + + _, lines = openLogged(t, cfg) + line = findLine(t, lines, startupCheckpointMessage) + + for _, key := range []string{"busy", "wal_frames", "checkpointed_frames"} { + if line[key] != 0.0 { + t.Errorf("start-up checkpoint logged %s = %v, want 0", key, line[key]) + } + } + + loggedBytes, _ := line["wal_bytes"].(float64) + if loggedBytes < float64(walBytes) { + t.Errorf("start-up checkpoint logged wal_bytes = %v, want at least "+ + "the %d bytes the first database left", line["wal_bytes"], walBytes) + } +} + +// TestStartupCheckpointWarnsWhenIncomplete keeps a read open on a database +// while a route is written after the read began, then opens the same file +// again. The second open's start-up checkpoint cannot copy the frames written +// after the read began, and gives up once the busy timeout has passed. It +// checks that the checkpoint is logged as a warning with busy 1 and fewer +// frames copied than the WAL holds. +func TestStartupCheckpointWarnsWhenIncomplete(t *testing.T) { + t.Parallel() + + cfg := &config.Config{StateDir: t.TempDir()} + db, _ := openLogged(t, cfg) + ctx := context.Background() + + 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, "192.0.2.0/24", lockTestASN, time.Now().UTC()), + }) + if err != nil { + t.Fatalf("failed to store a route after the read: %v", err) + } + + _, lines := openLogged(t, cfg) + line := findLine(t, lines, startupCheckpointIncompleteMessage) + + if line["busy"] != 1.0 { + t.Errorf("start-up checkpoint logged busy = %v, want 1", line["busy"]) + } + + walFrames, _ := line["wal_frames"].(float64) + checkpointedFrames, _ := line["checkpointed_frames"].(float64) + + if checkpointedFrames >= walFrames { + t.Errorf("start-up checkpoint logged checkpointed_frames = %v and "+ + "wal_frames = %v with a read open, want fewer copied than the WAL "+ + "holds", line["checkpointed_frames"], line["wal_frames"]) + } +} + +// openLogged opens the database of cfg with a logger that writes JSON, and +// returns it with the lines it logged while it opened, decoded. +func openLogged(t *testing.T, cfg *config.Config) (*Database, []map[string]any) { + t.Helper() + + var logged bytes.Buffer + + db, err := New(cfg, + &logger.Logger{Logger: slog.New(slog.NewJSONHandler(&logged, nil))}) + if err != nil { + t.Fatalf("failed to open the database: %v", err) + } + + t.Cleanup(func() { _ = db.Close() }) + + var lines []map[string]any + + decoder := json.NewDecoder(&logged) + for decoder.More() { + var line map[string]any + + err := decoder.Decode(&line) + if err != nil { + t.Fatalf("failed to decode the log: %v", err) + } + + lines = append(lines, line) + } + + return db, lines +} + +// findLine returns the first of lines logged with message. +func findLine( + t *testing.T, + lines []map[string]any, + message string, +) map[string]any { + t.Helper() + + for _, line := range lines { + if line[slog.MessageKey] == message { + return line + } + } + + t.Fatalf("no line %q in the log", message) + + return nil +} + // 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.