fix: validate physical record length > 0 and type (H1+H2)

DecodePhysicalRecord only checked length upper bound, not lower bound
(length > 0 per design §3.2 line 389). It also didn't validate the
fragment type field (RecInvalid=0 and types > RecLast are corruption
indicators per design §3.2 line 366-372).

Corrupt data with length=0 could pass CRC (payload is empty, CRC only
covers length+type bytes) and inject empty records into the fragment
collector. Invalid type values would only be caught at the collector
level, wrapped as TailCorruptionError, rather than rejected at the
parser level.

Changes:
- wal/record.go: DecodePhysicalRecord now rejects length=0 and
  type ∉ {RecFull..RecLast} before payload copy and CRC check.
  Checks are ordered to reject invalid records ASAP.
- wal/record_test.go: 5 tests:
  - TestDecodePhysicalRecord_RejectZeroLength (H1)
  - TestDecodePhysicalRecord_RejectInvalidType (H2, type=0)
  - TestDecodePhysicalRecord_RejectUnknownType (H2, type>RecLast)
  - TestDecodePhysicalRecord_ValidRecordsUnaffected (regression for
    all 4 valid types)
  - TestParseBlockWrapsInvalidRecordAsTailCorruption (integration:
    ParseBlock wraps H1/H2 errors as TailCorruptionError)

Verified: all existing tests pass. Full suite green including
go test -race ./... .

Audit context: docs/audit-3.2.md H1+H2.

.omo/plans/fix-h1-h2-record-validation.md

