fix(vfs): tolerate L1 files that straddle the poll watermark

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
 		}
 

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
 		}
 

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 &ltxFileIterator{
 		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()

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
 		}
 

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
 			}
 

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
 			}
 

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
 		}
 

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
 			}

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 &ltxFixture{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 := &ltx.FileInfo{
+		Level:     level,
+		MinTXID:   minTXID,
+		MaxTXID:   maxTXID,
+		Size:      int64(buf.Len()),
+		CreatedAt: time.Now().UTC(),
+	}
+	return &ltxFixture{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()

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()

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
 		}