From 108059146d67cb4e7e4ee1cb53a8154ddf0680aa Mon Sep 17 00:00:00 2001 From: dailz Date: Wed, 17 Jun 2026 16:34:28 +0800 Subject: [PATCH] fix: write correct startSequence on segment rotation (C8) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit SegmentManager.AppendBatch was passing sm.active.CurrentOffset() (byte offset from file header) as the new segment's startSequence on rotation. The result: segment-N+1's header.startSequence was a byte count (e.g. 50000), not the actual sequence number. Recovery's continuity check at recovery.go:181-184 (segment.StartSequence != expectedSequence) failed, making Phase 1 multi-segment recovery completely broken. Oracle bg_ef425776 noted: "C8 是隐藏炸弹：单 segment 时一切正常， 第一次轮转后就坏". Changes: - wal/segment_manager.go: AppendBatch now takes batchStartSequence uint64 parameter. On rotation, passes it to rotate (which writes it to the new segment's header.startSequence). The previous byte-offset argument is replaced by the actual sequence number. - wal/writer.go: processBatch passes baseSequence (already allocated by seqManager.AllocateBatch) to AppendBatch. - wal/segment_manager_test.go: 6 existing AppendBatch call sites updated to pass batchStartSequence (tracked via local currentSeq variable). Added 2 new tests: - TestSegmentManagerRotationWritesCorrectStartSequence: verifies new segment's header.startSequence matches the first rotated batch's sequence (and explicitly != old byte offset, catching C8 regression). - TestSegmentManagerMultiSegmentRecoveryRoundTrip: end-to-end test that writes across multiple segments, closes, recovers, and verifies all batches replay. Before C8 fix, recovery failed at continuity check. Verified: each new test fails on pre-fix code (segment-1 startSequence is byte offset, recovery fails) and passes after the fix. Full suite green including go test -race ./... . Audit context: docs/audit-3.2.md C8 (Oracle-discovered bg_2e86d33b). --- .../fix-c8-segment-rotation-startsequence.md | 409 ++++++++++++++++++ wal/segment_manager.go | 19 +- wal/segment_manager_test.go | 155 ++++++- wal/writer.go | 2 +- 4 files changed, 568 insertions(+), 17 deletions(-) create mode 100644 .omo/plans/fix-c8-segment-rotation-startsequence.md diff --git a/.omo/plans/fix-c8-segment-rotation-startsequence.md b/.omo/plans/fix-c8-segment-rotation-startsequence.md new file mode 100644 index 0000000..e97b695 --- /dev/null +++ b/.omo/plans/fix-c8-segment-rotation-startsequence.md @@ -0,0 +1,409 @@ +# C8 修复方案：SegmentManager 轮转时传正确的 startSequence + +## TL;DR + +> **目标**：让 SegmentManager 在 segment 轮转时把**真正的 sequence number** 写入新 segment header，而不是当前的 `CurrentOffset()`（字节偏移）。修复后 Phase 1 多 segment 场景的 recovery 能正常通过 `segment.startSequence == expectedSequence` 校验。 +> +> **交付**： +> - `SegmentManager.AppendBatch` 加 `batchStartSequence uint64` 参数 +> - `rotate` 调用从 `sm.active.CurrentOffset()` 改为 `batchStartSequence` +> - `WalWriter.processBatch` 传 `baseSequence` 给 AppendBatch +> - 6 个测试调用点 + 1 个 production 调用点签名更新 +> - 2 个新测试：单元测试验证 segment header startSequence + e2e 多 segment recovery +> - 单次 commit +> +> **预估工时**：1.5-2 小时 +> **风险**：低。改动局限在签名 + 一个调用值 + +--- + +## Context + +### Bug 摘要 + +`wal/segment_manager.go:64-67`： + +```go +if sm.active.RemainingPayload() < worstCaseSize { + if err := sm.rotate(sm.active.CurrentOffset()); err != nil { // ← bug + return fmt.Errorf("wal: rotate segment: %w", err) + } +} +``` + +`CurrentOffset()` 返回字节偏移（从 `WalFileHeaderSize=32` 累加），不是 sequence number。`rotate(newStartSequence uint64)` 把这个值写入新 segment header 的 `startSequence` 字段。 + +**类型 mismatch 隐藏**：`rotate` 的参数叫 `newStartSequence`（uint64），`CurrentOffset()` 返回 uint64，编译器抓不到。 + +### 失败场景 + +``` +segment-0: startSequence=0, 写入 100 batches (seq 0..99) + currentOffset = 32 + encoded_bytes (e.g. 50000) +触发轮转 → segment-1: header.startSequence = 50000 ← bug +写 batch seq 100..199 到 segment-1 + +重启 recovery: + segment-0: startSeq=0, replay 100 batches, nextSequence=100 + segment-1: startSeq=50000, expected=100 → mismatch → ERROR +``` + +### Oracle 发现（bg_ef425776） + +> C8 是隐藏炸弹：单 segment 时一切正常，第一次轮转后就坏。 + +### 设计依据 + +`docs/design.md` §3.2 line 639-663 Segment 连续性校验： + +```text +for segment in recoverySegments: + require segment.segmentID == expectedSegmentID + require segment.startSequence == expectedSequence ← C8 在这失败 + recover all complete batches in segment + expectedSegmentID += 1 + expectedSequence = next sequence after last recovered batch +``` + +### 协同：C4 / C5+H8 + +- **C4 已修**：非尾段损坏硬错误 +- **C5+H8 已修**：truncation 持久化 + batch-aware offset +- **C8 修复后**：多 segment recovery 能跑通 → C4/C5+H8 的多 segment 行为才能在生产中体现 + +--- + +## 执行计划 + +### Phase A：代码改动（15-30 分钟） + +#### A.1 `SegmentManager.AppendBatch` 加 `batchStartSequence` 参数 + +文件：`wal/segment_manager.go` + +```go +// 改前： +func (sm *SegmentManager) AppendBatch(encodedBatch []byte) error { + worstCaseSize := uint64(len(encodedBatch)) + uint64(PhysicalRecordHeaderSize) + uint64(PhysicalRecordHeaderSize) + if sm.active.RemainingPayload() < worstCaseSize { + if err := sm.rotate(sm.active.CurrentOffset()); err != nil { // ← bug + return fmt.Errorf("wal: rotate segment: %w", err) + } + } + return sm.active.AppendBatch(encodedBatch) +} + +// 改后： +// AppendBatch writes encodedBatch to the active segment, rotating first if +// the batch doesn't fit. batchStartSequence is the sequence number of the +// FIRST entry in this batch — used as the new segment's startSequence when +// rotation occurs, so multi-segment recovery's continuity check passes per +// design §3.2 line 639-663. +func (sm *SegmentManager) AppendBatch(encodedBatch []byte, batchStartSequence uint64) error { + worstCaseSize := uint64(len(encodedBatch)) + uint64(PhysicalRecordHeaderSize) + uint64(PhysicalRecordHeaderSize) + if sm.active.RemainingPayload() < worstCaseSize { + // C8 fix: new segment's first batch is THIS batch, so its + // startSequence must equal batchStartSequence (not byte offset). + if err := sm.rotate(batchStartSequence); err != nil { + return fmt.Errorf("wal: rotate segment: %w", err) + } + } + return sm.active.AppendBatch(encodedBatch) +} +``` + +> **必要 docstring**：解释 `batchStartSequence` 的用途（防止未来调用方传错值，类似当前 C8 bug）。属于 security-related 注释。 + +#### A.2 `WalWriter.processBatch` 传 `baseSequence` + +文件：`wal/writer.go:210` + +```go +// 改前： +if err := ww.segManager.AppendBatch(encoded); err != nil { + +// 改后： +if err := ww.segManager.AppendBatch(encoded, baseSequence); err != nil { +``` + +`baseSequence` 在 processBatch 函数内已经通过 `seqManager.AllocateBatch` 分配（writer.go:197），直接传给 AppendBatch。 + +#### A.3 测试调用点更新 + +文件：`wal/segment_manager_test.go`（6 个调用点） + +```go +// 改前（每个调用）： +if err := sm.AppendBatch(batch); err != nil { + +// 改后（每个调用，加 sequence 参数）： +if err := sm.AppendBatch(batch, currentSeq); err != nil { + t.Fatalf("AppendBatch: %v", err) +} +currentSeq += uint64(len(batch.entries)) // 模拟 sequence 推进 +``` + +需要为每个测试维护一个本地 `currentSeq` 变量，模拟 seqManager 的行为。具体修改见 B.3。 + +### Phase B：测试（60-90 分钟） + +#### B.1 单元测试：segment header startSequence 正确性 + +> **Oracle 修订（bg_282d3b17）BLOCKING**：原版循环跑 100 个 batch，每次轮转覆盖 `rotationSeq`。但 segment-1 的 header 是**第一次**轮转时创建的，多次轮转后断言会用错误的 sequence。修复：第一次轮转后就 break，记录 `firstRotationSeq`。同时加 `!= oldOffset` 断言显式抓 C8 regression。 + +新增到 `wal/segment_manager_test.go`： + +```go +// Regression guard for C8: after rotation, the new segment's header +// startSequence must equal the rotated batch's baseSequence, NOT the +// previous segment's byte offset. +func TestSegmentManagerRotationWritesCorrectStartSequence(t *testing.T) { + dir := t.TempDir() + cfg := tinyWalConfig() + + sm, err := NewSegmentManager(dir, 0, 0, cfg) + if err != nil { + t.Fatalf("NewSegmentManager: %v", err) + } + defer sm.Close() + + var firstRotationSeq uint64 + var oldOffset uint64 // byte offset of segment-0 right before first rotation + var hadRotation bool + + for i := 0; ; i++ { + // Snapshot offset + segment ID BEFORE append to capture pre-rotation state. + offsetBefore := sm.active.CurrentOffset() + segIDBefore := sm.ActiveSegmentID() + + encoded, err := EncodeWalBatch(uint64(i), []*WalEntry{ + makePutEntry(fmt.Sprintf("k%d", i), "v"), + }) + if err != nil { + t.Fatalf("EncodeWalBatch %d: %v", i, err) + } + if err := sm.AppendBatch(encoded, uint64(i)); err != nil { + t.Fatalf("AppendBatch %d: %v", i, err) + } + + if !hadRotation && sm.ActiveSegmentID() != segIDBefore { + // First rotation just happened. + hadRotation = true + firstRotationSeq = uint64(i) + oldOffset = offsetBefore + break // Stop at first rotation to avoid overwriting. + } + } + + if !hadRotation { + t.Skip("no rotation occurred with tinyWalConfig; test setup needs adjustment") + } + + // Read segment-1 header and verify startSequence. + seg1Path := filepath.Join(dir, "segment-1.wal") + if _, err := os.Stat(seg1Path); err != nil { + t.Fatalf("segment-1 should exist after rotation: %v", err) + } + + seg1Header := readSegmentHeader(t, seg1Path) + if seg1Header.StartSequence != firstRotationSeq { + t.Errorf("segment-1 startSequence = %d, want %d (first rotation sequence)", + seg1Header.StartSequence, firstRotationSeq) + } + // Explicit C8 regression check: buggy code would set startSequence = oldOffset. + if seg1Header.StartSequence == oldOffset { + t.Errorf("segment-1 startSequence = %d matches old byte offset (C8 regression)", + seg1Header.StartSequence) + } +} + +func readSegmentHeader(t *testing.T, path string) *WalFileHeader { + t.Helper() + f, err := os.Open(path) + if err != nil { + t.Fatalf("Open %s: %v", path, err) + } + defer f.Close() + hdrBuf := make([]byte, WalFileHeaderSize) + if _, err := io.ReadFull(f, hdrBuf); err != nil { + t.Fatalf("read header: %v", err) + } + hdr, err := DecodeWalHeader(hdrBuf) + if err != nil { + t.Fatalf("DecodeWalHeader: %v", err) + } + return hdr +} +``` + +#### B.2 集成测试：多 segment recovery 端到端 + +新增到 `wal/segment_manager_test.go` 或 `db_test.go`： + +```go +// Regression guard for C8: writing across multiple segments, then closing +// and reopening, must successfully recover ALL data. Before C8 fix, the +// second segment's startSequence was a byte offset, causing recovery's +// continuity check to fail. +func TestSegmentManagerMultiSegmentRecoveryRoundTrip(t *testing.T) { + dir := t.TempDir() + cfg := tinyWalConfig() + + sm, err := NewSegmentManager(dir, 0, 0, cfg) + if err != nil { + t.Fatalf("NewSegmentManager: %v", err) + } + + // Write enough batches to force rotation. + const totalBatches = 50 + for i := 0; i < totalBatches; i++ { + encoded, err := EncodeWalBatch(uint64(i), []*WalEntry{ + makePutEntry(fmt.Sprintf("k%d", i), fmt.Sprintf("v%d", i)), + }) + if err != nil { + t.Fatalf("EncodeWalBatch %d: %v", i, err) + } + if err := sm.AppendBatch(encoded, uint64(i)); err != nil { + t.Fatalf("AppendBatch %d: %v", i, err) + } + } + if err := sm.Close(); err != nil { + t.Fatalf("Close: %v", err) + } + + // Verify multiple segments were created. + entries, _ := os.ReadDir(dir) + segCount := 0 + for _, e := range entries { + if _, ok := ParseSegmentFilename(e.Name()); ok { + segCount++ + } + } + if segCount < 2 { + t.Fatalf("expected at least 2 segments after rotation, got %d", segCount) + } + + // Recover and verify all batches replay. + replayer := &mockReplayer{} + result, err := Recover(dir, replayer) + if err != nil { + t.Fatalf("Recover failed (C8 regression): %v", err) + } + if result.NextSequence != totalBatches { + t.Errorf("NextSequence = %d, want %d", result.NextSequence, totalBatches) + } + if len(replayer.puts) != totalBatches { + t.Errorf("replayed puts = %d, want %d", len(replayer.puts), totalBatches) + } +} +``` + +#### B.3 现有测试调用点签名更新 + +> **Oracle 修订（bg_282d3b17）**：原计划写 `currentSeq += uint64(len(batch.entries))` 是错的，因为现有 `segment_manager_test.go` 的调用点用的是 dummy `[]byte`（不是带 entries 的 batch struct）。改成"每次 append 后 currentSeq++"（假设每个 batch 1 entry）。 + +`wal/segment_manager_test.go` 的 6 个调用点： + +- Line 84, 119, 132, 180, 295, 304 + +每个调用点的 `batchStartSequence` 值需要根据测试上下文确定。大多数测试用 `startSequence=0`，所以维护一个本地 `currentSeq := uint64(0)`，每次 append 后 `currentSeq++`（假设每个 batch 1 entry）。具体修改在执行阶段读上下文做。 + +### Phase C：验证（15 分钟） + +```bash +# 1. 编译 +go build ./... + +# 2. 重点测试 +go test ./wal -run 'TestSegmentManagerRotation|TestSegmentManagerMultiSegment|TestSegmentManager' -count=1 -v + +# 3. wal 包全量 +go test ./wal/... -count=1 + +# 4. 全仓 +go test ./... -count=1 + +# 5. race +go test -race ./... -count=1 + +# 6. vet +go vet ./... +``` + +### Phase D：Commit message draft + +``` +fix: write correct startSequence on segment rotation (C8) + +SegmentManager.AppendBatch was passing sm.active.CurrentOffset() (byte +offset from file header) as the new segment's startSequence on rotation. +The result: segment-N+1's header.startSequence was a byte count (e.g. +50000), not the actual sequence number. Recovery's continuity check at +recovery.go:181-184 (segment.StartSequence != expectedSequence) failed, +making Phase 1 multi-segment recovery completely broken. + +Oracle bg_ef425776 noted: "C8 是隐藏炸弹：单 segment 时一切正常， +第一次轮转后就坏". + +Changes: +- wal/segment_manager.go: AppendBatch now takes batchStartSequence uint64 + parameter. On rotation, passes it to rotate (which writes it to the new + segment's header.startSequence). The previous byte-offset argument is + replaced by the actual sequence number. +- wal/writer.go: processBatch passes baseSequence (already allocated by + seqManager.AllocateBatch) to AppendBatch. +- wal/segment_manager_test.go: 6 existing AppendBatch call sites updated + to pass batchStartSequence. Added 2 new tests: + - TestSegmentManagerRotationWritesCorrectStartSequence: verifies new + segment's header.startSequence matches the first rotated batch's + sequence (and explicitly != old byte offset, catching C8 regression). + - TestSegmentManagerMultiSegmentRecoveryRoundTrip: end-to-end test that + writes across multiple segments, closes, recovers, and verifies all + batches replay. Before C8 fix, recovery failed at continuity check. + +Verified: each new test fails on pre-fix code (segment-1 startSequence +is byte offset, recovery fails) and passes after the fix. Full suite +green including go test -race ./... . + +Audit context: docs/audit-3.2.md C8 (Oracle-discovered bg_2e86d33b). +``` + +--- + +## 验收清单 + +- [ ] Phase A.1：`AppendBatch(encodedBatch []byte, batchStartSequence uint64)` 签名存在 +- [ ] Phase A.1：`rotate` 调用从 `CurrentOffset()` 改为 `batchStartSequence` +- [ ] Phase A.2：`writer.go:210` 传 `baseSequence` +- [ ] Phase A.3：6 个测试调用点全部更新 +- [ ] Phase B.1：`TestSegmentManagerRotationWritesCorrectStartSequence` 存在 +- [ ] Phase B.2：`TestSegmentManagerMultiSegmentRecoveryRoundTrip` 存在 +- [ ] `go test ./wal/... -count=1` 全绿 +- [ ] `go test ./... -count=1` 全绿 +- [ ] `go test -race ./... -count=1` 全绿 +- [ ] `go vet ./...` 无新增警告 +- [ ] 单次 commit，message 引用 audit C8 + +--- + +## 不在本次范围内（后续 issue） + +| 编号 | 为什么不放进来 | +|------|---------------| +| C1 | CRC 多项式 IEEE → crc32c，独立 | +| C7 | Put/Close 竞态，独立 | +| H1-H7 | 其他 High，独立 | + +--- + +## 修订记录 + +- **v1（原始）**：C8 修复方案初稿，送 Momus 审 +- **v1.0（Momus 审核 bg_22b882bf）**：[OKAY]，无 blocking。提醒 B.1 测试需要 `fmt` import（执行时补） +- **v1.1（Oracle 修订 bg_282d3b17）**： + - **BLOCKING**：B.1 测试逻辑错 —— 循环跑 100 个 batch，每次轮转覆盖 `rotationSeq`，但 segment-1 的 header 是第一次轮转时创建的。修复：第一次轮转后 break，记录 `firstRotationSeq` + - **NEW**：B.1 加 `!= oldOffset` 断言，显式抓 C8 regression（buggy 代码下值是字节偏移） + - **FIX**：A.3 的 `currentSeq += len(batch.entries)` 误导（现有测试用 dummy `[]byte`），改成"每次 append 后 currentSeq++" + - **FIX**：commit message 行号 `recovery.go:155` 过期（C4 fix 后行号变了），改成 `recovery.go:181-184` + - **MINOR**：`readSegmentHeader` 用 `io.ReadFull` 而不是 `f.Read`（避免 short read） + - **NOTE**：B.2 测试必须留在 wal 包（用 `mockReplayer`，是包内部） diff --git a/wal/segment_manager.go b/wal/segment_manager.go index bfaf800..5a9d4fb 100644 --- a/wal/segment_manager.go +++ b/wal/segment_manager.go @@ -50,19 +50,18 @@ func NewSegmentManager( return sm, nil } -// AppendBatch writes an encoded batch to the active segment. If the batch does -// not fit in the remaining payload space (with worst-case physical record -// overhead), the manager rotates to a fresh segment first so the entire batch -// lands in one segment. -func (sm *SegmentManager) AppendBatch(encodedBatch []byte) error { - // Calculate the worst-case on-disk size for this batch: - // len(encodedBatch) + at least one physical record header + block padding margin - // This is a conservative upper bound. The actual overhead may be less due to - // block alignment, but we must guarantee the batch won't exceed MaxSegmentSize. +// AppendBatch writes encodedBatch to the active segment, rotating first if +// the batch doesn't fit. batchStartSequence is the sequence number of the +// FIRST entry in this batch — used as the new segment's startSequence when +// rotation occurs, so multi-segment recovery's continuity check passes per +// design §3.2 line 639-663. +func (sm *SegmentManager) AppendBatch(encodedBatch []byte, batchStartSequence uint64) error { worstCaseSize := uint64(len(encodedBatch)) + uint64(PhysicalRecordHeaderSize) + uint64(PhysicalRecordHeaderSize) if sm.active.RemainingPayload() < worstCaseSize { - if err := sm.rotate(sm.active.CurrentOffset()); err != nil { + // C8 fix: new segment's first batch is THIS batch, so its + // startSequence must equal batchStartSequence (not byte offset). + if err := sm.rotate(batchStartSequence); err != nil { return fmt.Errorf("wal: rotate segment: %w", err) } } diff --git a/wal/segment_manager_test.go b/wal/segment_manager_test.go index 99cfb71..0eb962b 100644 --- a/wal/segment_manager_test.go +++ b/wal/segment_manager_test.go @@ -2,6 +2,8 @@ package wal import ( "errors" + "fmt" + "io" "os" "path/filepath" "strings" @@ -80,10 +82,12 @@ func TestSegmentManagerRotation(t *testing.T) { } // Write until we rotate past segment 0. + currentSeq := uint64(1) // matches NewSegmentManager startSequence above for i := 0; i < 20; i++ { - if err := sm.AppendBatch(batch); err != nil { + if err := sm.AppendBatch(batch, currentSeq); err != nil { t.Fatalf("AppendBatch %d: %v", i, err) } + currentSeq++ } // After many writes, we should have rotated to a higher segment. @@ -115,10 +119,12 @@ func TestSegmentManagerBatchNotSplit(t *testing.T) { } // Write until we're close to rotation threshold. + currentSeq := uint64(1) // matches NewSegmentManager startSequence above for sm.RemainingPayload() > 256 { - if err := sm.AppendBatch(smallBatch); err != nil { + if err := sm.AppendBatch(smallBatch, currentSeq); err != nil { t.Fatalf("AppendBatch small: %v", err) } + currentSeq++ } // Now write a batch that triggers rotation. @@ -129,7 +135,7 @@ func TestSegmentManagerBatchNotSplit(t *testing.T) { } segIDBefore := sm.ActiveSegmentID() - if err := sm.AppendBatch(triggerBatch); err != nil { + if err := sm.AppendBatch(triggerBatch, currentSeq); err != nil { t.Fatalf("AppendBatch trigger: %v", err) } segIDAfter := sm.ActiveSegmentID() @@ -176,10 +182,12 @@ func TestSegmentManagerCurrentFile(t *testing.T) { // Write enough to force rotation. batch := make([]byte, 32) + currentSeq := uint64(1) // matches NewSegmentManager startSequence above for i := 0; i < 20; i++ { - if err := sm.AppendBatch(batch); err != nil { + if err := sm.AppendBatch(batch, currentSeq); err != nil { t.Fatalf("AppendBatch %d: %v", i, err) } + currentSeq++ } // CURRENT should now point to the active segment. @@ -291,17 +299,152 @@ func TestSegmentManagerRotateFailsOnDirFsyncFailure(t *testing.T) { // segment_manager.go:62 triggers rotate when // RemainingPayload() < len(encoded) + 2*PhysicalRecordHeaderSize worstCaseSize := uint64(len(encoded)) + 2*uint64(PhysicalRecordHeaderSize) + currentSeq := uint64(0) // matches NewSegmentManager startSequence above for sm.RemainingPayload() >= worstCaseSize { - if err := sm.AppendBatch(encoded); err != nil { + if err := sm.AppendBatch(encoded, currentSeq); err != nil { t.Fatalf("fill AppendBatch: %v", err) } + currentSeq++ } orig := dirFsyncFn dirFsyncFn = func(string) error { return errors.New("simulated dir fsync failure") } t.Cleanup(func() { dirFsyncFn = orig }) - if err := sm.AppendBatch(encoded); err == nil { + if err := sm.AppendBatch(encoded, currentSeq); err == nil { t.Fatal("AppendBatch: expected rotation failure, got nil") } } + +// -------- C8 regression guards -------- + +// Regression guard for C8: after rotation, the new segment's header +// startSequence must equal the rotated batch's baseSequence, NOT the +// previous segment's byte offset. +func TestSegmentManagerRotationWritesCorrectStartSequence(t *testing.T) { + dir := t.TempDir() + cfg := tinyWalConfig() + + sm, err := NewSegmentManager(dir, 0, 0, cfg) + if err != nil { + t.Fatalf("NewSegmentManager: %v", err) + } + defer sm.Close() + + var firstRotationSeq uint64 + var oldOffset uint64 + var hadRotation bool + + for i := 0; ; i++ { + // Snapshot offset + segment ID BEFORE append to capture pre-rotation state. + offsetBefore := sm.active.CurrentOffset() + segIDBefore := sm.ActiveSegmentID() + + encoded, err := EncodeWalBatch(uint64(i), []*WalEntry{ + makePutEntry(fmt.Sprintf("k%d", i), "v"), + }) + if err != nil { + t.Fatalf("EncodeWalBatch %d: %v", i, err) + } + if err := sm.AppendBatch(encoded, uint64(i)); err != nil { + t.Fatalf("AppendBatch %d: %v", i, err) + } + + if !hadRotation && sm.ActiveSegmentID() != segIDBefore { + hadRotation = true + firstRotationSeq = uint64(i) + oldOffset = offsetBefore + break + } + } + + if !hadRotation { + t.Skip("no rotation occurred with tinyWalConfig; test setup needs adjustment") + } + + seg1Path := filepath.Join(dir, "segment-1.wal") + if _, err := os.Stat(seg1Path); err != nil { + t.Fatalf("segment-1 should exist after rotation: %v", err) + } + + seg1Header := readSegmentHeader(t, seg1Path) + if seg1Header.StartSequence != firstRotationSeq { + t.Errorf("segment-1 startSequence = %d, want %d (first rotation sequence)", + seg1Header.StartSequence, firstRotationSeq) + } + if seg1Header.StartSequence == oldOffset { + t.Errorf("segment-1 startSequence = %d matches old byte offset (C8 regression)", + seg1Header.StartSequence) + } +} + +// Regression guard for C8: writing across multiple segments, then closing +// and reopening, must successfully recover ALL data. Before C8 fix, the +// second segment's startSequence was a byte offset, causing recovery's +// continuity check to fail. +func TestSegmentManagerMultiSegmentRecoveryRoundTrip(t *testing.T) { + dir := t.TempDir() + cfg := tinyWalConfig() + + sm, err := NewSegmentManager(dir, 0, 0, cfg) + if err != nil { + t.Fatalf("NewSegmentManager: %v", err) + } + + const totalBatches = 50 + for i := 0; i < totalBatches; i++ { + encoded, err := EncodeWalBatch(uint64(i), []*WalEntry{ + makePutEntry(fmt.Sprintf("k%d", i), fmt.Sprintf("v%d", i)), + }) + if err != nil { + t.Fatalf("EncodeWalBatch %d: %v", i, err) + } + if err := sm.AppendBatch(encoded, uint64(i)); err != nil { + t.Fatalf("AppendBatch %d: %v", i, err) + } + } + if err := sm.Close(); err != nil { + t.Fatalf("Close: %v", err) + } + + entries, _ := os.ReadDir(dir) + segCount := 0 + for _, e := range entries { + if _, ok := ParseSegmentFilename(e.Name()); ok { + segCount++ + } + } + if segCount < 2 { + t.Fatalf("expected at least 2 segments after rotation, got %d", segCount) + } + + replayer := &mockReplayer{} + result, err := Recover(dir, replayer) + if err != nil { + t.Fatalf("Recover failed (C8 regression): %v", err) + } + if result.NextSequence != totalBatches { + t.Errorf("NextSequence = %d, want %d", result.NextSequence, totalBatches) + } + if len(replayer.puts) != totalBatches { + t.Errorf("replayed puts = %d, want %d", len(replayer.puts), totalBatches) + } +} + +func readSegmentHeader(t *testing.T, path string) *WalFileHeader { + t.Helper() + f, err := os.Open(path) + if err != nil { + t.Fatalf("Open %s: %v", path, err) + } + defer f.Close() + hdrBuf := make([]byte, WalFileHeaderSize) + if _, err := io.ReadFull(f, hdrBuf); err != nil { + t.Fatalf("read header: %v", err) + } + hdr, err := DecodeWalHeader(hdrBuf) + if err != nil { + t.Fatalf("DecodeWalHeader: %v", err) + } + return hdr +} diff --git a/wal/writer.go b/wal/writer.go index b214651..ad38966 100644 --- a/wal/writer.go +++ b/wal/writer.go @@ -207,7 +207,7 @@ func (ww *WalWriter) processBatch(requests []*CommitRequest) { return } - if err := ww.segManager.AppendBatch(encoded); err != nil { + if err := ww.segManager.AppendBatch(encoded, baseSequence); err != nil { ww.stopWithError(requests, errkit.ErrCommitUnknown) return }