@@ -0,0 +1,271 @@
+# H1 + H2 修复方案：Physical Record 加 length + type 校验
+
+## TL;DR
+
+> **目标**：`DecodePhysicalRecord` 加两个校验：`length > 0`（H1）和 `type ∈ {Full, First, Middle, Last}`（H2）。防止损坏数据注入非法 record 绕过语义检查。
+>
+> **交付**：
+> - 2 个 if 检查加到 `DecodePhysicalRecord`
+> - 4 个测试（length=0、type=0、type=5、正常回归）
+> - 单次 commit
+>
+> **预估工时**：~1 小时
+> **风险**：极低。两行 if + 测试
+
+---
+
+## Context
+
+### Bug 摘要
+
+`wal/record.go:39-68` `DecodePhysicalRecord`：
+
+```go
+func DecodePhysicalRecord(data []byte) (rec *PhysicalRecord, consumed int, err error) {
+    ...
+    crc := binary.LittleEndian.Uint32(data[0:4])
+    length := binary.LittleEndian.Uint16(data[4:6])
+    recType := data[6]
+
+    if int(length) > len(data)-PhysicalRecordHeaderSize {  // 只校验上界
+        return nil, 0, errors.New("record: data too short for payload")
+    }
+    // ← H1: 缺 length > 0 检查
+    // ← H2: 缺 type 合法性检查
+    ...
+}
+```
+
+### 设计依据
+
+§3.2 line 389：
+
+> Physical Record 的 `length` 必须 `> 0`
+
+§3.2 line 366-372 fragment 类型表：
+
+> | 0 | Invalid | 非法值，用于损坏检测 |
+> | 1-4 | Full/First/Middle/Last | 合法 |
+
+### 后果
+
+| Bug | 触发条件 | 后果 |
+|-----|---------|------|
+| H1 | length=0 record（CRC 可匹配空 payload） | 空 record 进入 fragment collector，被分类为 TailCorruptionError 而非硬错误 |
+| H2 | type=0 或 type>4 record | 物理 CRC 可通过，到 collector 才报错。ParseBlock 包装为 TailCorruptionError（截断），而非设计期望的损坏检测 |
+
+C4 fix 后，TailCorruptionError 在非尾段会被转硬错误。但尾段仍然截断。按设计，这类 invalid record 应该被 parser 层直接拒绝。
+
+### 已有的零头检测
+
+`ParseBlock` 在调 `DecodePhysicalRecord` 之前检查 `isAllZeros(data[pos:pos+7])`。如果 header 7 bytes 全 0，当作 padding 处理。所以 **DecodePhysicalRecord 只在非全零 header 时被调**。H1/H2 的校验覆盖：
+- length=0 但 type≠0（header 非全零）→ H1 拒绝
+- type=0 但 length≠0 或 crc≠0（header 非全零）→ H2 拒绝
+- type=5+ → H2 拒绝
+
+---
+
+## 执行计划
+
+### Phase A：代码改动（10 分钟）
+
+#### A.1 `DecodePhysicalRecord` 加两个检查
+
+文件：`wal/record.go`
+
+```go
+func DecodePhysicalRecord(data []byte) (rec *PhysicalRecord, consumed int, err error) {
+    if len(data) < PhysicalRecordHeaderSize {
+        return nil, 0, errors.New("record: data too short for header")
+    }
+
+    crc := binary.LittleEndian.Uint32(data[0:4])
+    length := binary.LittleEndian.Uint16(data[4:6])
+    recType := data[6]
+
+    // H1: length must be > 0 per design §3.2 line 389.
+    if length == 0 {
+        return nil, 0, errors.New("record: length must be > 0")
+    }
+    // H2: type must be valid per design §3.2 line 366-372 (RecInvalid=0
+    // is for corruption detection; valid types are RecFull..RecLast).
+    if recType < RecFull || recType > RecLast {
+        return nil, 0, fmt.Errorf("record: invalid type %d", recType)
+    }
+
+    if int(length) > len(data)-PhysicalRecordHeaderSize {
+        return nil, 0, errors.New("record: data too short for payload")
+    }
+    ...（后续不变）
+}
+```
+
+**检查顺序**：H1 → H2 → length 上界 → payload 复制 → CRC。非法 record 尽早拒绝，不做无意义的 payload 复制和 CRC 计算。
+
+> **必要注释**：引用设计行号，防回归。
+
+### Phase B：测试（30 分钟）
+
+新增到 `wal/record_test.go`：
+
+```go
+func TestDecodePhysicalRecord_RejectZeroLength(t *testing.T) {
+    // Construct: [crc 4][length=0 2][type=Full 1] = 7 bytes
+    // CRC covers length(0,0) + type(1) = data[4:7]
+    buf := make([]byte, PhysicalRecordHeaderSize)
+    binary.LittleEndian.PutUint16(buf[4:6], 0)
+    buf[6] = RecFull
+    crc := crc32.Checksum(buf[4:7], crc32cTable)
+    binary.LittleEndian.PutUint32(buf[0:4], crc)
+
+    _, _, err := DecodePhysicalRecord(buf)
+    if err == nil {
+        t.Fatal("expected error for length=0")
+    }
+    if !strings.Contains(err.Error(), "length") {
+        t.Errorf("error should mention length, got: %v", err)
+    }
+}
+
+func TestDecodePhysicalRecord_RejectInvalidType(t *testing.T) {
+    // Construct: [crc 4][length=4 2][type=0 (Invalid) 1][payload 4]
+    payload := []byte("test")
+    buf := make([]byte, PhysicalRecordHeaderSize+len(payload))
+    binary.LittleEndian.PutUint16(buf[4:6], uint16(len(payload)))
+    buf[6] = RecInvalid // 0
+    copy(buf[7:], payload)
+    crc := crc32.Checksum(buf[4:], crc32cTable)
+    binary.LittleEndian.PutUint32(buf[0:4], crc)
+
+    _, _, err := DecodePhysicalRecord(buf)
+    if err == nil {
+        t.Fatal("expected error for type=Invalid(0)")
+    }
+    if !strings.Contains(err.Error(), "type") {
+        t.Errorf("error should mention type, got: %v", err)
+    }
+}
+
+func TestDecodePhysicalRecord_RejectUnknownType(t *testing.T) {
+    // type=5 (beyond RecLast=4)
+    payload := []byte("test")
+    buf := make([]byte, PhysicalRecordHeaderSize+len(payload))
+    binary.LittleEndian.PutUint16(buf[4:6], uint16(len(payload)))
+    buf[6] = 5 // RecLast=4, so 5 is unknown
+    copy(buf[7:], payload)
+    crc := crc32.Checksum(buf[4:], crc32cTable)
+    binary.LittleEndian.PutUint32(buf[0:4], crc)
+
+    _, _, err := DecodePhysicalRecord(buf)
+    if err == nil {
+        t.Fatal("expected error for type=5")
+    }
+    if !strings.Contains(err.Error(), "type") {
+        t.Errorf("error should mention type, got: %v", err)
+    }
+}
+
+// Regression: valid records still decode correctly.
+func TestDecodePhysicalRecord_ValidRecordsUnaffected(t *testing.T) {
+    for _, recType := range []uint8{RecFull, RecFirst, RecMiddle, RecLast} {
+        t.Run(fmt.Sprintf("type=%d", recType), func(t *testing.T) {
+            payload := []byte("test-payload")
+            encoded := EncodePhysicalRecord(recType, payload)
+            rec, consumed, err := DecodePhysicalRecord(encoded)
+            if err != nil {
+                t.Fatalf("type %d: %v", recType, err)
+            }
+            if rec.Type != recType {
+                t.Errorf("Type = %d, want %d", rec.Type, recType)
+            }
+            if consumed != len(encoded) {
+                t.Errorf("consumed = %d, want %d", consumed, len(encoded))
+            }
+        })
+    }
+}
+```
+
+### Phase C：验证（10 分钟）
+
+```bash
+# 1. 编译
+go build ./...
+
+# 2. 重点测试
+go test ./wal -run 'TestDecodePhysicalRecord' -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: validate physical record length > 0 and type (H1+H2)
+
+DecodePhysicalRecord only checked length upper bound, not lower bound
+(length > 0 per design §3.2 line 389). It also didn't validate the
+fragment type field (RecInvalid=0 and types > RecLast are corruption
+indicators per design §3.2 line 366-372).
+
+Corrupt data with length=0 could pass CRC (payload is empty, CRC only
+covers length+type bytes) and inject empty records into the fragment
+collector. Invalid type values would only be caught at the collector
+level, wrapped as TailCorruptionError, rather than rejected at the
+parser level.
+
+Changes:
+- wal/record.go: DecodePhysicalRecord now rejects length=0 and
+  type ∉ {RecFull..RecLast} before payload copy and CRC check.
+  Checks are ordered to reject invalid records ASAP.
+- wal/record_test.go: 4 tests:
+  - TestDecodePhysicalRecord_RejectZeroLength (H1)
+  - TestDecodePhysicalRecord_RejectInvalidType (H2, type=0)
+  - TestDecodePhysicalRecord_RejectUnknownType (H2, type=5)
+  - TestDecodePhysicalRecord_ValidRecordsUnaffected (regression for
+    all 4 valid types)
+
+Verified: all existing tests pass. Full suite green including
+go test -race ./... .
+
+Audit context: docs/audit-3.2.md H1+H2.
+```
+
+---
+
+## 验收清单
+
+- [ ] Phase A.1：`length == 0` 检查存在
+- [ ] Phase A.1：`recType < RecFull || recType > RecLast` 检查存在
+- [ ] Phase B：4 个测试存在并 PASS
+- [ ] `go test ./wal/... -count=1` 全绿
+- [ ] `go test -race ./... -count=1` 全绿
+- [ ] `go vet ./...` 无新增警告
+- [ ] 单次 commit，message 引用 audit H1+H2
+
+---
+
+## 不在本次范围内
+
+| 项 | 为什么不放进来 |
+|------|---------------|
+| H3 | worstCaseSize 估算，独立 |
+| H4 | MaxImmutableCount，独立 |
+| H5-H7 | 模型偏离，独立 |
+| ParseBlock 错误分类 | 当前所有 DecodePhysicalRecord 错误包装为 TailCorruptionError，由 C4 的 isLastSegment 区分。不改 |
+
+---
+
+## 修订记录
+
+- **v1（原始）**：H1+H2 修复方案初稿，送 Momus 审

