diff --git a/vfs.go b/vfs.go index bf2a96d65..e7b7b2a51 100644 --- a/vfs.go +++ b/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() diff --git a/vfs_test.go b/vfs_test.go index 23db0bad6..1ee7b56e0 100644 --- a/vfs_test.go +++ b/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() 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()