diff --git a/abs/replica_client.go b/abs/replica_client.go index 65822a55c..b9de8e271 100644 --- a/abs/replica_client.go +++ b/abs/replica_client.go @@ -380,12 +380,13 @@ func newLTXFileIterator(ctx context.Context, client *ReplicaClient, level int, s seek: seek, } - // Create paginator for listing blobs with level prefix + // Create paginator for listing blobs with level prefix. List the whole + // level and filter by seek in the iterator (like the s3 backend). We can't + // narrow the list with a seek-derived key prefix: a file whose range + // straddles the seek has a MinTXID below it, so a prefix on the seek value + // would exclude the very file that covers the next needed TXID. dir := litestream.LTXLevelDir(client.Path, level) prefix := dir + "/" - if seek != 0 { - prefix += seek.String() - } itr.pager = client.client.NewListBlobsFlatPager(client.Bucket, &azblob.ListBlobsFlatOptions{ Prefix: &prefix, @@ -457,8 +458,10 @@ func (itr *ltxFileIterator) loadNextPage() bool { Size: *item.Properties.ContentLength, } - // Skip if below seek TXID - if info.MinTXID < itr.seek { + // Skip only if the file's whole range is below the seek TXID. A file + // whose range straddles the seek (MinTXID < seek <= MaxTXID) still + // covers the next TXID a follower needs, so it must be returned. + if info.MaxTXID < itr.seek { continue } diff --git a/file/replica_client.go b/file/replica_client.go index 4c447e417..63f321047 100644 --- a/file/replica_client.go +++ b/file/replica_client.go @@ -114,7 +114,8 @@ func (c *ReplicaClient) LTXFiles(ctx context.Context, level int, seek ltx.TXID, minTXID, maxTXID, err := ltx.ParseFilename(fi.Name()) if err != nil { continue - } else if minTXID < seek { + } else if maxTXID < seek { + // Skip if the whole range is below seek. continue } diff --git a/gs/replica_client.go b/gs/replica_client.go index ab89d2d47..2154b93c7 100644 --- a/gs/replica_client.go +++ b/gs/replica_client.go @@ -130,11 +130,11 @@ func (c *ReplicaClient) LTXFiles(ctx context.Context, level int, seek ltx.TXID, dir := litestream.LTXLevelDir(c.Path, level) prefix := dir + "/" - if seek != 0 { - prefix += seek.String() - } - - return newLTXFileIterator(c.bkt.Objects(ctx, &storage.Query{Prefix: prefix}), c, level), nil + // List the whole level and filter by seek in the iterator (like the s3 + // backend). We can't narrow the list with a seek-derived key prefix: a file + // whose range straddles the seek has a MinTXID below it, so a prefix on the + // seek value would exclude the very file that covers the next needed TXID. + return newLTXFileIterator(c.bkt.Objects(ctx, &storage.Query{Prefix: prefix}), c, level, seek), nil } // WriteLTXFile writes an LTX file from rd to a remote path. @@ -240,15 +240,17 @@ type ltxFileIterator struct { it *storage.ObjectIterator client *ReplicaClient level int + seek ltx.TXID info *ltx.FileInfo err error } -func newLTXFileIterator(it *storage.ObjectIterator, client *ReplicaClient, level int) *ltxFileIterator { +func newLTXFileIterator(it *storage.ObjectIterator, client *ReplicaClient, level int, seek ltx.TXID) *ltxFileIterator { return <xFileIterator{ it: it, client: client, level: level, + seek: seek, } } @@ -278,6 +280,13 @@ func (itr *ltxFileIterator) Next() bool { continue } + // Skip only if the file's whole range is below the seek TXID. A file + // whose range straddles the seek (minTXID < seek <= maxTXID) still + // covers the next TXID a follower needs, so it must be returned. + if maxTXID < itr.seek { + continue + } + // Always use accurate timestamp from metadata since it's zero-cost // GCS includes metadata in LIST operations, so no extra API call needed createdAt := attrs.Created.UTC() diff --git a/nats/replica_client.go b/nats/replica_client.go index 3a0f614f3..6cd956ea4 100644 --- a/nats/replica_client.go +++ b/nats/replica_client.go @@ -317,8 +317,8 @@ func (c *ReplicaClient) LTXFiles(ctx context.Context, level int, seek ltx.TXID, continue } - // Apply seek filter - if minTXID < seek { + // Apply seek filter. Skip if the whole range is below seek. + if maxTXID < seek { continue } diff --git a/oss/replica_client.go b/oss/replica_client.go index 6216a3b01..f16327ebd 100644 --- a/oss/replica_client.go +++ b/oss/replica_client.go @@ -578,8 +578,8 @@ func (itr *fileIterator) Next() bool { MaxTXID: maxTXID, } - // Skip if below seek TXID - if info.MinTXID < itr.seek { + // Skip if the file's whole range is below the seek TXID. + if info.MaxTXID < itr.seek { continue } diff --git a/s3/replica_client.go b/s3/replica_client.go index 21d368f11..5a15b691f 100644 --- a/s3/replica_client.go +++ b/s3/replica_client.go @@ -1680,8 +1680,8 @@ func (itr *fileIterator) Next() bool { MaxTXID: maxTXID, } - // Skip if below seek TXID - if info.MinTXID < itr.seek { + // Skip if the file's whole range is below the seek TXID. + if info.MaxTXID < itr.seek { continue } diff --git a/sftp/replica_client.go b/sftp/replica_client.go index c2ea17001..6917bb5e0 100644 --- a/sftp/replica_client.go +++ b/sftp/replica_client.go @@ -247,7 +247,7 @@ func (c *ReplicaClient) LTXFiles(ctx context.Context, level int, seek ltx.TXID, minTXID, maxTXID, err := ltx.ParseFilename(path.Base(fi.Name())) if err != nil { continue - } else if minTXID < seek { + } else if maxTXID < seek { continue } diff --git a/vfs.go b/vfs.go index bf2a96d65..2dc4a5764 100644 --- a/vfs.go +++ b/vfs.go @@ -2639,11 +2639,28 @@ func (f *VFSFile) pollLevel(ctx context.Context, level int, prevMaxTXID ltx.TXID for itr.Next() { info := itr.Item() - f.mu.Lock() - isNextTXID := info.MinTXID == maxTXID+1 - f.mu.Unlock() - if !isNextTXID { - if level == 0 && info.MinTXID > maxTXID+1 { + // Position this file relative to the highest TXID applied so far: + // + // 1. Already covered (MaxTXID <= maxTXID): nothing new — skip. Defends + // against a narrower file listed alongside a wider one that overlaps + // it. litestream does not rewrite files within a level today, so this + // is robustness, not a path exercised in normal operation. + // 2. Contiguous continuation, including a *straddling* compacted file + // whose range starts at or below maxTXID but extends past it + // (MinTXID <= maxTXID+1 <= MaxTXID): apply it and advance to its + // MaxTXID. Re-applying the already-covered portion is an idempotent + // page-index overwrite. + // 3. Real gap (MinTXID > maxTXID+1): at L0 defer to higher levels; + // above L0 it is an error. + // + // Case 2 keeps a follower from wedging when compaction merges files into + // wider ranges: seeking by TXID must not reject a file that actually + // covers the next TXID we need (e.g. resuming at 6 and meeting a 1-b file). + if info.MaxTXID <= maxTXID { + continue + } + if info.MinTXID > maxTXID+1 { + if level == 0 { f.logger.Warn("ltx gap detected at L0, deferring to higher levels", "expected", maxTXID+1, "next", info.MinTXID) break } diff --git a/vfs_test.go b/vfs_test.go index 23db0bad6..00a382724 100644 --- a/vfs_test.go +++ b/vfs_test.go @@ -384,6 +384,96 @@ func TestVFSFile_OpenSeedsLevel1PositionFromPos(t *testing.T) { } } +// TestVFSFile_StraddlingL1FileAfterSeed reproduces the permanent +// "non-contiguous ltx file" wedge at L1. +// +// When the index is built while L1 is empty, maxTXID1 is seeded from pos.TXID +// (an L0/snapshot position) — here TXID 6 — which need not fall on an L1 file +// boundary. When the writer's first L1 compaction then emits a file whose range +// *straddles* that seed (1..b covers 6), the level-1 seek (seek=maxTXID1+1=7) +// skips it because its MinTXID (1) < seek, and the following L1 file (c..1d) +// then trips the strict MinTXID==maxTXID+1 contiguity check — wedging the +// follower forever even though the data on disk is fully contiguous. +// +// Correct behavior: the straddling 1..b file is ingested (advancing maxTXID1 to +// b), after which c..1d continues contiguously. +func TestVFSFile_StraddlingL1FileAfterSeed(t *testing.T) { + client := newMockReplicaClient() + + // Snapshot covering TXIDs 1..6, no L1 files yet. On open this makes + // pos.TXID = 6 and — since L1 is empty — seeds maxTXID1 = 6. + client.addFixture(t, buildLTXRangeFixture(t, SnapshotLevel, 1, 6, 's')) + + f := NewVFSFile(client, "straddle.db", slog.Default()) + if err := f.Open(); err != nil { + t.Fatalf("open vfs file: %v", err) + } + defer f.Close() + + if got := f.maxTXID1; got != 6 { + t.Fatalf("precondition: maxTXID1 seeded to %s, want 6", got) + } + + // The first L1 compaction emits 1..b — a range that straddles the seed (6). + // It must be ingested; today it is silently skipped (MinTXID 1 < seek 7). + client.addFixture(t, buildLTXRangeFixture(t, 1, 1, 0xb, 'A')) + if err := f.pollReplicaClient(context.Background()); err != nil { + t.Fatalf("poll after straddling L1 file: %v", err) + } + if got := f.maxTXID1; got != 0xb { + t.Fatalf("straddling L1 file 1-b was not ingested: maxTXID1=%s, want b", got) + } + + // The next L1 file continues contiguously from b: c..1d. Without ingesting + // 1..b above, this is where the wedge surfaces as a non-contiguous error. + client.addFixture(t, buildLTXRangeFixture(t, 1, 0xc, 0x1d, 'B')) + if err := f.pollReplicaClient(context.Background()); err != nil { + t.Fatalf("poll after following L1 file: %v", err) + } + if got := f.maxTXID1; got != 0x1d { + t.Fatalf("following L1 file c-1d not ingested: maxTXID1=%s, want 1d", got) + } +} + +// TestVFSFile_MergedWiderL1FileAfterConsume is a defensive test for overlapping +// files within a level. litestream compacts level N into level N+1 and does not +// rewrite files within a level, so this exact state should not arise in normal +// operation — but pollLevel must not wedge if it ever encounters a wider file +// whose range overlaps what a follower has already consumed. Here a follower at +// watermark b meets a wider 7..1d file (MinTXID 7 < b); it must be ingested and +// advance the watermark to 1d rather than be rejected as non-contiguous. +func TestVFSFile_MergedWiderL1FileAfterConsume(t *testing.T) { + client := newMockReplicaClient() + client.addFixture(t, buildLTXRangeFixture(t, SnapshotLevel, 1, 6, 's')) + + f := NewVFSFile(client, "merge.db", slog.Default()) + if err := f.Open(); err != nil { + t.Fatalf("open vfs file: %v", err) + } + defer f.Close() + + // Consume an aligned L1 file 7..b, advancing maxTXID1 to b. + client.addFixture(t, buildLTXRangeFixture(t, 1, 7, 0xb, 'A')) + if err := f.pollReplicaClient(context.Background()); err != nil { + t.Fatalf("poll after aligned L1 file: %v", err) + } + if got := f.maxTXID1; got != 0xb { + t.Fatalf("aligned L1 file not ingested: maxTXID1=%s, want b", got) + } + + // A wider 7..1d file appears whose MinTXID (7) is below the current watermark + // (b) — a within-level overlap. seek=c naturally excludes the narrower 7..b + // (MaxTXID b < c), so the wider file is what must be ingested; it straddles b + // and advances the watermark to 1d. + client.addFixture(t, buildLTXRangeFixture(t, 1, 7, 0x1d, 'B')) + if err := f.pollReplicaClient(context.Background()); err != nil { + t.Fatalf("poll after merged wider L1 file: %v", err) + } + if got := f.maxTXID1; got != 0x1d { + t.Fatalf("merged wider L1 file not ingested: maxTXID1=%s, want 1d", got) + } +} + func TestVFSFile_HeaderForcesDeleteJournal(t *testing.T) { client := newMockReplicaClient() client.addFixture(t, buildLTXFixture(t, 1, 'h')) @@ -868,6 +958,8 @@ func (c *countingReplicaClient) DeleteLTXFiles(context.Context, []*ltx.FileInfo) func (c *countingReplicaClient) DeleteAll(context.Context) error { return nil } +func (c *countingReplicaClient) SetLogger(*slog.Logger) {} + func newMockReplicaClient() *mockReplicaClient { return &mockReplicaClient{data: make(map[string][]byte)} } @@ -883,6 +975,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() @@ -896,7 +990,9 @@ func (c *mockReplicaClient) LTXFiles(ctx context.Context, level int, seek ltx.TX defer c.mu.Unlock() var out []*ltx.FileInfo for _, info := range c.files { - if info.Level == level && info.MinTXID >= seek { + // Mirror the real backends: return a file whose range reaches the seek + // (MaxTXID >= seek), including one that straddles it (MinTXID < seek). + if info.Level == level && info.MaxTXID >= seek { out = append(out, info) } } @@ -1045,6 +1141,49 @@ func buildLTXFixtureWithPages(tb testing.TB, txid ltx.TXID, pageSize uint32, pgn return <xFixture{info: info, data: buf.Bytes()} } +// buildLTXRangeFixture builds an LTX fixture spanning [minTXID, maxTXID] at the +// given level, so tests can model a compacted file (snapshot/L1/L2+) whose range +// covers multiple transactions — including one that straddles a poll watermark. +// The existing builders only produce single-TXID (MinTXID==MaxTXID) files. +func buildLTXRangeFixture(tb testing.TB, level int, minTXID, maxTXID ltx.TXID, fill byte) *ltxFixture { + tb.Helper() + const pageSize = 4096 + + var buf bytes.Buffer + enc, err := ltx.NewEncoder(&buf) + if err != nil { + tb.Fatalf("new encoder: %v", err) + } + hdr := ltx.Header{ + Version: ltx.Version, + PageSize: pageSize, + Commit: 1, // single-page db; page contents are irrelevant to poll-watermark tests + MinTXID: minTXID, + MaxTXID: maxTXID, + Timestamp: time.Now().UnixMilli(), + Flags: ltx.HeaderFlagNoChecksum, + } + if err := enc.EncodeHeader(hdr); err != nil { + tb.Fatalf("encode header: %v", err) + } + page := bytes.Repeat([]byte{fill}, pageSize) + if err := enc.EncodePage(ltx.PageHeader{Pgno: 1}, page); err != nil { + tb.Fatalf("encode page: %v", err) + } + if err := enc.Close(); err != nil { + tb.Fatalf("close encoder: %v", err) + } + + info := <x.FileInfo{ + Level: level, + MinTXID: minTXID, + MaxTXID: maxTXID, + Size: int64(buf.Len()), + CreatedAt: time.Now().UTC(), + } + return <xFixture{info: info, data: buf.Bytes()} +} + // TestVFSFile_Hydration_Basic tests that hydration completes and reads from local file. func TestVFSFile_Hydration_Basic(t *testing.T) { client := newMockReplicaClient() diff --git a/vfs_write_test.go b/vfs_write_test.go index 6afb5dcef..a5d92d169 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() diff --git a/webdav/replica_client.go b/webdav/replica_client.go index 6b5bbcf8d..c5968a527 100644 --- a/webdav/replica_client.go +++ b/webdav/replica_client.go @@ -157,7 +157,7 @@ func (c *ReplicaClient) LTXFiles(ctx context.Context, level int, seek ltx.TXID, minTXID, maxTXID, err := ltx.ParseFilename(path.Base(fi.Name())) if err != nil { continue - } else if minTXID < seek { + } else if maxTXID < seek { continue }