wal/record.go

@@ -3,6 +3,7 @@ package wal
 import (
 	"encoding/binary"
 	"errors"
+	"fmt"
 	"hash/crc32"
 )
 
@@ -45,6 +46,15 @@ func DecodePhysicalRecord(data []byte) (rec *PhysicalRecord, consumed int, err e
 	length := binary.LittleEndian.Uint16(data[4:6])
 	recType := data[6]
 
+	// H1: length must be > 0 per design §3.2 line 389.
+	if length == 0 {
+		return nil, 0, errors.New("record: length must be > 0")
+	}
+	// H2: type must be valid per design §3.2 line 366-372.
+	if recType < RecFull || recType > RecLast {
+		return nil, 0, fmt.Errorf("record: invalid type %d", recType)
+	}
+
 	if int(length) > len(data)-PhysicalRecordHeaderSize {
 		return nil, 0, errors.New("record: data too short for payload")
 	}

wal/record_test.go

@@ -2,6 +2,9 @@ package wal
 
 import (
 	"bytes"
+	"encoding/binary"
+	"hash/crc32"
+	"strings"
 	"testing"
 )
 
@@ -141,3 +144,97 @@ func TestDataTooShort(t *testing.T) {
 		t.Error("expected error for data too short")
 	}
 }
+
+// -------- H1+H2 regression guards --------
+
+func TestDecodePhysicalRecord_RejectZeroLength(t *testing.T) {
+	buf := make([]byte, PhysicalRecordHeaderSize)
+	binary.LittleEndian.PutUint16(buf[4:6], 0)
+	buf[6] = RecFull
+	crc := crc32.Checksum(buf[4:7], crc32cTable)
+	binary.LittleEndian.PutUint32(buf[0:4], crc)
+
+	_, _, err := DecodePhysicalRecord(buf)
+	if err == nil {
+		t.Fatal("expected error for length=0")
+	}
+	if !strings.Contains(err.Error(), "length") {
+		t.Errorf("error should mention length, got: %v", err)
+	}
+}
+
+func TestDecodePhysicalRecord_RejectInvalidType(t *testing.T) {
+	payload := []byte("test")
+	buf := make([]byte, PhysicalRecordHeaderSize+len(payload))
+	binary.LittleEndian.PutUint16(buf[4:6], uint16(len(payload)))
+	buf[6] = RecInvalid
+	copy(buf[7:], payload)
+	crc := crc32.Checksum(buf[4:], crc32cTable)
+	binary.LittleEndian.PutUint32(buf[0:4], crc)
+
+	_, _, err := DecodePhysicalRecord(buf)
+	if err == nil {
+		t.Fatal("expected error for type=Invalid(0)")
+	}
+	if !strings.Contains(err.Error(), "type") {
+		t.Errorf("error should mention type, got: %v", err)
+	}
+}
+
+func TestDecodePhysicalRecord_RejectUnknownType(t *testing.T) {
+	payload := []byte("test")
+	buf := make([]byte, PhysicalRecordHeaderSize+len(payload))
+	binary.LittleEndian.PutUint16(buf[4:6], uint16(len(payload)))
+	buf[6] = RecLast + 1
+	copy(buf[7:], payload)
+	crc := crc32.Checksum(buf[4:], crc32cTable)
+	binary.LittleEndian.PutUint32(buf[0:4], crc)
+
+	_, _, err := DecodePhysicalRecord(buf)
+	if err == nil {
+		t.Fatal("expected error for type > RecLast")
+	}
+	if !strings.Contains(err.Error(), "type") {
+		t.Errorf("error should mention type, got: %v", err)
+	}
+}
+
+func TestDecodePhysicalRecord_ValidRecordsUnaffected(t *testing.T) {
+	for _, recType := range []uint8{RecFull, RecFirst, RecMiddle, RecLast} {
+		payload := []byte("test-payload")
+		encoded := EncodePhysicalRecord(recType, payload)
+		rec, consumed, err := DecodePhysicalRecord(encoded)
+		if err != nil {
+			t.Errorf("type %d: %v", recType, err)
+		}
+		if rec.Type != recType {
+			t.Errorf("Type = %d, want %d", rec.Type, recType)
+		}
+		if consumed != len(encoded) {
+			t.Errorf("consumed = %d, want %d", consumed, len(encoded))
+		}
+	}
+}
+
+// Oracle nice-to-have: integration test verifying ParseBlock wraps H1/H2
+// errors as TailCorruptionError.
+func TestParseBlockWrapsInvalidRecordAsTailCorruption(t *testing.T) {
+	// Build a block containing a valid record followed by a length=0 record
+	// with non-zero header (so ParseBlock doesn't treat it as zero padding).
+	validRec := EncodePhysicalRecord(RecFull, []byte("valid"))
+
+	invalidRec := make([]byte, PhysicalRecordHeaderSize)
+	binary.LittleEndian.PutUint16(invalidRec[4:6], 0) // length=0
+	invalidRec[6] = RecFull                            // non-zero type
+	crc := crc32.Checksum(invalidRec[4:7], crc32cTable)
+	binary.LittleEndian.PutUint32(invalidRec[0:4], crc)
+
+	block := append(validRec, invalidRec...)
+	_, err := ParseBlock(block)
+	if err == nil {
+		t.Fatal("expected TailCorruptionError for invalid record in block")
+	}
+	if !IsTailCorruption(err) {
+		t.Errorf("expected TailCorruptionError, got: %v", err)
+	}
+}