diff --git a/db.go b/db.go index 974e2debd..8ac6ee081 100644 --- a/db.go +++ b/db.go @@ -2077,7 +2077,7 @@ func (db *DB) sync(ctx context.Context, checkpointing bool, exec *syncExecutor, if info.snapshotting { maxSyncWALBytes = 0 } - pageMap, maxOffset, walCommit, limited, err := rd.pageMap(ctx, maxSyncWALBytes) + pageMap, maxOffset, walCommit, limited, err := rd.pageMap(ctx, maxSyncWALBytes, 0) if err != nil { return result, fmt.Errorf("page map: %w", err) } @@ -2829,12 +2829,14 @@ func (db *DB) snapshotReader(ctx context.Context, pos *snapshotReadPosition) (io } // Build a mapping of changed page numbers and their latest content. - maxBytes := pos.walEndOffset - WALHeaderSize + // The read is hard-bounded to the advertised WAL end so a transaction + // that straddles the bound is discarded rather than read through to a + // commit frame past the bound (#1490). pageMap := make(map[uint32]int64) var maxOffset int64 var walCommit uint32 - if maxBytes > 0 { - pageMap, maxOffset, walCommit, _, err = rd.pageMap(ctx, maxBytes) + if pos.walEndOffset > WALHeaderSize { + pageMap, maxOffset, walCommit, _, err = rd.pageMap(ctx, 0, pos.walEndOffset) if err != nil { pw.CloseWithError(fmt.Errorf("page map: %w", err)) return diff --git a/db_internal_test.go b/db_internal_test.go index d7f1b4a8c..10697d361 100644 --- a/db_internal_test.go +++ b/db_internal_test.go @@ -1029,7 +1029,7 @@ func TestWALReaderPageMapLimitStopsAtCommittedFrame(t *testing.T) { t.Fatal(err) } - pageMap, maxOffset, commit, limited, err := r.pageMap(context.Background(), 1) + pageMap, maxOffset, commit, limited, err := r.pageMap(context.Background(), 1, 0) if err != nil { t.Fatal(err) } @@ -1047,6 +1047,170 @@ func TestWALReaderPageMapLimitStopsAtCommittedFrame(t *testing.T) { } } +// TestWALReaderPageMapEndOffsetStopsBeforeStraddlingTransaction verifies that +// a hard end offset is never read past, even when a transaction straddles it. +// Regression test for #1490: the commit-frame break used to read a straddling +// transaction through to its commit frame past the bound, which snapshotReader +// then rejected with "snapshot wal read exceeded bound". +func TestWALReaderPageMapEndOffsetStopsBeforeStraddlingTransaction(t *testing.T) { + // The fixture WAL holds two transactions: frames at offsets 32 and 4152 + // committing at offset 8272, and one frame at offset 8272 committing at + // offset 12392. + b, err := os.ReadFile("testdata/wal-reader/ok/wal") + if err != nil { + t.Fatal(err) + } + + t.Run("MidTransaction", func(t *testing.T) { + r, err := NewWALReader(bytes.NewReader(b), slog.Default()) + if err != nil { + t.Fatal(err) + } + + // Bound inside the first transaction: the partial transaction must be + // discarded, not read through to its commit frame at offset 8272. + pageMap, maxOffset, commit, limited, err := r.pageMap(context.Background(), 0, 4152) + if err != nil { + t.Fatal(err) + } + if !limited { + t.Fatal("expected page map to stop at end offset") + } + if got, want := maxOffset, int64(0); got != want { + t.Fatalf("maxOffset=%d, want %d", got, want) + } + if got, want := commit, uint32(0); got != want { + t.Fatalf("commit=%d, want %d", got, want) + } + if got, want := len(pageMap), 0; got != want { + t.Fatalf("len(pageMap)=%d, want %d", got, want) + } + }) + + t.Run("AtCommitBoundary", func(t *testing.T) { + r, err := NewWALReader(bytes.NewReader(b), slog.Default()) + if err != nil { + t.Fatal(err) + } + + // Bound exactly at the first transaction's commit frame end: the + // transaction must still be included. + pageMap, maxOffset, commit, limited, err := r.pageMap(context.Background(), 0, 8272) + if err != nil { + t.Fatal(err) + } + if !limited { + t.Fatal("expected page map to stop at end offset") + } + if got, want := maxOffset, int64(8272); got != want { + t.Fatalf("maxOffset=%d, want %d", got, want) + } + if got, want := commit, uint32(2); got != want { + t.Fatalf("commit=%d, want %d", got, want) + } + if got, want := len(pageMap), 2; got != want { + t.Fatalf("len(pageMap)=%d, want %d", got, want) + } + }) +} + +// TestDB_SnapshotReaderWALEndOffsetMidTransaction verifies that a snapshot +// succeeds when the advertised WAL end offset lands inside a transaction. +// Regression test for #1490: the snapshot WAL scan used to read the straddling +// transaction through to its commit frame and then permanently fail with +// "snapshot wal read exceeded bound". +func TestDB_SnapshotReaderWALEndOffsetMidTransaction(t *testing.T) { + dir := t.TempDir() + dbPath := filepath.Join(dir, "db") + + db := NewDB(dbPath) + db.MonitorInterval = 0 + db.Replica = NewReplica(db) + db.Replica.Client = &testReplicaClient{dir: t.TempDir()} + db.Replica.MonitorEnabled = false + if err := db.Open(); err != nil { + t.Fatal(err) + } + defer func() { + if err := db.Close(context.Background()); err != nil { + t.Fatal(err) + } + }() + + sqldb, err := sql.Open("sqlite", dbPath) + if err != nil { + t.Fatal(err) + } + defer sqldb.Close() + + if _, err := sqldb.Exec(`PRAGMA journal_mode = wal;`); err != nil { + t.Fatal(err) + } + if _, err := sqldb.Exec(`CREATE TABLE t (id INTEGER PRIMARY KEY, data BLOB);`); err != nil { + t.Fatal(err) + } + if _, err := sqldb.Exec(`INSERT INTO t(data) VALUES (zeroblob(100));`); err != nil { + t.Fatal(err) + } + + // End the WAL with a transaction spanning multiple frames. + if _, err := sqldb.Exec(`INSERT INTO t(data) VALUES (zeroblob(8000));`); err != nil { + t.Fatal(err) + } + + if err := db.Sync(t.Context()); err != nil { + t.Fatal(err) + } + + // Rewind the advertised WAL end by one frame so it lands inside the final + // multi-frame transaction, as a stale lastSyncedWALOffset can after a WAL + // restart. + frameSize := int64(WALFrameHeaderSize) + int64(db.PageSize()) + db.mu.Lock() + db.syncState.lastSyncedWALOffset -= frameSize + walEndOffset := db.syncState.lastSyncedWALOffset + db.mu.Unlock() + + wal, err := os.ReadFile(db.WALPath()) + if err != nil { + t.Fatal(err) + } + if commit := binary.BigEndian.Uint32(wal[walEndOffset-frameSize+4:]); commit != 0 { + t.Fatalf("precondition: wal end offset %d must land inside a transaction", walEndOffset) + } + + _, r, err := db.SnapshotReader(context.Background()) + if err != nil { + t.Fatal(err) + } + defer r.Close() + + var buf bytes.Buffer + if _, err := io.Copy(&buf, r); err != nil { + t.Fatalf("read snapshot: %v", err) + } + + dec := ltx.NewDecoder(bytes.NewReader(buf.Bytes())) + if err := dec.DecodeHeader(); err != nil { + t.Fatal(err) + } + if got := dec.Header().WALOffset + dec.Header().WALSize; got > walEndOffset { + t.Fatalf("snapshot wal range end=%d exceeds advertised wal end offset %d", got, walEndOffset) + } + pageBuf := make([]byte, db.PageSize()) + for { + var phdr ltx.PageHeader + if err := dec.DecodePage(&phdr, pageBuf); err == io.EOF { + break + } else if err != nil { + t.Fatal(err) + } + } + if err := dec.Close(); err != nil { + t.Fatal(err) + } +} + // TestCalcWALSize ensures calcWALSize doesn't overflow with large page sizes. // Regression test for uint32 overflow bug where large page sizes (>=16KB) // caused incorrect WAL size calculations, triggering checkpoints too early. diff --git a/wal_reader.go b/wal_reader.go index e323c7d1b..5b6a3f307 100644 --- a/wal_reader.go +++ b/wal_reader.go @@ -200,17 +200,34 @@ func (r *WALReader) readFrame(ctx context.Context, data []byte, verifyChecksum b // map of pgno to offset of the latest version of each page. Also returns the // max offset of the wal segment read, and the final database size, in pages. func (r *WALReader) PageMap(ctx context.Context) (m map[uint32]int64, maxOffset int64, commit uint32, err error) { - m, maxOffset, commit, _, err = r.pageMap(ctx, 0) + m, maxOffset, commit, _, err = r.pageMap(ctx, 0, 0) return m, maxOffset, commit, err } -func (r *WALReader) pageMap(ctx context.Context, maxBytes int64) (m map[uint32]int64, maxOffset int64, commit uint32, limited bool, err error) { +// pageMap reads committed frames and returns a map of pgno to the offset of +// the latest version of each page, along with the max offset read and the +// final database size, in pages. +// +// maxBytes, when positive, soft-bounds the number of frame bytes read from +// the reader's starting position: the scan stops at the first commit frame +// at or past the bound, so a transaction that straddles the bound is still +// read through to its commit and at least one transaction is consumed. +// +// endOffset, when positive, hard-bounds the scan to frames that end at or +// before that WAL file offset: no frame past the bound is read, so a +// transaction whose commit frame lies past the bound is discarded entirely. +func (r *WALReader) pageMap(ctx context.Context, maxBytes, endOffset int64) (m map[uint32]int64, maxOffset int64, commit uint32, limited bool, err error) { m = make(map[uint32]int64) txMap := make(map[uint32]int64) data := make([]byte, r.pageSize) frameSize := int64(WALFrameHeaderSize + r.pageSize) startOffset := WALHeaderSize + int64(r.frameN)*frameSize for { + if endOffset > 0 && WALHeaderSize+(int64(r.frameN)+1)*frameSize > endOffset { + limited = true + break + } + pgno, fcommit, err := r.ReadFrame(ctx, data) if errors.Is(err, io.EOF) { break