fix(wal): stop snapshot WAL reads at the advertised end offset

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

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.

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