diff --git a/internal/database/database.go b/internal/database/database.go index c4832bb..0ac9e32 100644 --- a/internal/database/database.go +++ b/internal/database/database.go @@ -81,6 +81,11 @@ const ( sqliteHardHeapLimitBytes = 2147483648 ) +// sqliteWALSizeLimitBytes (64 MiB) is 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. +const sqliteWALSizeLimitBytes = 67108864 + // Settings of the ANALYZE that Analyze runs. const ( // analysisLimit is about how many rows of each index ANALYZE reads @@ -112,6 +117,9 @@ var ( ErrNoRoute = errors.New("no route found") // ErrNoStaleASN is returned when no ASN needs WHOIS refresh ErrNoStaleASN = errors.New("no stale ASN found") + // ErrCheckpointBusy is returned by Checkpoint when SQLite reports the + // checkpoint busy: another connection was checkpointing at the same time. + ErrCheckpointBusy = errors.New("WAL checkpoint busy") errInvalidASN = errors.New("invalid ASN") errSameASNInPeering = errors.New("cannot create peering with same ASN") @@ -194,6 +202,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{}, @@ -2245,9 +2260,13 @@ 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 it runs a PASSIVE checkpoint, which never waits: it copies what no reader +// still needs and leaves the rest for the next one. TRUNCATE or RESTART would +// wait up to the busy timeout for every reader to finish. The WAL file shrinks +// instead when SQLite starts it again from the beginning +// (sqliteWALSizeLimitBytes). It returns ErrCheckpointBusy when SQLite reports +// the checkpoint busy. func (d *Database) Checkpoint(ctx context.Context) error { d.lock("Checkpoint") defer d.unlock() @@ -2257,11 +2276,18 @@ 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)") + var busy, walPages, checkpointedPages int + + err = d.writer.QueryRowContext(ctx, "PRAGMA wal_checkpoint(PASSIVE)"). + Scan(&busy, &walPages, &checkpointedPages) if err != nil { return fmt.Errorf("failed to checkpoint WAL: %w", err) } + if busy != 0 { + return ErrCheckpointBusy + } + return nil } diff --git a/internal/database/database_internal_test.go b/internal/database/database_internal_test.go index 7a48559..7cc281b 100644 --- a/internal/database/database_internal_test.go +++ b/internal/database/database_internal_test.go @@ -67,6 +67,10 @@ 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 ) // Parameters for the prefix-length sample tests. @@ -457,6 +461,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: Checkpoint does +// not truncate the WAL file, so the limit is what shrinks it. +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 +830,90 @@ func TestCheckpointLockWaitEndsWithContext(t *testing.T) { db.mu.Unlock() } +// TestCheckpointDoesNotWaitForReader keeps a read transaction on +// live_routes_v4 open while a route is committed, and checks that Checkpoint, +// and a batch write started while Checkpoint holds the database lock, both end +// within readerCheckpointLimit. 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, messages := newDatabaseWithLockMessages(t) + 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) }() + + waitForLockMessage(t, messages, lockHeldMessage, "Checkpoint", checkpointed) + + written := make(chan error, 1) + + go func() { + written <- db.GetOrCreateASNBatch(map[int]time.Time{lockTestASN: ts}) + }() + + err = <-checkpointed + if err != nil { + t.Errorf("Checkpoint: %v", err) + } + + 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) + } +} + // 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..ac9df99 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,9 @@ const ( analyzeTimeout = 5 * time.Minute ) +// checkpointBusyMessage is logged when SQLite reports a checkpoint busy. +const checkpointBusyMessage = "WAL checkpoint busy" + // DBMaintainer handles background database maintenance tasks. type DBMaintainer struct { db database.Store @@ -123,9 +127,12 @@ 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.ErrCheckpointBusy): + m.logger.Warn(checkpointBusyMessage, "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..39242a2 --- /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" +) + +// busyCheckpointStore is a store whose every checkpoint SQLite reports busy. +type busyCheckpointStore struct{ *mockStore } + +// Checkpoint reports the checkpoint busy. +func (busyCheckpointStore) Checkpoint(context.Context) error { + return database.ErrCheckpointBusy +} + +// TestBusyCheckpointLoggedAsBusy runs a checkpoint that SQLite reports busy +// and checks that the maintainer logs it as busy, at warning level, and does +// not count it as a checkpoint that completed. +func TestBusyCheckpointLoggedAsBusy(t *testing.T) { + t.Parallel() + + var logs bytes.Buffer + + logger := slog.New(slog.NewJSONHandler(&logs, + &slog.HandlerOptions{Level: slog.LevelDebug})) + m := NewDBMaintainer(busyCheckpointStore{newMockStore()}, logger) + + m.runCheckpoint(t.Context()) + + var logged struct { + Level string `json:"level"` + 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.Level != slog.LevelWarn.String() || logged.Msg != checkpointBusyMessage { + t.Errorf("logged %s %q, want %s %q", logged.Level, logged.Msg, + slog.LevelWarn, checkpointBusyMessage) + } + + if m.checkpointCount != 0 { + t.Errorf("checkpointCount = %d after a busy checkpoint, want 0", + m.checkpointCount) + } +}