diff --git a/internal/database/database.go b/internal/database/database.go index 248acee..a536e77 100644 --- a/internal/database/database.go +++ b/internal/database/database.go @@ -68,6 +68,18 @@ const ( sqliteHardHeapLimitBytes = 1610612736 ) +// Settings of the ANALYZE that Analyze runs. +const ( + // analysisLimit is about how many rows of each index ANALYZE reads + // (PRAGMA analysis_limit); SQLite estimates the rest. + analysisLimit = 1000 + // optimizeAllTables is the PRAGMA optimize mask that runs ANALYZE (0x2) + // on every table that has an index without statistics or whose size has + // changed tenfold since it was last analyzed, not only on the tables this + // connection has queried (0x10000). + optimizeAllTables = 0x10002 +) + // Bounds on the work GetRandomPrefixesByLength does to pick its sample. const ( // maxPrefixesToShuffle is the most prefixes of one mask length that are @@ -182,7 +194,6 @@ func (d *Database) Initialize() error { // enough for the whole pool. pragmas := []string{ "PRAGMA journal_mode=WAL", // Write-Ahead Logging - "PRAGMA analysis_limit=0", // Disable automatic ANALYZE fmt.Sprintf("PRAGMA soft_heap_limit=%d", sqliteSoftHeapLimitBytes), fmt.Sprintf("PRAGMA hard_heap_limit=%d", sqliteHardHeapLimitBytes), } @@ -2151,14 +2162,18 @@ func (d *Database) GetIPInfoContext( return info, nil } -// Vacuum runs incremental vacuum to reclaim unused pages without blocking -// writes. It frees up to the specified number of pages per call (0 = all -// freeable pages). +// Vacuum runs incremental vacuum to return up to 1000 unused pages to the +// filesystem. Like Analyze and Checkpoint, it holds the database lock, so it +// and the batch writes wait for each other instead of failing on SQLite's +// write lock. func (d *Database) Vacuum(ctx context.Context) error { // Free up to 1000 pages per call (~4MB with default 4KB page size) - // This keeps each vacuum operation quick and non-blocking + // This keeps each vacuum operation quick const pagesToFree = 1000 + d.lock("Vacuum") + defer d.unlock() + rows, err := d.db.QueryContext(ctx, fmt.Sprintf("PRAGMA incremental_vacuum(%d)", pagesToFree)) if err != nil { @@ -2180,9 +2195,20 @@ func (d *Database) Vacuum(ctx context.Context) error { return nil } -// Analyze runs the SQLite ANALYZE command to update query planner statistics. +// Analyze updates the query planner statistics. It runs ANALYZE, through +// PRAGMA optimize, only on the tables that have an index without statistics or +// whose size has changed tenfold since they were last analyzed: even with +// analysisLimit, ANALYZE counts every entry of each index, so it takes longer +// as the tables grow. It holds the database lock. func (d *Database) Analyze(ctx context.Context) error { - _, err := d.db.ExecContext(ctx, "ANALYZE") + d.lock("Analyze") + defer d.unlock() + + // analysis_limit is a setting of the connection, so it is set in the same + // call as PRAGMA optimize, which runs both statements on one connection. + _, err := d.db.ExecContext(ctx, fmt.Sprintf( + "PRAGMA analysis_limit=%d; PRAGMA optimize=%#x", + analysisLimit, optimizeAllTables)) if err != nil { return fmt.Errorf("failed to analyze database: %w", err) } @@ -2192,8 +2218,11 @@ func (d *Database) Analyze(ctx context.Context) error { // 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. +// keeping the WAL small for fast read performance. It holds the database lock. func (d *Database) Checkpoint(ctx context.Context) error { + d.lock("Checkpoint") + defer d.unlock() + _, err := d.db.ExecContext(ctx, "PRAGMA wal_checkpoint(TRUNCATE)") if err != nil { return fmt.Errorf("failed to checkpoint WAL: %w", err) @@ -2214,21 +2243,26 @@ func (d *Database) Ping(ctx context.Context) error { 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 ( + lockWaitMessage = "Acquiring database lock" + lockHeldMessage = "Database lock acquired" +) + // lock acquires the database mutex and logs debug information func (d *Database) lock(operation string) { // Get caller information _, file, line, _ := runtime.Caller(1) caller := fmt.Sprintf("%s:%d", filepath.Base(file), line) - d.logger.Debug("Acquiring database lock", - "operation", operation, "caller", caller) + d.logger.Debug(lockWaitMessage, "operation", operation, "caller", caller) d.mu.Lock() d.lockedAt = time.Now() d.lockedBy = fmt.Sprintf("%s (%s)", operation, caller) - d.logger.Debug("Database lock acquired", - "operation", operation, "caller", caller) + d.logger.Debug(lockHeldMessage, "operation", operation, "caller", caller) } // unlock releases the database mutex and logs debug information including diff --git a/internal/database/database_internal_test.go b/internal/database/database_internal_test.go index f2b2e63..d93b188 100644 --- a/internal/database/database_internal_test.go +++ b/internal/database/database_internal_test.go @@ -5,6 +5,7 @@ import ( "database/sql" "errors" "fmt" + "log/slog" "net/netip" "slices" "sync" @@ -47,6 +48,23 @@ const ( asnSecondBand = 100 ) +// Parameters for the tests of maintenance and the database lock. +const ( + // lockMessageBuffer is how many lock messages a test database's logger + // holds before logging waits for the test to read them. + lockMessageBuffer = 64 + // operationKey is the log attribute that names the operation taking the + // lock. + operationKey = "operation" + // testBatchWrite is the operation name the tests take the lock under. + testBatchWrite = "test batch write" + // lockWaitTimeout is the context timeout of a maintenance call that the + // tests let run out while it waits for the lock. + lockWaitTimeout = time.Millisecond + // lockTestASN is the AS the batch write in the lock tests stores. + lockTestASN = 64500 +) + // Parameters for the prefix-length sample tests. const ( // sampleLimit is how many prefixes the tests ask the sample for. @@ -474,14 +492,10 @@ func TestVacuumReturnsFreePages(t *testing.T) { } } -// TestBatchWriteDuringCheckpoint reproduces issue #25. A batch write reads -// (SELECT) before it writes (INSERT/UPDATE). Under the default deferred locking -// the transaction begins as a reader and, when the maintainer's WAL checkpoint -// holds the write lock, its upgrade to writer fails immediately with "database -// is locked" without honouring busy_timeout, dropping the batch. With -// _txlock=immediate the transaction takes the write lock at BEGIN and waits, so -// no batch is dropped. The checkpoint runs without the Database mutex, exactly -// as the background maintainer does in production. +// TestBatchWriteDuringCheckpoint runs checkpoints back to back while batch +// writes run and checks that every batch write succeeds: a checkpoint holds the +// database lock as a batch write does, and checkpoints must not keep batch +// writes from getting it. func TestBatchWriteDuringCheckpoint(t *testing.T) { t.Parallel() @@ -530,6 +544,210 @@ func TestBatchWriteDuringCheckpoint(t *testing.T) { wg.Wait() } +// lockMessage is a debug message the database lock logged, and the name of the +// operation that logged it. +type lockMessage struct { + text string + operation string +} + +// lockMessages is a log handler that passes on each message the database lock +// logs. The lock tests read it to know that an operation waits for the lock or +// holds it, instead of guessing how long that takes. +type lockMessages chan lockMessage + +// Enabled lets every message through, so the lock's debug messages arrive. +func (m lockMessages) Enabled(context.Context, slog.Level) bool { return true } + +// Handle passes on a message that names an operation and drops the others. +func (m lockMessages) Handle(_ context.Context, record slog.Record) error { + record.Attrs(func(attr slog.Attr) bool { + if attr.Key != operationKey { + return true + } + + m <- lockMessage{text: record.Message, operation: attr.Value.String()} + + return false + }) + + return nil +} + +// WithAttrs returns m: the tests need no attributes added. +func (m lockMessages) WithAttrs([]slog.Attr) slog.Handler { return m } + +// WithGroup returns m: the tests need no groups. +func (m lockMessages) WithGroup(string) slog.Handler { return m } + +// newDatabaseWithLockMessages opens a new database whose logger passes the lock's +// messages to the returned channel. +func newDatabaseWithLockMessages(t *testing.T) (*Database, lockMessages) { + t.Helper() + + messages := make(lockMessages, lockMessageBuffer) + cfg := &config.Config{StateDir: t.TempDir()} + + db, err := New(cfg, &logger.Logger{Logger: slog.New(messages)}) + if err != nil { + t.Fatalf("failed to create database: %v", err) + } + + t.Cleanup(func() { _ = db.Close() }) + + return db, messages +} + +// waitForLockMessage waits until operation logs text, and fails the test if +// returned delivers first: the call running operation returned without logging +// text. +func waitForLockMessage( + t *testing.T, + messages lockMessages, + text, operation string, + returned <-chan error, +) { + t.Helper() + + for { + select { + case message := <-messages: + if message.text == text && message.operation == operation { + return + } + case err := <-returned: + t.Fatalf("%s returned (error: %v) without logging %q", + operation, err, text) + } + } +} + +// TestVacuumWaitsForBatchWrite holds the database lock as a batch write does and +// checks that Vacuum waits for it and then succeeds. Vacuum ran without the +// lock, met the batch writes at SQLite's write lock and failed with "database +// is locked" (https://git.eeqj.de/sneak/routewatch/issues/94). +func TestVacuumWaitsForBatchWrite(t *testing.T) { + t.Parallel() + + db, messages := newDatabaseWithLockMessages(t) + + db.lock(testBatchWrite) + + vacuumed := make(chan error, 1) + + go func() { vacuumed <- db.Vacuum(context.Background()) }() + + waitForLockMessage(t, messages, lockWaitMessage, "Vacuum", vacuumed) + + db.unlock() + + err := <-vacuumed + if err != nil { + t.Fatalf("Vacuum after the batch write: %v", err) + } +} + +// TestBatchWriteWaitsForAnalyze keeps Analyze waiting for SQLite's write lock, +// which another connection holds, and checks that a batch write started +// meanwhile waits for Analyze to give up the database lock, that both then +// succeed, and that Analyze stored statistics. ANALYZE ran without the database +// lock, and a batch write that waited longer than the busy timeout for it failed +// with "database is locked". +func TestBatchWriteWaitsForAnalyze(t *testing.T) { + t.Parallel() + + db, messages := newDatabaseWithLockMessages(t) + ctx := context.Background() + asns := map[int]time.Time{lockTestASN: time.Now().UTC()} + + // A stored AS gives ANALYZE a table to analyze and statistics to write. + err := db.GetOrCreateASNBatch(asns) + if err != nil { + t.Fatalf("failed to store an AS: %v", err) + } + + // With _txlock=immediate this transaction takes SQLite's write lock at + // BEGIN, so ANALYZE has to wait for it while Analyze holds the database lock. + writer, err := db.db.BeginTx(ctx, nil) + if err != nil { + t.Fatalf("failed to begin the transaction ANALYZE waits for: %v", err) + } + + defer func() { _ = writer.Rollback() }() + + analyzed := make(chan error, 1) + + go func() { analyzed <- db.Analyze(ctx) }() + + waitForLockMessage(t, messages, lockHeldMessage, "Analyze", analyzed) + + written := make(chan error, 1) + + go func() { written <- db.GetOrCreateASNBatch(asns) }() + + waitForLockMessage(t, messages, lockWaitMessage, "GetOrCreateASNBatch", written) + + err = writer.Rollback() + if err != nil { + t.Fatalf("failed to end the transaction ANALYZE waits for: %v", err) + } + + err = <-analyzed + if err != nil { + t.Errorf("Analyze: %v", err) + } + + err = <-written + if err != nil { + t.Errorf("batch write after Analyze: %v", err) + } + + var statistics int + + err = db.db.QueryRowContext(ctx, + "SELECT count(*) FROM sqlite_stat1 WHERE tbl = 'asns'").Scan(&statistics) + if err != nil { + t.Fatalf("failed to read the statistics: %v", err) + } + + if statistics == 0 { + t.Error("Analyze stored no statistics for the asns table") + } +} + +// TestCheckpointLockWaitEndsWithContext holds the database lock while the +// context of a waiting Checkpoint times out, and checks that Checkpoint then +// returns the context's error and leaves the lock free. +func TestCheckpointLockWaitEndsWithContext(t *testing.T) { + t.Parallel() + + db, messages := newDatabaseWithLockMessages(t) + + db.lock(testBatchWrite) + + ctx, cancel := context.WithTimeout(context.Background(), lockWaitTimeout) + defer cancel() + + checkpointed := make(chan error, 1) + + go func() { checkpointed <- db.Checkpoint(ctx) }() + + waitForLockMessage(t, messages, lockWaitMessage, "Checkpoint", checkpointed) + <-ctx.Done() + db.unlock() + + err := <-checkpointed + if !errors.Is(err, context.DeadlineExceeded) { + t.Fatalf("Checkpoint = %v, want %v", err, context.DeadlineExceeded) + } + + if !db.mu.TryLock() { + t.Fatal("Checkpoint kept the database lock after its context ended") + } + + db.mu.Unlock() +} + // 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.