Checkpoint without waiting on readers and log a busy checkpoint (closes #102)

The 5-second checkpoint ran in TRUNCATE mode, which waits up to the
5-second busy timeout for every open read while holding the database
lock every batch write needs. It now runs in PASSIVE mode, which never
waits. A WAL file over 64 MiB instead gets one TRUNCATE checkpoint whose
wait for reads is capped at 500 ms and logged; it empties the file once
no read is open, and while reads follow one another with no pause it
can give up every time. A checkpoint that leaves part of the WAL
uncopied returns ErrCheckpointIncomplete, which the maintainer logs as
incomplete instead of completed. The writing connection's
journal_size_limit is 64 MiB.

Model: opus-5-5

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{},
@@ -2241,9 +2271,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()
@@ -2253,11 +2298,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
 }
 
@@ -2273,6 +2339,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 (