fix(sync): snapshot WAL into memory to prevent checkpoint race

Instead of buffering each page individually (which caused severe
performance regression due to excessive allocations), snapshot the
entire WAL file into memory once at the start of sync. Both PageMap
and the page encoding step read from this immutable in-memory copy,
eliminating the TOCTOU race where a concurrent checkpoint could
rewrite the WAL between the initial read and the encoding step.

This approach uses the same memory as the OS page cache would
(one copy of the WAL), but the copy is immutable and immune to
concurrent modification.

Fixes #1164

Author: corylanou

db.go

@@ -1480,32 +1480,36 @@ func (db *DB) sync(ctx context.Context, checkpointing bool, info syncInfo) (sync
 	mode := fi.Mode()
 	commit := uint32(fi.Size() / int64(db.pageSize))
 
-	walFile, err := os.Open(db.WALPath())
+	// Snapshot WAL file into memory to prevent TOCTOU races with concurrent
+	// checkpoints. The application's SQLite driver can checkpoint (rewriting
+	// the WAL) between our initial read and the page encoding step. Reading
+	// the entire WAL upfront gives us an immutable copy to work from.
+	walData, err := os.ReadFile(db.WALPath())
 	if err != nil {
 		return false, err
 	}
-	defer walFile.Close()
+	walReader := bytes.NewReader(walData)
 
 	var rd *WALReader
 	if info.offset == WALHeaderSize {
-		if rd, err = NewWALReader(walFile, db.Logger); err != nil {
+		if rd, err = NewWALReader(walReader, db.Logger); err != nil {
 			return false, fmt.Errorf("new wal reader: %w", err)
 		}
 	} else {
 		// If we cannot verify the previous frame
 		var pfmError *PrevFrameMismatchError
-		if rd, err = NewWALReaderWithOffset(ctx, walFile, info.offset, info.salt1, info.salt2, db.Logger); errors.As(err, &pfmError) {
+		if rd, err = NewWALReaderWithOffset(ctx, walReader, info.offset, info.salt1, info.salt2, db.Logger); errors.As(err, &pfmError) {
 			db.Logger.Log(ctx, internal.LevelTrace, "prev frame mismatch, snapshotting", "err", pfmError.Err)
 			info.offset = WALHeaderSize
-			if rd, err = NewWALReader(walFile, db.Logger); err != nil {
+			if rd, err = NewWALReader(walReader, db.Logger); err != nil {
 				return false, fmt.Errorf("new wal reader, after reset")
 			}
 		} else if err != nil {
 			return false, fmt.Errorf("new wal reader with offset: %w", err)
 		}
 	}
 
-	// Build a mapping of changed page numbers and their latest content.
+	// Build a mapping of changed page numbers and their offsets.
 	pageMap, maxOffset, walCommit, err := rd.PageMap(ctx)
 	if err != nil {
 		return false, fmt.Errorf("page map: %w", err)
@@ -1578,11 +1582,11 @@ func (db *DB) sync(ctx context.Context, checkpointing bool, info syncInfo) (sync
 	// If we need a full snapshot, then copy from the database & WAL.
 	// Otherwise, just copy incrementally from the WAL.
 	if info.snapshotting {
-		if err := db.writeLTXFromDB(ctx, enc, walFile, commit, pageMap); err != nil {
+		if err := db.writeLTXFromDB(ctx, enc, walReader, commit, pageMap); err != nil {
 			return false, fmt.Errorf("write ltx from db: %w", err)
 		}
 	} else {
-		if err := db.writeLTXFromWAL(ctx, enc, walFile, pageMap); err != nil {
+		if err := db.writeLTXFromWAL(ctx, enc, walReader, pageMap); err != nil {
 			return false, fmt.Errorf("write ltx from wal: %w", err)
 		}
 	}
@@ -1640,7 +1644,7 @@ func (db *DB) sync(ctx context.Context, checkpointing bool, info syncInfo) (sync
 	return true, nil
 }
 
-func (db *DB) writeLTXFromDB(ctx context.Context, enc *ltx.Encoder, walFile *os.File, commit uint32, pageMap map[uint32]int64) error {
+func (db *DB) writeLTXFromDB(ctx context.Context, enc *ltx.Encoder, walReader io.ReaderAt, commit uint32, pageMap map[uint32]int64) error {
 	lockPgno := ltx.LockPgno(uint32(db.pageSize))
 	data := make([]byte, db.pageSize)
 
@@ -1649,7 +1653,6 @@ func (db *DB) writeLTXFromDB(ctx context.Context, enc *ltx.Encoder, walFile *os.
 			continue
 		}
 
-		// Check if the caller has canceled during processing.
 		select {
 		case <-ctx.Done():
 			return context.Cause(ctx)
@@ -1660,7 +1663,7 @@ func (db *DB) writeLTXFromDB(ctx context.Context, enc *ltx.Encoder, walFile *os.
 		if offset, ok := pageMap[pgno]; ok {
 			db.Logger.Log(ctx, internal.LevelTrace, "encode page from wal", "txid", enc.Header().MinTXID, "offset", offset, "pgno", pgno, "type", "db+wal")
 
-			if n, err := walFile.ReadAt(data, offset+WALFrameHeaderSize); err != nil {
+			if n, err := walReader.ReadAt(data, offset+WALFrameHeaderSize); err != nil {
 				return fmt.Errorf("read page %d @ %d: %w", pgno, offset, err)
 			} else if n != len(data) {
 				return fmt.Errorf("short read page %d @ %d", pgno, offset)
@@ -1675,7 +1678,6 @@ func (db *DB) writeLTXFromDB(ctx context.Context, enc *ltx.Encoder, walFile *os.
 		offset := int64(pgno-1) * int64(db.pageSize)
 		db.Logger.Log(ctx, internal.LevelTrace, "encode page from database", "offset", offset, "pgno", pgno)
 
-		// Otherwise read directly from the database file.
 		if _, err := db.f.ReadAt(data, offset); err != nil {
 			return fmt.Errorf("read database page %d: %w", pgno, err)
 		}
@@ -1687,8 +1689,7 @@ func (db *DB) writeLTXFromDB(ctx context.Context, enc *ltx.Encoder, walFile *os.
 	return nil
 }
 
-func (db *DB) writeLTXFromWAL(ctx context.Context, enc *ltx.Encoder, walFile *os.File, pageMap map[uint32]int64) error {
-	// Create an ordered list of page numbers since the LTX encoder requires it.
+func (db *DB) writeLTXFromWAL(ctx context.Context, enc *ltx.Encoder, walReader io.ReaderAt, pageMap map[uint32]int64) error {
 	pgnos := make([]uint32, 0, len(pageMap))
 	for pgno := range pageMap {
 		pgnos = append(pgnos, pgno)
@@ -1701,14 +1702,12 @@ func (db *DB) writeLTXFromWAL(ctx context.Context, enc *ltx.Encoder, walFile *os
 
 		db.Logger.Log(ctx, internal.LevelTrace, "encode page from wal", "txid", enc.Header().MinTXID, "offset", offset, "pgno", pgno, "type", "walonly")
 
-		// Read source page using page map.
-		if n, err := walFile.ReadAt(data, offset+WALFrameHeaderSize); err != nil {
+		if n, err := walReader.ReadAt(data, offset+WALFrameHeaderSize); err != nil {
 			return fmt.Errorf("read page %d @ %d: %w", pgno, offset, err)
 		} else if n != len(data) {
 			return fmt.Errorf("short read page %d @ %d", pgno, offset)
 		}
 
-		// Write page to LTX encoder.
 		if err := enc.EncodePage(ltx.PageHeader{Pgno: pgno}, data); err != nil {
 			return fmt.Errorf("encode ltx frame (pgno=%d): %w", pgno, err)
 		}
@@ -1867,20 +1866,21 @@ func (db *DB) SnapshotReader(ctx context.Context) (ltx.Pos, io.Reader, error) {
 	// Execute encoding in a separate goroutine so the caller can initialize before reading.
 	pr, pw := io.Pipe()
 	go func() {
-		walFile, err := os.Open(db.WALPath())
+		// Snapshot WAL into memory to prevent TOCTOU races with concurrent checkpoints.
+		walData, err := os.ReadFile(db.WALPath())
 		if err != nil {
 			pw.CloseWithError(err)
 			return
 		}
-		defer walFile.Close()
+		walReader := bytes.NewReader(walData)
 
-		rd, err := NewWALReader(walFile, db.Logger)
+		rd, err := NewWALReader(walReader, db.Logger)
 		if err != nil {
 			pw.CloseWithError(fmt.Errorf("new wal reader: %w", err))
 			return
 		}
 
-		// Build a mapping of changed page numbers and their latest content.
+		// Build a mapping of changed page numbers and their offsets.
 		pageMap, maxOffset, walCommit, err := rd.PageMap(ctx)
 		if err != nil {
 			pw.CloseWithError(fmt.Errorf("page map: %w", err))
@@ -1925,7 +1925,7 @@ func (db *DB) SnapshotReader(ctx context.Context) (ltx.Pos, io.Reader, error) {
 			return
 		}
 
-		if err := db.writeLTXFromDB(ctx, enc, walFile, commit, pageMap); err != nil {
+		if err := db.writeLTXFromDB(ctx, enc, walReader, commit, pageMap); err != nil {
 			pw.CloseWithError(fmt.Errorf("write snapshot ltx: %w", err))
 			return
 		}