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

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

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
+}

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
 	}