fix(vfs): stop wedging the level-1 tailer on a straddling first compaction

vfs.go

@@ -1209,10 +1209,13 @@ func (f *VFSFile) rebuildIndex(ctx context.Context, infos []*ltx.FileInfo, targe
 	}
 
 	maxTXID1 := maxLevelTXID(infos, 1)
-	// Seed maxTXID1 from pos when there are no L1 files
-	if maxTXID1 == 0 {
-		maxTXID1 = pos.TXID
-	}
+	// When there are no L1 files yet, maxTXID1 must stay zero: the first
+	// L1 compaction emits a file starting at TXID 1 whose range can
+	// straddle the current L0 position. Seeding from the L0-derived
+	// pos.TXID skipped that file in the level-1 seek forever and wedged
+	// the follower on "non-contiguous ltx file" once the next L1 file
+	// appeared (#1460). Starting from zero consumes the first L1 file
+	// (and any later ones) contiguously instead.
 
 	f.mu.Lock()
 	defer f.mu.Unlock()

vfs_test.go

@@ -360,7 +360,7 @@ func TestVFSFile_OpenSeedsLevel1Position(t *testing.T) {
 	}
 }
 
-func TestVFSFile_OpenSeedsLevel1PositionFromPos(t *testing.T) {
+func TestVFSFile_OpenKeepsLevel1PositionAtZeroWithoutL1Files(t *testing.T) {
 	client := newMockReplicaClient()
 	snapshot := buildLTXFixture(t, 1, 's')
 	snapshot.info.Level = SnapshotLevel
@@ -375,12 +375,59 @@ func TestVFSFile_OpenSeedsLevel1PositionFromPos(t *testing.T) {
 	}
 	defer f.Close()
 
-	pos := f.Pos().TXID
-	if pos == 0 {
+	if pos := f.Pos().TXID; pos == 0 {
 		t.Fatalf("expected non-zero position")
 	}
-	if got := f.maxTXID1; got != pos {
-		t.Fatalf("expected maxTXID1 to equal pos when no L1 files, got %s want %s", got, pos)
+	// maxTXID1 must stay zero so the first L1 file — which starts at
+	// TXID 1 regardless of how far L0 has advanced — is consumed
+	// contiguously (#1460). Seeding it from the L0 position skipped the
+	// straddling first compaction output forever.
+	if got := f.maxTXID1; got != 0 {
+		t.Fatalf("expected maxTXID1 to stay zero when no L1 files, got %s", got)
+	}
+}
+
+func TestVFSFile_PollConsumesStraddlingL1Compaction(t *testing.T) {
+	client := newMockReplicaClient()
+	snapshot := buildLTXFixture(t, 1, 's')
+	snapshot.info.Level = SnapshotLevel
+	client.addFixture(t, snapshot)
+	for txid := ltx.TXID(2); txid <= 6; txid++ {
+		l0 := buildLTXFixture(t, txid, byte('0'+int(txid)%10))
+		l0.info.Level = 0
+		client.addFixture(t, l0)
+	}
+
+	f := NewVFSFile(client, "l1-straddle.db", slog.Default())
+	if err := f.Open(); err != nil {
+		t.Fatalf("open vfs file: %v", err)
+	}
+	defer f.Close()
+
+	// The writer's first L0->L1 compaction emits a file whose range
+	// straddles the follower's L0 position (1..0xb vs pos 6), followed
+	// by the next L1 file (0xc..0x1d). The fixture bodies are single-TXID
+	// LTX files; only the advertised TXID range matters for the seek and
+	// contiguity checks under test.
+	straddler := buildLTXFixture(t, 0xb, 'A')
+	straddler.info.Level = 1
+	straddler.info.MinTXID = 1
+	client.addFixture(t, straddler)
+	next := buildLTXFixture(t, 0x1d, 'B')
+	next.info.Level = 1
+	next.info.MinTXID = 0xc
+	client.addFixture(t, next)
+
+	// Before the fix this wedged forever with:
+	//   poll L1: non-contiguous ltx file: level=1, current=6, next=c-1d
+	if err := f.pollReplicaClient(context.Background()); err != nil {
+		t.Fatalf("poll replica: %v", err)
+	}
+	if got, want := f.maxTXID1, ltx.TXID(0x1d); got != want {
+		t.Fatalf("unexpected maxTXID1: got %s want %s", got, want)
+	}
+	if got, want := f.Pos().TXID, ltx.TXID(0x1d); got != want {
+		t.Fatalf("unexpected pos after straddling L1 consumption: got %s want %s", got, want)
 	}
 }
 
@@ -851,6 +898,8 @@ 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 +932,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()

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