diff --git a/.github/workflows/commit.yml b/.github/workflows/commit.yml index ff2216501..20ecbb148 100644 --- a/.github/workflows/commit.yml +++ b/.github/workflows/commit.yml @@ -62,6 +62,18 @@ jobs: - name: Verify shared library created run: file dist/litestream-vfs-linux-amd64.so + vfs-unit-test: + name: VFS Unit Test + runs-on: ubuntu-latest + steps: + - uses: actions/checkout@v4 + + - uses: actions/setup-go@v5 + with: + go-version-file: "go.mod" + + - run: go test -tags vfs -race -count=1 . + build-windows: name: Build Windows runs-on: ubuntu-latest diff --git a/vfs.go b/vfs.go index bf2a96d65..05f57ba08 100644 --- a/vfs.go +++ b/vfs.go @@ -515,18 +515,19 @@ type VFSFile struct { client ReplicaClient name string - pos ltx.Pos // Last TXID read from level 0 or 1 - maxTXID1 ltx.TXID // Last TXID read from level 1 - index map[uint32]ltx.PageIndexElem - pending map[uint32]ltx.PageIndexElem - pendingReplace bool - cache *lru.Cache[uint32, []byte] // LRU cache for page data - targetTime *time.Time // Target view time; nil means latest - latestLTXTime time.Time // Timestamp of most recent LTX file - lastPollSuccess time.Time // Time of last successful poll - lockType sqlite3vfs.LockType // Current lock state - pageSize uint32 - commit uint32 + pos ltx.Pos // Last TXID read from level 0 or 1 + maxTXID1 ltx.TXID // Last TXID read from level 1 + maxTXID1Anchored bool + index map[uint32]ltx.PageIndexElem + pending map[uint32]ltx.PageIndexElem + pendingReplace bool + cache *lru.Cache[uint32, []byte] // LRU cache for page data + targetTime *time.Time // Target view time; nil means latest + latestLTXTime time.Time // Timestamp of most recent LTX file + lastPollSuccess time.Time // Time of last successful poll + lockType sqlite3vfs.LockType // Current lock state + pageSize uint32 + commit uint32 // Write support fields (only used when writeEnabled is true) writeEnabled bool // Whether write support is enabled @@ -1209,6 +1210,7 @@ func (f *VFSFile) rebuildIndex(ctx context.Context, infos []*ltx.FileInfo, targe } maxTXID1 := maxLevelTXID(infos, 1) + maxTXID1Anchored := maxTXID1 != 0 // Seed maxTXID1 from pos when there are no L1 files if maxTXID1 == 0 { maxTXID1 = pos.TXID @@ -1221,6 +1223,7 @@ func (f *VFSFile) rebuildIndex(ctx context.Context, infos []*ltx.FileInfo, targe f.pendingReplace = false f.pos = pos f.maxTXID1 = maxTXID1 + f.maxTXID1Anchored = maxTXID1Anchored if len(infos) > 0 { f.latestLTXTime = infos[len(infos)-1].CreatedAt } @@ -1247,6 +1250,22 @@ func maxLevelTXID(infos []*ltx.FileInfo, level int) ltx.TXID { return maxTXID } +func mergePageIndexes(dst, baseline, src, applied map[uint32]ltx.PageIndexElem) { + for pgno, elem := range src { + current, ok := dst[pgno] + if !ok && baseline != nil { + current, ok = baseline[pgno] + } + if ok && (current.MaxTXID > elem.MaxTXID || (current.MaxTXID == elem.MaxTXID && current.Level >= elem.Level)) { + continue + } + dst[pgno] = elem + if applied != nil { + applied[pgno] = elem + } + } +} + // buildIndexMap constructs a lookup of pgno to LTX file offsets. func (f *VFSFile) buildIndexMap(ctx context.Context, infos []*ltx.FileInfo) (map[uint32]ltx.PageIndexElem, error) { index := make(map[uint32]ltx.PageIndexElem) @@ -1263,8 +1282,8 @@ func (f *VFSFile) buildIndexMap(ctx context.Context, infos []*ltx.FileInfo) (map // Replace pages in overall index with new pages. for k, v := range idx { f.logger.Debug("adding page index", "page", k, "elem", v) - index[k] = v } + mergePageIndexes(index, nil, idx, nil) hdr, err := FetchLTXHeader(ctx, f.client, info) if err != nil { return nil, fmt.Errorf("fetch header: %w", err) @@ -1911,13 +1930,18 @@ func (f *VFSFile) syncToRemoteWithLock() error { } // Create LTX file from dirty pages - ltxReader := f.createLTXFromDirty() + ltxReader, encodedCh := f.createLTXFromDirty() // Upload LTX file to remote info, err := f.client.WriteLTXFile(ctx, 0, f.pendingTXID, f.pendingTXID, ltxReader) if err != nil { + _ = ltxReader.CloseWithError(err) return fmt.Errorf("upload LTX: %w", err) } + encoded := <-encodedCh + if encoded.err != nil { + return fmt.Errorf("encode LTX: %w", encoded.err) + } f.logger.Info("synced to remote", "txid", info.MaxTXID, @@ -1936,7 +1960,9 @@ func (f *VFSFile) syncToRemoteWithLock() error { f.vfs.writeMu.Unlock() } - // Update cache with synced pages (index will be populated naturally when pages are fetched) + mergePageIndexes(f.index, nil, encoded.index, nil) + + // Update cache with synced pages for pgno, bufferOff := range f.dirty { cachedData := make([]byte, f.pageSize) if _, err := f.bufferFile.ReadAt(cachedData, bufferOff); err != nil { @@ -1999,12 +2025,18 @@ func (f *VFSFile) checkForConflict(ctx context.Context) error { return nil } +type encodedLTXResult struct { + index map[uint32]ltx.PageIndexElem + err error +} + // createLTXFromDirty creates an LTX file from dirty pages. // Returns a streaming reader for the LTX data using io.Pipe to avoid loading // all data into memory at once. // Must be called with f.mu held. -func (f *VFSFile) createLTXFromDirty() io.Reader { +func (f *VFSFile) createLTXFromDirty() (*io.PipeReader, <-chan encodedLTXResult) { pr, pw := io.Pipe() + resultCh := make(chan encodedLTXResult, 1) // Sort page numbers (LTX encoder requires ordered pages) pgnos := make([]uint32, 0, len(f.dirty)) @@ -2027,8 +2059,11 @@ func (f *VFSFile) createLTXFromDirty() io.Reader { go func() { var err error + index := make(map[uint32]ltx.PageIndexElem, len(pgnos)) defer func() { - pw.CloseWithError(err) + resultCh <- encodedLTXResult{index: index, err: err} + close(resultCh) + _ = pw.CloseWithError(err) }() enc, encErr := ltx.NewEncoder(pw) @@ -2066,10 +2101,18 @@ func (f *VFSFile) createLTXFromDirty() io.Reader { return } + offset := enc.N() if err = enc.EncodePage(ltx.PageHeader{Pgno: pgno}, data); err != nil { err = fmt.Errorf("encode page %d: %w", pgno, err) return } + index[pgno] = ltx.PageIndexElem{ + Level: 0, + MinTXID: pendingTXID, + MaxTXID: pendingTXID, + Offset: offset, + Size: enc.N() - offset, + } } // Close encoder (writes trailer and page index) @@ -2079,7 +2122,7 @@ func (f *VFSFile) createLTXFromDirty() io.Reader { } }() - return pr + return pr, resultCh } // initWriteBuffer initializes the write buffer file for durability. @@ -2264,8 +2307,8 @@ func (f *VFSFile) Unlock(elock sqlite3vfs.LockType) error { } else if len(f.pending) > 0 { // Merge pending into index count := len(f.pending) - for k, v := range f.pending { - f.index[k] = v + mergePageIndexes(f.index, nil, f.pending, nil) + for k := range f.pending { f.cache.Remove(k) } f.logger.Debug("cache invalidated pages", "count", count) @@ -2506,47 +2549,45 @@ func (f *VFSFile) pollReplicaClient(ctx context.Context) error { f.mu.Lock() baseCommit := f.commit maxTXID1Snapshot := f.maxTXID1 + maxTXID1AnchoredSnapshot := f.maxTXID1Anchored f.mu.Unlock() newCommit := baseCommit replaceIndex := false - maxTXID0, idx0, commit0, replace0, err := f.pollLevel(ctx, 0, pos.TXID, baseCommit) + level0, err := f.pollLevel(ctx, 0, pos.TXID, baseCommit, true) if err != nil { return fmt.Errorf("poll L0: %w", err) } - if replace0 { + if level0.replaceIndex { replaceIndex = true - baseCommit = commit0 - newCommit = commit0 - combined = idx0 + baseCommit = level0.commit + newCommit = level0.commit + combined = make(map[uint32]ltx.PageIndexElem) + mergePageIndexes(combined, nil, level0.index, nil) } else { - if len(idx0) > 0 { - baseCommit = commit0 - } - for k, v := range idx0 { - combined[k] = v + if len(level0.index) > 0 { + baseCommit = level0.commit } - if commit0 > newCommit { - newCommit = commit0 + mergePageIndexes(combined, nil, level0.index, nil) + if level0.commit > newCommit { + newCommit = level0.commit } } - maxTXID1, idx1, commit1, replace1, err := f.pollLevel(ctx, 1, maxTXID1Snapshot, baseCommit) + level1, err := f.pollLevel(ctx, 1, maxTXID1Snapshot, baseCommit, maxTXID1AnchoredSnapshot) if err != nil { return fmt.Errorf("poll L1: %w", err) } - if replace1 { + if level1.replaceIndex { replaceIndex = true - baseCommit = commit1 - newCommit = commit1 - combined = idx1 + newCommit = level1.commit + combined = make(map[uint32]ltx.PageIndexElem) + mergePageIndexes(combined, nil, level1.index, nil) } else { - for k, v := range idx1 { - combined[k] = v - } - if commit1 > newCommit { - newCommit = commit1 + mergePageIndexes(combined, nil, level1.index, nil) + if level1.commit > newCommit { + newCommit = level1.commit } } @@ -2563,9 +2604,11 @@ func (f *VFSFile) pollReplicaClient(ctx context.Context) error { // Apply updates and invalidate cache entries for updated pages invalidateN := 0 target := f.index + var baseline map[uint32]ltx.PageIndexElem targetIsMain := true if f.lockType >= sqlite3vfs.LockShared { target = f.pending + baseline = f.index targetIsMain = false } else { f.pendingReplace = false @@ -2574,17 +2617,20 @@ func (f *VFSFile) pollReplicaClient(ctx context.Context) error { if f.lockType < sqlite3vfs.LockShared { f.index = make(map[uint32]ltx.PageIndexElem) target = f.index + baseline = nil targetIsMain = true f.pendingReplace = false } else { f.pending = make(map[uint32]ltx.PageIndexElem) target = f.pending + baseline = nil targetIsMain = false f.pendingReplace = true } } - for k, v := range combined { - target[k] = v + applied := make(map[uint32]ltx.PageIndexElem) + mergePageIndexes(target, baseline, combined, applied) + for k := range applied { // Invalidate cache if we're updating the main index if targetIsMain { f.cache.Remove(k) @@ -2602,80 +2648,141 @@ func (f *VFSFile) pollReplicaClient(ctx context.Context) error { f.commit = newCommit } - if maxTXID0 > maxTXID1 { - f.pos.TXID = maxTXID0 + if level0.maxTXID > level1.maxTXID { + f.pos.TXID = level0.maxTXID } else { - f.pos.TXID = maxTXID1 + f.pos.TXID = level1.maxTXID } - f.maxTXID1 = maxTXID1 + f.maxTXID1 = level1.maxTXID + f.maxTXID1Anchored = level1.anchored f.logger.Debug("txid updated", "txid", f.pos.TXID.String(), "maxTXID1", f.maxTXID1.String()) // Apply updates to hydrated file if hydration is complete - if f.hydrator != nil && f.hydrator.Complete() && len(combined) > 0 { - if err := f.hydrator.ApplyUpdates(f.ctx, combined); err != nil { - f.logger.Error("failed to apply updates to hydrated file", "error", err) + if f.hydrator != nil && f.hydrator.Complete() && len(applied) > 0 { + if err := f.hydrator.ApplyUpdates(ctx, applied); err != nil { + f.hydrator.Disable() + return fmt.Errorf("apply hydration updates: %w", err) } } return nil } +type pollLevelResult struct { + maxTXID ltx.TXID + index map[uint32]ltx.PageIndexElem + commit uint32 + replaceIndex bool + anchored bool +} + // pollLevel fetches LTX files for a specific level and returns the highest TXID seen, // any index updates, the latest commit value, and if the index should be replaced. -func (f *VFSFile) pollLevel(ctx context.Context, level int, prevMaxTXID ltx.TXID, baseCommit uint32) (ltx.TXID, map[uint32]ltx.PageIndexElem, uint32, bool, error) { - itr, err := f.client.LTXFiles(ctx, level, prevMaxTXID+1, false) - if err != nil { - return prevMaxTXID, nil, baseCommit, false, fmt.Errorf("ltx files: %w", err) +func (f *VFSFile) pollLevel(ctx context.Context, level int, prevMaxTXID ltx.TXID, baseCommit uint32, anchored bool) (pollLevelResult, error) { + result := pollLevelResult{ + maxTXID: prevMaxTXID, + index: make(map[uint32]ltx.PageIndexElem), + commit: baseCommit, + anchored: anchored, } - defer func() { _ = itr.Close() }() - - index := make(map[uint32]ltx.PageIndexElem) - maxTXID := prevMaxTXID lastCommit := baseCommit - newCommit := baseCommit - replaceIndex := false - for itr.Next() { - info := itr.Item() + poll := func(seek ltx.TXID) (*ltx.FileInfo, bool, error) { + itr, err := f.client.LTXFiles(ctx, level, seek, false) + if err != nil { + return nil, false, fmt.Errorf("ltx files: %w", err) + } + defer func() { _ = itr.Close() }() - f.mu.Lock() - isNextTXID := info.MinTXID == maxTXID+1 - f.mu.Unlock() - if !isNextTXID { - if level == 0 && info.MinTXID > maxTXID+1 { - f.logger.Warn("ltx gap detected at L0, deferring to higher levels", "expected", maxTXID+1, "next", info.MinTXID) - break + applied := false + for itr.Next() { + info := itr.Item() + sameUnanchoredBoundary := level >= 1 && !result.anchored && info.MaxTXID == result.maxTXID + if info.MaxTXID <= result.maxTXID && !sameUnanchoredBoundary { + continue } - return maxTXID, nil, newCommit, replaceIndex, fmt.Errorf("non-contiguous ltx file: level=%d, current=%s, next=%s-%s", level, maxTXID, info.MinTXID, info.MaxTXID) - } - f.logger.Debug("new ltx file", "level", info.Level, "min", info.MinTXID, "max", info.MaxTXID) + if !sameUnanchoredBoundary { + nextTXID := result.maxTXID + 1 + if info.MinTXID > nextTXID { + gap := *info + return &gap, applied, nil + } + } - idx, err := FetchPageIndex(ctx, f.client, info) - if err != nil { - return maxTXID, nil, newCommit, replaceIndex, fmt.Errorf("fetch page index: %w", err) + f.logger.Debug("new ltx file", "level", info.Level, "min", info.MinTXID, "max", info.MaxTXID) + + idx, err := FetchPageIndex(ctx, f.client, info) + if err != nil { + return nil, applied, fmt.Errorf("fetch page index: %w", err) + } + hdr, err := FetchLTXHeader(ctx, f.client, info) + if err != nil { + return nil, applied, fmt.Errorf("fetch header: %w", err) + } + + if hdr.Commit < lastCommit { + result.replaceIndex = true + result.index = make(map[uint32]ltx.PageIndexElem) + } + lastCommit = hdr.Commit + result.commit = hdr.Commit + + for k, v := range idx { + f.logger.Debug("adding new page index", "page", k, "elem", v) + } + mergePageIndexes(result.index, nil, idx, nil) + result.maxTXID = info.MaxTXID + if level >= 1 { + result.anchored = true + } + applied = true } - hdr, err := FetchLTXHeader(ctx, f.client, info) - if err != nil { - return maxTXID, nil, newCommit, replaceIndex, fmt.Errorf("fetch header: %w", err) + if err := itr.Err(); err != nil { + return nil, applied, fmt.Errorf("iterate ltx files: %w", err) } + return nil, applied, nil + } - if hdr.Commit < lastCommit { - replaceIndex = true - index = make(map[uint32]ltx.PageIndexElem) + gap, _, err := poll(prevMaxTXID + 1) + if err != nil { + return result, err + } + if level == 0 { + if gap != nil { + f.logger.Warn("ltx gap detected at L0, deferring to higher levels", "expected", result.maxTXID+1, "next", gap.MinTXID) } - lastCommit = hdr.Commit - newCommit = hdr.Commit + return result, nil + } + if gap == nil && result.anchored { + return result, nil + } - for k, v := range idx { - f.logger.Debug("adding new page index", "page", k, "elem", v) - index[k] = v + normalGap := gap + gap, recovered, err := poll(0) + if err != nil { + return result, err + } + if gap != nil { + return result, fmt.Errorf("non-contiguous ltx file: level=%d, current=%s, next=%s-%s", level, result.maxTXID, gap.MinTXID, gap.MaxTXID) + } + if !recovered { + if normalGap != nil { + return result, fmt.Errorf("non-contiguous ltx file: level=%d, current=%s, next=%s-%s", level, result.maxTXID, normalGap.MinTXID, normalGap.MaxTXID) } - maxTXID = info.MaxTXID + return result, nil + } + + gap, _, err = poll(result.maxTXID + 1) + if err != nil { + return result, err + } + if gap != nil { + return result, fmt.Errorf("non-contiguous ltx file: level=%d, current=%s, next=%s-%s", level, result.maxTXID, gap.MinTXID, gap.MaxTXID) } - return maxTXID, index, newCommit, replaceIndex, nil + return result, nil } func (f *VFSFile) pageSizeBytes() (uint32, error) { diff --git a/vfs_link_test.go b/vfs_link_test.go new file mode 100644 index 000000000..0db9d917d --- /dev/null +++ b/vfs_link_test.go @@ -0,0 +1,6 @@ +//go:build vfs + +package litestream + +// Link SQLite symbols required by sqlite3vfs in the tagged test binary. +import _ "github.com/mattn/go-sqlite3" diff --git a/vfs_test.go b/vfs_test.go index 23db0bad6..eca3dd40f 100644 --- a/vfs_test.go +++ b/vfs_test.go @@ -845,12 +845,23 @@ type countingReplicaClient struct { calls atomic.Uint64 } +type failingPageReplicaClient struct { + *mockReplicaClient + failPageReads atomic.Bool +} + func newCountingReplicaClient() *countingReplicaClient { return &countingReplicaClient{} } +func newFailingPageReplicaClient() *failingPageReplicaClient { + return &failingPageReplicaClient{mockReplicaClient: newMockReplicaClient()} +} + func (c *countingReplicaClient) Type() string { return "count" } func (c *countingReplicaClient) Init(context.Context) error { return nil } +func (c *countingReplicaClient) SetLogger(*slog.Logger) {} + func (c *countingReplicaClient) LTXFiles(ctx context.Context, level int, seek ltx.TXID, useMetadata bool) (ltx.FileIterator, error) { c.calls.Add(1) return ltx.NewFileInfoSliceIterator(nil), nil @@ -883,6 +894,8 @@ func (c *mockReplicaClient) Type() string { return "mock" } func (c *mockReplicaClient) Init(context.Context) error { return nil } +func (c *mockReplicaClient) SetLogger(*slog.Logger) {} + func (c *mockReplicaClient) addFixture(tb testing.TB, fx *ltxFixture) { tb.Helper() c.mu.Lock() @@ -921,6 +934,13 @@ func (c *mockReplicaClient) OpenLTXFile(ctx context.Context, level int, minTXID, return io.NopCloser(bytes.NewReader(slice)), nil } +func (c *failingPageReplicaClient) OpenLTXFile(ctx context.Context, level int, minTXID, maxTXID ltx.TXID, offset, size int64) (io.ReadCloser, error) { + if c.failPageReads.Load() && offset > 0 && size > 0 { + return nil, errors.New("injected page read failure") + } + return c.mockReplicaClient.OpenLTXFile(ctx, level, minTXID, maxTXID, offset, size) +} + func (c *mockReplicaClient) WriteLTXFile(context.Context, int, ltx.TXID, ltx.TXID, io.Reader) (*ltx.FileInfo, error) { return nil, fmt.Errorf("not implemented") } @@ -980,16 +1000,24 @@ func buildLTXFixture(tb testing.TB, txid ltx.TXID, fill byte) *ltxFixture { return buildLTXFixtureWithPage(tb, txid, 4096, 1, fill) } +func buildLTXFixtureRange(tb testing.TB, minTXID, maxTXID ltx.TXID, fill byte) *ltxFixture { + return buildLTXFixtureRangeWithPages(tb, minTXID, maxTXID, 4096, []uint32{1}, fill) +} + func buildLTXFixtureWithPage(tb testing.TB, txid ltx.TXID, pageSize, pgno uint32, fill byte) *ltxFixture { return buildLTXFixtureWithPages(tb, txid, pageSize, []uint32{pgno}, fill) } func buildLTXFixtureWithPages(tb testing.TB, txid ltx.TXID, pageSize uint32, pgnos []uint32, fill byte) *ltxFixture { + return buildLTXFixtureRangeWithPages(tb, txid, txid, pageSize, pgnos, fill) +} + +func buildLTXFixtureRangeWithPages(tb testing.TB, minTXID, maxTXID ltx.TXID, pageSize uint32, pgnos []uint32, fill byte) *ltxFixture { tb.Helper() if len(pgnos) == 0 { tb.Fatalf("pgnos required") } - if txid == 1 { + if minTXID == 1 { if len(pgnos) == 0 || pgnos[0] != 1 { tb.Fatalf("snapshot fixture must start at page 1") } @@ -1013,8 +1041,8 @@ func buildLTXFixtureWithPages(tb testing.TB, txid ltx.TXID, pageSize uint32, pgn Version: ltx.Version, PageSize: pageSize, Commit: maxPg, - MinTXID: txid, - MaxTXID: txid, + MinTXID: minTXID, + MaxTXID: maxTXID, Timestamp: time.Now().UnixMilli(), Flags: ltx.HeaderFlagNoChecksum, } @@ -1036,8 +1064,8 @@ func buildLTXFixtureWithPages(tb testing.TB, txid ltx.TXID, pageSize uint32, pgn info := <x.FileInfo{ Level: 0, - MinTXID: txid, - MaxTXID: txid, + MinTXID: minTXID, + MaxTXID: maxTXID, Size: int64(buf.Len()), CreatedAt: time.Now().UTC(), } @@ -1045,6 +1073,40 @@ func buildLTXFixtureWithPages(tb testing.TB, txid ltx.TXID, pageSize uint32, pgn return <xFixture{info: info, data: buf.Bytes()} } +func openVFSFileAtSnapshot(tb testing.TB, maxTXID ltx.TXID, fill byte) (*VFSFile, *mockReplicaClient) { + tb.Helper() + client := newMockReplicaClient() + snapshot := buildLTXFixtureRange(tb, 1, maxTXID, fill) + snapshot.info.Level = SnapshotLevel + client.addFixture(tb, snapshot) + + f := NewVFSFile(client, "test.db", slog.Default()) + if err := f.Open(); err != nil { + tb.Fatalf("open vfs file: %v", err) + } + tb.Cleanup(func() { _ = f.Close() }) + return f, client +} + +func waitForHydration(tb testing.TB, f *VFSFile) { + tb.Helper() + ticker := time.NewTicker(10 * time.Millisecond) + defer ticker.Stop() + timer := time.NewTimer(5 * time.Second) + defer timer.Stop() + + for { + if f.hydrator != nil && f.hydrator.Complete() { + return + } + select { + case <-ticker.C: + case <-timer.C: + tb.Fatalf("hydration did not complete in time") + } + } +} + // TestVFSFile_Hydration_Basic tests that hydration completes and reads from local file. func TestVFSFile_Hydration_Basic(t *testing.T) { client := newMockReplicaClient() @@ -1399,3 +1461,385 @@ func TestVFSFile_Hydration_PersistentResumeOnReopen(t *testing.T) { t.Fatalf("expected hydration file modtime unchanged on reopen resume") } } + +func TestVFSFile_PollLevel1DoesNotRegressNewerL0Pages(t *testing.T) { + client := newMockReplicaClient() + client.addFixture(t, buildLTXFixture(t, 1, 'a')) + + f := NewVFSFile(client, "regress.db", slog.Default()) + if err := f.Open(); err != nil { + t.Fatalf("open vfs file: %v", err) + } + defer f.Close() + + client.addFixture(t, buildLTXFixture(t, 2, 'b')) + client.addFixture(t, buildLTXFixture(t, 3, 'c')) + if err := f.pollReplicaClient(context.Background()); err != nil { + t.Fatalf("poll L0: %v", err) + } + buf := make([]byte, 4096) + if _, err := f.ReadAt(buf, 0); err != nil { + t.Fatalf("read after L0: %v", err) + } + if buf[0] != 'c' { + t.Fatalf("expected newest L0 page, got %q", buf[0]) + } + + l1 := buildLTXFixture(t, 2, 'b') + l1.info.Level = 1 + client.addFixture(t, l1) + if err := f.pollReplicaClient(context.Background()); err != nil { + t.Fatalf("poll L1: %v", err) + } + if got := f.MaxTXID1(); got != 2 { + t.Fatalf("maxTXID1 = %s, want 2", got) + } + if got := f.Pos().TXID; got != 3 { + t.Fatalf("pos = %s, want 3", got) + } + if _, err := f.ReadAt(buf, 0); err != nil { + t.Fatalf("read after L1: %v", err) + } + if buf[0] != 'c' { + t.Fatalf("L1 poll regressed page 1 to older version %q, want 'c'", buf[0]) + } +} + +func TestVFSFile_PollLevel1RecoversStraddlingCompactions(t *testing.T) { + f, client := openVFSFileAtSnapshot(t, 6, 'a') + + straddler := buildLTXFixtureRange(t, 1, 11, 'b') + straddler.info.Level = 1 + client.addFixture(t, straddler) + if err := f.pollReplicaClient(context.Background()); err != nil { + t.Fatalf("poll straddling L1: %v", err) + } + if got := f.MaxTXID1(); got != 11 { + t.Fatalf("maxTXID1 = %s, want 11", got) + } + + next := buildLTXFixtureRange(t, 12, 29, 'c') + next.info.Level = 1 + client.addFixture(t, next) + if err := f.pollReplicaClient(context.Background()); err != nil { + t.Fatalf("poll next L1: %v", err) + } + if got := f.MaxTXID1(); got != 29 { + t.Fatalf("maxTXID1 = %s, want 29", got) + } + if got := f.Pos().TXID; got != 29 { + t.Fatalf("pos = %s, want 29", got) + } + + buf := make([]byte, 4096) + if _, err := f.ReadAt(buf, 0); err != nil { + t.Fatalf("read after L1 recovery: %v", err) + } + if buf[0] != 'c' { + t.Fatalf("read = %q, want newest L1 page 'c'", buf[0]) + } +} + +func TestVFSFile_PollLevel1RecoversStraddlerAndNextFileTogether(t *testing.T) { + f, client := openVFSFileAtSnapshot(t, 6, 'a') + + straddler := buildLTXFixtureRange(t, 1, 11, 'b') + straddler.info.Level = 1 + client.addFixture(t, straddler) + next := buildLTXFixtureRange(t, 12, 29, 'c') + next.info.Level = 1 + client.addFixture(t, next) + + if err := f.pollReplicaClient(context.Background()); err != nil { + t.Fatalf("poll L1 files: %v", err) + } + if got := f.MaxTXID1(); got != 29 { + t.Fatalf("maxTXID1 = %s, want 29", got) + } + if got := f.Pos().TXID; got != 29 { + t.Fatalf("pos = %s, want 29", got) + } +} + +func TestVFSFile_PollLevel1SkipsFileBelowWatermark(t *testing.T) { + f, client := openVFSFileAtSnapshot(t, 6, 'a') + + covered := buildLTXFixtureRange(t, 1, 5, 'b') + covered.info.Level = 1 + client.addFixture(t, covered) + + if err := f.pollReplicaClient(context.Background()); err != nil { + t.Fatalf("poll covered L1: %v", err) + } + if got := f.MaxTXID1(); got != 6 { + t.Fatalf("maxTXID1 = %s, want 6", got) + } +} + +func TestVFSFile_PollLevel1RejectsRealGap(t *testing.T) { + f, client := openVFSFileAtSnapshot(t, 6, 'a') + + gap := buildLTXFixtureRange(t, 8, 11, 'b') + gap.info.Level = 1 + client.addFixture(t, gap) + + err := f.pollReplicaClient(context.Background()) + if err == nil { + t.Fatal("expected non-contiguous L1 error") + } + if got, want := err.Error(), "poll L1: non-contiguous ltx file: level=1, current=0000000000000006, next=0000000000000008-000000000000000b"; got != want { + t.Fatalf("error = %q, want %q", got, want) + } +} + +func TestVFSFile_PollLevel1RecoversFirstFileStartingAboveOne(t *testing.T) { + f, client := openVFSFileAtSnapshot(t, 6, 'a') + + straddler := buildLTXFixtureRange(t, 4, 9, 'b') + straddler.info.Level = 1 + client.addFixture(t, straddler) + + if err := f.pollReplicaClient(context.Background()); err != nil { + t.Fatalf("poll first L1: %v", err) + } + if got := f.MaxTXID1(); got != 9 { + t.Fatalf("maxTXID1 = %s, want 9", got) + } +} + +func TestVFSFile_PollStraddlingLevel1DoesNotRegressNewerL0Pages(t *testing.T) { + client := newMockReplicaClient() + client.addFixture(t, buildLTXFixture(t, 1, 'a')) + + f := NewVFSFile(client, "straddling-regress.db", slog.Default()) + if err := f.Open(); err != nil { + t.Fatalf("open vfs file: %v", err) + } + defer f.Close() + + client.addFixture(t, buildLTXFixture(t, 2, 'b')) + client.addFixture(t, buildLTXFixture(t, 3, 'c')) + if err := f.pollReplicaClient(context.Background()); err != nil { + t.Fatalf("poll L0: %v", err) + } + + l1 := buildLTXFixtureRange(t, 1, 2, 'b') + l1.info.Level = 1 + client.addFixture(t, l1) + if err := f.pollReplicaClient(context.Background()); err != nil { + t.Fatalf("poll straddling L1: %v", err) + } + + buf := make([]byte, 4096) + if _, err := f.ReadAt(buf, 0); err != nil { + t.Fatalf("read after L1: %v", err) + } + if buf[0] != 'c' { + t.Fatalf("L1 poll regressed page 1 to %q, want 'c'", buf[0]) + } +} + +func TestVFSFile_PendingLevel1DoesNotRegressNewerL0Pages(t *testing.T) { + client := newMockReplicaClient() + client.addFixture(t, buildLTXFixture(t, 1, 'a')) + + f := NewVFSFile(client, "pending-regress.db", slog.Default()) + if err := f.Open(); err != nil { + t.Fatalf("open vfs file: %v", err) + } + defer f.Close() + + client.addFixture(t, buildLTXFixture(t, 2, 'b')) + client.addFixture(t, buildLTXFixture(t, 3, 'c')) + if err := f.pollReplicaClient(context.Background()); err != nil { + t.Fatalf("poll L0: %v", err) + } + if err := f.Lock(sqlite3vfs.LockShared); err != nil { + t.Fatalf("lock shared: %v", err) + } + + l1 := buildLTXFixture(t, 2, 'b') + l1.info.Level = 1 + client.addFixture(t, l1) + if err := f.pollReplicaClient(context.Background()); err != nil { + t.Fatalf("poll L1: %v", err) + } + f.mu.Lock() + elem, ok := f.pending[1] + f.mu.Unlock() + if ok { + t.Fatalf("pending page regressed to L1 entry %+v", elem) + } + if err := f.Unlock(sqlite3vfs.LockNone); err != nil { + t.Fatalf("unlock: %v", err) + } + + buf := make([]byte, 4096) + if _, err := f.ReadAt(buf, 0); err != nil { + t.Fatalf("read after unlock: %v", err) + } + if buf[0] != 'c' { + t.Fatalf("L1 poll regressed page 1 to %q, want 'c'", buf[0]) + } +} + +func TestVFSFile_HydratedLevel1DoesNotRegressNewerL0Pages(t *testing.T) { + client := newMockReplicaClient() + client.addFixture(t, buildLTXFixture(t, 1, 'a')) + + f := NewVFSFile(client, "hydrated-regress.db", slog.Default()) + f.hydrationPath = filepath.Join(t.TempDir(), "hydrated-regress.db") + if err := f.Open(); err != nil { + t.Fatalf("open vfs file: %v", err) + } + defer f.Close() + waitForHydration(t, f) + + client.addFixture(t, buildLTXFixture(t, 2, 'b')) + client.addFixture(t, buildLTXFixture(t, 3, 'c')) + if err := f.pollReplicaClient(context.Background()); err != nil { + t.Fatalf("poll L0: %v", err) + } + + l1 := buildLTXFixture(t, 2, 'b') + l1.info.Level = 1 + client.addFixture(t, l1) + if err := f.pollReplicaClient(context.Background()); err != nil { + t.Fatalf("poll L1: %v", err) + } + + buf := make([]byte, 4096) + if _, err := f.ReadAt(buf, 0); err != nil { + t.Fatalf("read hydrated page: %v", err) + } + if buf[0] != 'c' { + t.Fatalf("L1 poll regressed hydrated page 1 to %q, want 'c'", buf[0]) + } +} + +func TestVFSFile_PollLevel1RepointsEqualTXIDPage(t *testing.T) { + client := newMockReplicaClient() + client.addFixture(t, buildLTXFixture(t, 1, 'a')) + l0 := buildLTXFixture(t, 2, 'b') + client.addFixture(t, l0) + + f := NewVFSFile(client, "equal-txid.db", slog.Default()) + if err := f.Open(); err != nil { + t.Fatalf("open vfs file: %v", err) + } + defer f.Close() + + f.mu.Lock() + elem := f.index[1] + anchored := f.maxTXID1Anchored + f.mu.Unlock() + if elem.Level != 0 || elem.MaxTXID != 2 || anchored { + t.Fatalf("initial page index = %+v, anchored=%t; want unanchored level 0 at TXID 2", elem, anchored) + } + + l1 := buildLTXFixtureRange(t, 1, 2, 'b') + l1.info.Level = 1 + client.addFixture(t, l1) + if err := f.pollReplicaClient(context.Background()); err != nil { + t.Fatalf("poll L1: %v", err) + } + + f.mu.Lock() + elem = f.index[1] + anchored = f.maxTXID1Anchored + f.mu.Unlock() + if elem.Level != 1 || elem.MaxTXID != 2 || !anchored { + t.Fatalf("page index = %+v, anchored=%t; want anchored level 1 at TXID 2", elem, anchored) + } + + client.mu.Lock() + delete(client.data, client.key(l0.info)) + client.mu.Unlock() + buf := make([]byte, 4096) + if _, err := f.ReadAt(buf, 0); err != nil { + t.Fatalf("read compacted page: %v", err) + } + if buf[0] != 'b' { + t.Fatalf("read compacted page = %q, want 'b'", buf[0]) + } +} + +func TestVFSFile_PollLevel1KeepsNewerL0Page(t *testing.T) { + client := newMockReplicaClient() + client.addFixture(t, buildLTXFixture(t, 1, 'a')) + + f := NewVFSFile(client, "newer-l0.db", slog.Default()) + if err := f.Open(); err != nil { + t.Fatalf("open vfs file: %v", err) + } + defer f.Close() + + client.addFixture(t, buildLTXFixture(t, 2, 'b')) + client.addFixture(t, buildLTXFixture(t, 3, 'c')) + if err := f.pollReplicaClient(context.Background()); err != nil { + t.Fatalf("poll L0: %v", err) + } + + l1 := buildLTXFixture(t, 2, 'b') + l1.info.Level = 1 + client.addFixture(t, l1) + if err := f.pollReplicaClient(context.Background()); err != nil { + t.Fatalf("poll L1: %v", err) + } + + f.mu.Lock() + elem := f.index[1] + f.mu.Unlock() + if elem.Level != 0 || elem.MaxTXID != 3 { + t.Fatalf("page index = %+v, want level 0 at TXID 3", elem) + } + buf := make([]byte, 4096) + if _, err := f.ReadAt(buf, 0); err != nil { + t.Fatalf("read newer L0 page: %v", err) + } + if buf[0] != 'c' { + t.Fatalf("read newer L0 page = %q, want 'c'", buf[0]) + } +} + +func TestVFSFile_HydrationUpdateFailureDisablesHydration(t *testing.T) { + client := newFailingPageReplicaClient() + client.addFixture(t, buildLTXFixture(t, 1, 'a')) + + f := NewVFSFile(client, "hydration-failure.db", slog.Default()) + f.hydrationPath = filepath.Join(t.TempDir(), "hydration-failure.db") + if err := f.Open(); err != nil { + t.Fatalf("open vfs file: %v", err) + } + defer f.Close() + waitForHydration(t, f) + + client.addFixture(t, buildLTXFixture(t, 2, 'b')) + l1 := buildLTXFixture(t, 2, 'b') + l1.info.Level = 1 + client.addFixture(t, l1) + client.failPageReads.Store(true) + + err := f.pollReplicaClient(context.Background()) + if err == nil || !strings.Contains(err.Error(), "apply hydration updates") { + t.Fatalf("poll error = %v, want hydration update failure", err) + } + if f.hydrator.Complete() { + t.Fatal("hydration remained enabled after update failure") + } + if got := f.Pos().TXID; got != 2 { + t.Fatalf("pos = %s, want 2", got) + } + if got := f.MaxTXID1(); got != 2 { + t.Fatalf("maxTXID1 = %s, want 2", got) + } + + client.failPageReads.Store(false) + buf := make([]byte, 4096) + if _, err := f.ReadAt(buf, 0); err != nil { + t.Fatalf("read remote fallback: %v", err) + } + if buf[0] != 'b' { + t.Fatalf("read remote fallback = %q, want 'b'", buf[0]) + } +} diff --git a/vfs_write_test.go b/vfs_write_test.go index 6afb5dcef..eb94c50a2 100644 --- a/vfs_write_test.go +++ b/vfs_write_test.go @@ -39,6 +39,8 @@ func (c *writeTestReplicaClient) Type() string { return "test" } func (c *writeTestReplicaClient) Init(ctx context.Context) error { return nil } +func (c *writeTestReplicaClient) SetLogger(*slog.Logger) {} + func (c *writeTestReplicaClient) LTXFiles(ctx context.Context, level int, seek ltx.TXID, useMetadata bool) (ltx.FileIterator, error) { c.mu.Lock() defer c.mu.Unlock() @@ -348,6 +350,56 @@ func TestVFSFile_SyncToRemote(t *testing.T) { client.mu.Unlock() } +func TestVFSFile_WriteModeCompactionPollKeepsNewerPage(t *testing.T) { + client := newWriteTestReplicaClient() + pageSize := uint32(4096) + page := bytes.Repeat([]byte{'a'}, int(pageSize)) + createTestLTXFile(t, client, 1, pageSize, 1, map[uint32][]byte{1: page}) + + f := setupWriteableVFSFile(t, client) + if err := f.Open(); err != nil { + t.Fatal(err) + } + defer f.Close() + + compactor := NewCompactor(client, slog.Default()) + if _, err := compactor.Compact(context.Background(), 1); err != nil { + t.Fatalf("compact L0 to L1: %v", err) + } + + newer := bytes.Repeat([]byte{'b'}, int(pageSize)) + if _, err := f.WriteAt(newer, 0); err != nil { + t.Fatalf("write newer page: %v", err) + } + if err := f.Sync(0); err != nil { + t.Fatalf("sync newer page: %v", err) + } + f.mu.Lock() + elem := f.index[1] + f.mu.Unlock() + if elem.Level != 0 || elem.MaxTXID != 2 { + t.Fatalf("synced page index = %+v, want level 0 at TXID 2", elem) + } + if err := f.pollReplicaClient(context.Background()); err != nil { + t.Fatalf("poll compacted L1: %v", err) + } + f.mu.Lock() + elem = f.index[1] + f.mu.Unlock() + if elem.Level != 0 || elem.MaxTXID != 2 { + t.Fatalf("polled page index = %+v, want level 0 at TXID 2", elem) + } + f.cache.Purge() + + buf := make([]byte, pageSize) + if _, err := f.ReadAt(buf, 0); err != nil { + t.Fatalf("read page: %v", err) + } + if buf[0] != 'b' { + t.Fatalf("L1 poll regressed writer page to %q, want 'b'", buf[0]) + } +} + func TestVFSFile_ConflictDetection(t *testing.T) { client := newWriteTestReplicaClient()