feat: initialize project structure and error types

- go.mod with github.com/dailz/go-kv, Go 1.26.3, testify
- config/config.go with WalConfig, Validate() with checked arithmetic
- errors.go with sentinel errors (ErrCommitUnknown, ErrWriteStopped, etc.)
- wal/constants.go with all WAL format constants and enums
- wal/header.go with WAL File Header encode/decode (CRC32 IEEE)
- wal/record.go with Physical Record codec, block boundary, SplitIntoRecords
- wal/entry.go with WAL Entry codec (varint keys/values, OpType, ValueKind)
- wal/sequence.go with SequenceManager (atomic, CAS, overflow-safe)
- manifest/manifest.go with MANIFEST stub (Load/Save atomic)
- manifest/current.go with CURRENT file (WriteCurrent/ReadCurrent)
- Comprehensive tests for all modules
- .golangci.yml configuration
This commit is contained in:
dailz
2026-06-12 13:23:27 +08:00
parent 59de99ca0b
commit cf913b1d52
52 changed files with 4538 additions and 0 deletions
+24
View File
@@ -0,0 +1,24 @@
run:
timeout: 5m
linters:
enable:
- errcheck
- govet
- staticcheck
- unused
- gosimple
- ineffassign
- typecheck
- misspell
- gofmt
linters-settings:
errcheck:
check-type-assertions: true
govet:
enable-all: true
issues:
max-issues-per-linter: 50
max-same-issues: 5
+89
View File
@@ -0,0 +1,89 @@
{
"schema_version": 2,
"active_work_id": "phase1-wal-3bc18f0c",
"works": {
"phase1-wal-3bc18f0c": {
"work_id": "phase1-wal-3bc18f0c",
"active_plan": "/home/dailz/workspace/src/go-kv/.omo/plans/phase1-wal.md",
"plan_name": "phase1-wal",
"status": "active",
"started_at": "2026-06-12T05:09:32.588Z",
"updated_at": "2026-06-12T05:22:47.301Z",
"session_ids": [
"opencode:ses_145c3bae9ffeTB2zbsTym0Cev8"
],
"session_origins": {
"opencode:ses_145c3bae9ffeTB2zbsTym0Cev8": "direct"
},
"agent": "atlas",
"task_sessions": {
"todo:1": {
"task_key": "todo:1",
"task_label": "1",
"task_title": "Project scaffolding + go.mod + config types",
"session_id": "opencode:ses_145c254a5ffemom68fK0rmB26W",
"agent": "Sisyphus-Junior",
"category": "quick",
"updated_at": "2026-06-12T05:15:14.028Z",
"started_at": "2026-06-12T05:14:15.365Z",
"status": "completed",
"ended_at": "2026-06-12T05:15:14.028Z",
"elapsed_ms": 58663
},
"todo:2": {
"task_key": "todo:2",
"task_label": "2",
"task_title": "Error types + constants",
"session_id": "opencode:ses_145bdf583ffe6PSExxk7J357zT",
"agent": "Sisyphus-Junior",
"category": "quick",
"updated_at": "2026-06-12T05:22:47.301Z",
"started_at": "2026-06-12T05:19:53.997Z",
"status": "completed",
"ended_at": "2026-06-12T05:22:47.301Z",
"elapsed_ms": 173304
}
}
}
},
"active_plan": "/home/dailz/workspace/src/go-kv/.omo/plans/phase1-wal.md",
"started_at": "2026-06-12T05:09:32.588Z",
"status": "active",
"updated_at": "2026-06-12T05:22:47.301Z",
"session_ids": [
"opencode:ses_145c3bae9ffeTB2zbsTym0Cev8"
],
"session_origins": {
"opencode:ses_145c3bae9ffeTB2zbsTym0Cev8": "direct"
},
"plan_name": "phase1-wal",
"task_sessions": {
"todo:1": {
"task_key": "todo:1",
"task_label": "1",
"task_title": "Project scaffolding + go.mod + config types",
"session_id": "opencode:ses_145c254a5ffemom68fK0rmB26W",
"agent": "Sisyphus-Junior",
"category": "quick",
"updated_at": "2026-06-12T05:15:14.028Z",
"started_at": "2026-06-12T05:14:15.365Z",
"status": "completed",
"ended_at": "2026-06-12T05:15:14.028Z",
"elapsed_ms": 58663
},
"todo:2": {
"task_key": "todo:2",
"task_label": "2",
"task_title": "Error types + constants",
"session_id": "opencode:ses_145bdf583ffe6PSExxk7J357zT",
"agent": "Sisyphus-Junior",
"category": "quick",
"updated_at": "2026-06-12T05:22:47.301Z",
"started_at": "2026-06-12T05:19:53.997Z",
"status": "completed",
"ended_at": "2026-06-12T05:22:47.301Z",
"elapsed_ms": 173304
}
},
"agent": "atlas"
}
View File
File diff suppressed because it is too large Load Diff
@@ -0,0 +1,10 @@
{
"sessionID": "ses_1465a5321ffevd4inJJZXxFenJ",
"updatedAt": "2026-06-12T02:26:27.295Z",
"sources": {
"background-task": {
"state": "idle",
"updatedAt": "2026-06-12T02:26:27.295Z"
}
}
}
@@ -0,0 +1,10 @@
{
"sessionID": "ses_1468f0cc5ffew53IA0I4W6HijM",
"updatedAt": "2026-06-12T01:41:18.251Z",
"sources": {
"background-task": {
"state": "idle",
"updatedAt": "2026-06-12T01:41:18.251Z"
}
}
}
@@ -0,0 +1,10 @@
{
"sessionID": "ses_1469860f1ffeAxT7jbUD3T1ZK5",
"updatedAt": "2026-06-12T01:18:59.508Z",
"sources": {
"background-task": {
"state": "idle",
"updatedAt": "2026-06-12T01:18:59.508Z"
}
}
}
@@ -0,0 +1,10 @@
{
"sessionID": "ses_146adbaa4ffehxje0lmKRg4ZpT",
"updatedAt": "2026-06-12T00:56:53.921Z",
"sources": {
"background-task": {
"state": "idle",
"updatedAt": "2026-06-12T00:56:53.921Z"
}
}
}
@@ -0,0 +1,10 @@
{
"sessionID": "ses_146b864adffetMjZkc08p16onz",
"updatedAt": "2026-06-12T00:44:22.791Z",
"sources": {
"background-task": {
"state": "idle",
"updatedAt": "2026-06-12T00:44:22.791Z"
}
}
}
@@ -0,0 +1,10 @@
{
"sessionID": "ses_146c330caffehDbiMQ7M0jL3Il",
"updatedAt": "2026-06-12T00:31:46.022Z",
"sources": {
"background-task": {
"state": "idle",
"updatedAt": "2026-06-12T00:31:46.022Z"
}
}
}
@@ -0,0 +1,10 @@
{
"sessionID": "ses_149585098ffeVZ53676PRrr2ic",
"updatedAt": "2026-06-11T12:31:44.016Z",
"sources": {
"background-task": {
"state": "idle",
"updatedAt": "2026-06-11T12:31:44.016Z"
}
}
}
@@ -0,0 +1,10 @@
{
"sessionID": "ses_14960b98effe7YjeN56yvxsoN3",
"updatedAt": "2026-06-11T12:21:18.579Z",
"sources": {
"background-task": {
"state": "idle",
"updatedAt": "2026-06-11T12:21:18.579Z"
}
}
}
@@ -0,0 +1,10 @@
{
"sessionID": "ses_14967e191ffeLfQYyfTQHP5TlH",
"updatedAt": "2026-06-11T12:12:33.047Z",
"sources": {
"background-task": {
"state": "idle",
"updatedAt": "2026-06-11T12:12:33.047Z"
}
}
}
@@ -0,0 +1,10 @@
{
"sessionID": "ses_149a13a61ffehpTGChldqMe0do",
"updatedAt": "2026-06-11T11:11:39.805Z",
"sources": {
"background-task": {
"state": "idle",
"updatedAt": "2026-06-11T11:11:39.805Z"
}
}
}
@@ -0,0 +1,10 @@
{
"sessionID": "ses_149e4f0c5ffe5hf5hErOGwmaO7",
"updatedAt": "2026-06-11T10:01:28.529Z",
"sources": {
"background-task": {
"state": "idle",
"updatedAt": "2026-06-11T10:01:28.529Z"
}
}
}
@@ -0,0 +1,10 @@
{
"sessionID": "ses_149f503d8ffeCitCeNNhuI5uPD",
"updatedAt": "2026-06-11T09:38:47.486Z",
"sources": {
"background-task": {
"state": "idle",
"updatedAt": "2026-06-11T09:38:47.486Z"
}
}
}
@@ -0,0 +1,10 @@
{
"sessionID": "ses_1551b5cb8ffeRq3O3tTE6gADNa",
"updatedAt": "2026-06-09T05:49:57.317Z",
"sources": {
"background-task": {
"state": "idle",
"updatedAt": "2026-06-09T05:49:57.317Z"
}
}
}
@@ -0,0 +1,10 @@
{
"sessionID": "ses_15535520fffenpCk4v55tlxfPN",
"updatedAt": "2026-06-09T05:14:00.942Z",
"sources": {
"background-task": {
"state": "idle",
"updatedAt": "2026-06-09T05:14:00.942Z"
}
}
}
@@ -0,0 +1,10 @@
{
"sessionID": "ses_155c694ebffeUxfvz2X0kcpMQz",
"updatedAt": "2026-06-09T03:20:21.146Z",
"sources": {
"background-task": {
"state": "idle",
"updatedAt": "2026-06-09T03:20:21.146Z"
}
}
}
@@ -0,0 +1,10 @@
{
"sessionID": "ses_162be27e5ffePD7UoGttrWFm7T",
"updatedAt": "2026-06-06T14:13:39.321Z",
"sources": {
"background-task": {
"state": "idle",
"updatedAt": "2026-06-06T14:13:39.321Z"
}
}
}
@@ -0,0 +1,10 @@
{
"sessionID": "ses_16f5a4050ffe4VOJm8EOoPcnZV",
"updatedAt": "2026-06-04T03:24:44.020Z",
"sources": {
"background-task": {
"state": "idle",
"updatedAt": "2026-06-04T03:24:44.020Z"
}
}
}
@@ -0,0 +1,10 @@
{
"sessionID": "ses_16f63a221ffew3fp5TM11NaRDD",
"updatedAt": "2026-06-04T03:18:04.363Z",
"sources": {
"background-task": {
"state": "idle",
"updatedAt": "2026-06-04T03:18:04.363Z"
}
}
}
@@ -0,0 +1,10 @@
{
"sessionID": "ses_17337fda2fferaZJ6tpXI2Cw44",
"updatedAt": "2026-06-04T03:25:22.600Z",
"sources": {
"background-task": {
"state": "idle",
"updatedAt": "2026-06-04T03:25:22.600Z"
}
}
}
@@ -0,0 +1,10 @@
{
"sessionID": "ses_1738cec10ffex8teaUh336WrCD",
"updatedAt": "2026-06-03T08:41:49.094Z",
"sources": {
"background-task": {
"state": "idle",
"updatedAt": "2026-06-03T08:41:49.094Z"
}
}
}
+205
View File
@@ -0,0 +1,205 @@
// Package config defines configuration types and validation for the go-kv storage engine.
package config
import (
"fmt"
"math"
)
// WAL format constants derived from the binary layout specification.
const (
walFileHeaderSize uint64 = 32
walBlockSize uint32 = 32 * 1024 // 32KB
physicalRecordHeaderSize uint64 = 7
walBatchHeaderSize uint64 = 18 // flags(2) + baseSequence(8) + entryCount(4) + entriesSize(4)
)
// WalConfig holds configuration for the Write-Ahead Log subsystem.
// Zero-value WalConfig is valid and uses defaults; call Validate() to apply
// defaults and verify invariants.
type WalConfig struct {
// MaxSegmentSize is the maximum size of a single WAL segment file in bytes.
// Default: 64MB. Must be large enough to hold the largest possible WAL Batch.
MaxSegmentSize uint64
// BlockSize is the WAL block size in bytes.
// Default: 32KB.
BlockSize uint32
// SyncMode controls when WAL writes are flushed to disk.
// Phase 1 only supports "always".
SyncMode string
// MaxBatchEntries is the maximum number of entries in a single WAL Batch.
// Default: 10000.
MaxBatchEntries uint32
// MaxBatchSize is the maximum total size of WAL Batch entries in bytes.
// Default: 4MB.
MaxBatchSize uint32
// MaxKeyBytes is the maximum size of a single key in bytes.
// Default: 4KB.
MaxKeyBytes uint32
// MaxInlineValue is the maximum size of an inline value in bytes.
// Values larger than this must use ValueLogPointer.
// Default: 4KB.
MaxInlineValue uint32
// MemTableSize is the target MemTable size in bytes before triggering flush.
// Default: 64MB.
MemTableSize uint32
// MaxImmutableCount is the maximum number of immutable MemTables allowed
// before writes are stalled. Default: 3.
MaxImmutableCount int
}
// Defaults returns a WalConfig populated with production defaults.
func Defaults() WalConfig {
return WalConfig{
MaxSegmentSize: 64 * 1024 * 1024, // 64MB
BlockSize: 32 * 1024, // 32KB
SyncMode: "always",
MaxBatchEntries: 10000,
MaxBatchSize: 4 * 1024 * 1024, // 4MB
MaxKeyBytes: 4 * 1024, // 4KB
MaxInlineValue: 4 * 1024, // 4KB
MemTableSize: 64 * 1024 * 1024, // 64MB
MaxImmutableCount: 3,
}
}
// applyDefaults fills zero-valued fields with production defaults.
func (c *WalConfig) applyDefaults() {
d := Defaults()
if c.MaxSegmentSize == 0 {
c.MaxSegmentSize = d.MaxSegmentSize
}
if c.BlockSize == 0 {
c.BlockSize = d.BlockSize
}
if c.SyncMode == "" {
c.SyncMode = d.SyncMode
}
if c.MaxBatchEntries == 0 {
c.MaxBatchEntries = d.MaxBatchEntries
}
if c.MaxBatchSize == 0 {
c.MaxBatchSize = d.MaxBatchSize
}
if c.MaxKeyBytes == 0 {
c.MaxKeyBytes = d.MaxKeyBytes
}
if c.MaxInlineValue == 0 {
c.MaxInlineValue = d.MaxInlineValue
}
if c.MemTableSize == 0 {
c.MemTableSize = d.MemTableSize
}
if c.MaxImmutableCount == 0 {
c.MaxImmutableCount = d.MaxImmutableCount
}
}
// Validate applies defaults and verifies that all configuration invariants hold.
// The key invariant ensures that the largest possible WAL Batch can fit into
// an empty WAL segment:
//
// maxWalSegmentPayload >= maxEncodedWalBatchSize + worstCasePhysicalRecordOverhead + worstCaseBlockPadding
//
// All arithmetic is checked for overflow.
func (c *WalConfig) Validate() error {
c.applyDefaults()
if c.SyncMode != "always" {
return fmt.Errorf("config: SyncMode %q not supported (Phase 1: only \"always\")", c.SyncMode)
}
if c.MaxImmutableCount < 1 {
return fmt.Errorf("config: MaxImmutableCount must be >= 1, got %d", c.MaxImmutableCount)
}
// --- Checked arithmetic invariant validation ---
// Mirrors the derivation in docs/design.md § WAL Segment Rotation.
blockSize := uint64(c.BlockSize)
prHeaderSize := physicalRecordHeaderSize
batchHeaderSize := walBatchHeaderSize
maxBatchEntriesSize := uint64(c.MaxBatchSize)
// maxEncodedWalBatchSize = batchHeaderSize + maxBatchEntriesSize
maxEncodedWalBatchSize, err := safeAdd(batchHeaderSize, maxBatchEntriesSize)
if err != nil {
return fmt.Errorf("config: WAL batch size overflow: %w", err)
}
// maxPhysicalRecordPayload = blockSize - prHeaderSize
if blockSize <= prHeaderSize {
return fmt.Errorf("config: BlockSize %d must be > physical record header size %d", blockSize, prHeaderSize)
}
maxPhysicalRecordPayload := blockSize - prHeaderSize
// maxPhysicalRecordCount = ceil(maxEncodedWalBatchSize / maxPhysicalRecordPayload)
maxPhysicalRecordCount := divCeil(maxEncodedWalBatchSize, maxPhysicalRecordPayload)
// worstCasePhysicalRecordOverhead = maxPhysicalRecordCount * prHeaderSize
worstCasePhysicalRecordOverhead, err := safeMul(maxPhysicalRecordCount, prHeaderSize)
if err != nil {
return fmt.Errorf("config: physical record overhead overflow: %w", err)
}
// worstCaseBlockPadding = blockSize - 1 (at most one partial block of padding)
// From design doc: worstCaseBlockPadding = 7 bytes with default block size.
// Generalized: blockSize - maxPhysicalRecordPayload = prHeaderSize
worstCaseBlockPadding := prHeaderSize
// minWalSegmentPayload = maxEncodedWalBatchSize + worstCasePhysicalRecordOverhead + worstCaseBlockPadding
partial, err := safeAdd(maxEncodedWalBatchSize, worstCasePhysicalRecordOverhead)
if err != nil {
return fmt.Errorf("config: segment payload calculation overflow: %w", err)
}
minWalSegmentPayload, err := safeAdd(partial, worstCaseBlockPadding)
if err != nil {
return fmt.Errorf("config: segment payload calculation overflow: %w", err)
}
// maxWalSegmentPayload = MaxSegmentSize - walFileHeaderSize
if c.MaxSegmentSize <= walFileHeaderSize {
return fmt.Errorf("config: MaxSegmentSize %d must be > WAL file header size %d",
c.MaxSegmentSize, walFileHeaderSize)
}
maxWalSegmentPayload := c.MaxSegmentSize - walFileHeaderSize
if maxWalSegmentPayload < minWalSegmentPayload {
return fmt.Errorf("config: MaxSegmentSize %d too small: "+
"segment payload (%d) < minimum required (%d); "+
"need MaxSegmentSize >= %d",
c.MaxSegmentSize, maxWalSegmentPayload, minWalSegmentPayload,
minWalSegmentPayload+walFileHeaderSize)
}
return nil
}
// safeAdd returns a + b or an error if the result overflows uint64.
func safeAdd(a, b uint64) (uint64, error) {
if a > math.MaxUint64-b {
return 0, fmt.Errorf("uint64 overflow: %d + %d", a, b)
}
return a + b, nil
}
// safeMul returns a * b or an error if the result overflows uint64.
func safeMul(a, b uint64) (uint64, error) {
if a != 0 && b > math.MaxUint64/a {
return 0, fmt.Errorf("uint64 overflow: %d * %d", a, b)
}
return a * b, nil
}
// divCeil returns ceil(a / b) for b > 0.
func divCeil(a, b uint64) uint64 {
return (a + b - 1) / b
}
+32
View File
@@ -0,0 +1,32 @@
package config
import (
"testing"
"github.com/stretchr/testify/assert"
"github.com/stretchr/testify/require"
)
func TestValidateDefaults(t *testing.T) {
cfg := WalConfig{}
err := cfg.Validate()
require.NoError(t, err, "default WalConfig should pass validation")
d := Defaults()
assert.Equal(t, d.MaxSegmentSize, cfg.MaxSegmentSize, "MaxSegmentSize should be defaulted")
assert.Equal(t, d.BlockSize, cfg.BlockSize, "BlockSize should be defaulted")
assert.Equal(t, d.SyncMode, cfg.SyncMode, "SyncMode should be defaulted")
assert.Equal(t, d.MaxBatchEntries, cfg.MaxBatchEntries, "MaxBatchEntries should be defaulted")
assert.Equal(t, d.MaxBatchSize, cfg.MaxBatchSize, "MaxBatchSize should be defaulted")
assert.Equal(t, d.MaxKeyBytes, cfg.MaxKeyBytes, "MaxKeyBytes should be defaulted")
assert.Equal(t, d.MaxInlineValue, cfg.MaxInlineValue, "MaxInlineValue should be defaulted")
assert.Equal(t, d.MemTableSize, cfg.MemTableSize, "MemTableSize should be defaulted")
assert.Equal(t, d.MaxImmutableCount, cfg.MaxImmutableCount, "MaxImmutableCount should be defaulted")
}
func TestValidateSegmentTooSmall(t *testing.T) {
cfg := WalConfig{MaxSegmentSize: 1024}
err := cfg.Validate()
require.Error(t, err, "MaxSegmentSize=1024 should fail validation")
assert.Contains(t, err.Error(), "too small")
}
+7
View File
@@ -0,0 +1,7 @@
{
"MD013": {
"line_length": 120,
"code_blocks": false,
"tables": false
}
}
+297
View File
@@ -0,0 +1,297 @@
# WAL Section 3.2 Oracle 审核报告 — Issues 清单
> 来源:Oracle 对 `docs/design.md` Section 3.2 WAL 的架构审核
> 日期:2026-06-09
---
## Critical Issues
### C1. WAL write failure 语义过于简化:`write()` 失败后 bytes 可能已落盘
**严重程度**: Critical
**位置**: Section 3.2 "写入流程" 步骤 ⑤ 及后续错误处理段落(~line 117)
**问题描述**:
当前设计将 WAL encode/write 失败统一当作 "definitely failed" 返回普通错误。但实际上存在两种不同情况:
1. **Encode 失败**(未触及 syscall):确实是 definitely failed,可以安全返回普通错误
2. **`write()` 失败**bytes 可能已进入 OS page cache 或部分写入文件):不是 definitely failed。Recovery 后可能发现该 batch CRC 合法并被重放,导致语义矛盾——调用方收到错误认为写入失败,但数据实际被恢复
**影响**: 调用方可能基于"写入失败"做非幂等业务决策(如放弃、走替代路径),但数据实际持久化了。
**建议修复**:
拆分 WAL write failure 处理:
```text
- encode 失败(未调用 write(): 普通错误,write-stopped
- write() 失败(bytes 可能已交给 OS: ErrCommitUnknown + write-stopped
除非实现能证明零 bytes 到达文件(例如 write 返回 0 且无副作用)
```
---
### C2. Segment 边界跨 Batch 行为未定义
**严重程度**: Critical
**位置**: Section 3.2 "Block 边界处理"~line 303)与 "Recovery 扫描流程"~line 436
**问题描述**:
WAL Batch 可拆成多个 Physical Record 跨多个 Block,但未定义 WAL Batch 是否可以跨 segment 文件。Recovery 中 incomplete fragment 的处理取决于是否位于"最后一个需要恢复的 segment"~line 493):
- 如果 Batch 可以跨 segmentrecovery 必须在 segment 间携带 fragment 收集状态(CollectingFragments),这增加了恢复复杂度
- 如果 Batch 不可跨 segment:需要显式约束 segment rotation 时机
当前 recovery 流程按 segment 顺序独立扫描,未定义跨 segment fragment 收集。
**影响**: 可能导致合法的跨 segment batch 被误判为中间损坏,或需要引入复杂的跨 segment 状态管理。
**建议修复**:
在 Section 3.2 明确添加约束:
```text
WAL Batch 不得跨 segment 文件。Segment rotation 只在 WAL Batch 边界发生。
当前 segment 写入完一个完整 WAL Batch 后,如果需要轮转,在下个 Batch 写入前切换到新 segment。
```
---
### C3. 预创建 segment 可破坏尾部截断逻辑
**严重程度**: Critical
**位置**: Section 3.2 "WAL 元数据持久化协议"~line 173)与 "Physical Record 解析规则"~line 491
**问题描述**:
新 segment 创建协议(步骤 1-8)允许 `segment-N+1.wal` 在文件系统上可见(已完成 rename + directory fsync),即使它尚未承载任何 batch。如果此时进程崩溃,`segment-N.wal` 可能有 crash-torn tail。
Recovery 扫描时,因为 `segment-N+1.wal` 存在且 header 合法,`segment-N` 不再被视为"最后一个需要恢复的 segment"。按照当前的尾部/中间损坏分类规则,`segment-N` 的尾部损坏会被升级为"WAL 中间损坏"→ 报错而非截断。
**影响**: 一个本应可截断恢复的尾部 partial write 场景被错误升级为不可恢复的中间损坏,导致整个 DB 无法启动。
**建议修复**:
方案 A(推荐):禁止预创建未来 segment,直到当前 segment 在完整 batch 边界 sealed
```text
新 segment 只在当前 active segment 写完一个完整 WAL Batch 后才创建。
确保 segment-N 永远在完整 batch 边界结束,segment-N+1 的创建不先于该 sealing。
```
方案 BRecovery 能识别并忽略空 segmentstartSequence == expectedSequence 但无任何 batch):
```text
如果 segment header 合法但不含任何 complete batch,且 startSequence == expectedSequence
视为空 segment,跳过或删除,继续扫描下一个 segment。
```
---
### C4. 恢复后未 fsync 确认的 Batch 可能变成已发布
**严重程度**: Critical
**位置**: Section 3.2 "恢复完成状态"~line 575)与 "持久化策略"~line 96
**问题描述**:
Recovery 重放所有 CRC 合法、sequence 连续的 WAL batch,并在恢复完成后将其全部标记为 `published`。但其中可能包含崩溃前从未 fsync 确认(调用方未收到成功)的 batch。
场景:
1. WAL bytes 已通过 `write()` 写入 OS page cache
2. 进程崩溃(非掉电),OS 将 page cache 刷盘
3. 重启后 recovery 发现该 batch 完整、CRC 合法、sequence 连续
4. 该 batch 被重放并标记为 published
5. 但调用方从未收到成功确认
**影响**: `Always` 策略下,调用方收到的语义是 "Put 返回成功 = 已持久化"。但如果进程 crash(非掉电),未确认的写入可能变成已发布。这是一个 API 语义问题而非数据安全问题。
**建议修复**:
在 Section 3.2 "持久化策略" 或 "恢复完成状态" 中显式声明:
```text
进程崩溃(非掉电)后恢复时,OS page cache 中已写入但尚未 fsync 的完整 WAL Batch
可能被恢复并视为已发布。这不是数据丢失,而是数据可见性前移。
调用方必须理解:进程崩溃重启后,比掉电场景可能多恢复一些写入。
如果需要严格区分"调用方已确认"与"未确认但存在于 WAL",需要后续引入
durable commit marker 或 confirmed-sequence 元数据。
```
---
### C5. Go 内存模型:lock-free read 需要原子发布机制
**严重程度**: Critical
**位置**: Section 3.3 MemTable "并发策略"~line 600)与 Section 3.2 步骤 ⑥⑧(~line 109-112
**问题描述**:
Section 3.3 声明 MemTable 使用 "Mutex 写 + 无锁读"Section 3.2 步骤 ⑥ 在 fsync 前将 entry 写入 MemTablepending 状态),步骤 ⑧ 通过更新 `publishedSequence` 使 entry 对无锁读可见。
在 Go 内存模型中:
1. **Skiplist 节点发布**:Mutex 保护下的写入对未持锁的并发读者不一定可见。需要 `atomic.Pointer` 或等效发布机制确保节点对读者可见。
2. **`publishedSequence` 更新**:作为普通变量写入,无锁读者可能看到过时值或部分写入。必须是 atomic 操作。
**影响**: 在 ARM 架构(弱内存序)上可能出现读者看到 `publishedSequence` 已更新但对应 skiplist 节点尚未可见的情况,导致读到不一致数据。
**建议修复**:
在 Section 3.2 或 3.3 中明确内存序要求:
```text
1. MemTable skiplist 节点必须通过 atomic storeatomic.Pointer 或自定义 release 操作)发布,
确保无锁读者看到完整的节点内容。
2. publishedSequence 必须是 atomic 变量(atomic.Uint64),
且其 Store 必须在所有 batch entries 的 skiplist 节点都已原子发布之后执行。
这保证读者先看到节点,再通过 publishedSequence 筛选可见 entry。
3. 读者必须先 atomic Load publishedSequence,再遍历 skiplist。
```
---
## Important Issues
### I1. MemTable 写入失败(Arena 满)在 WAL 写入后未覆盖
**严重程度**: Important
**位置**: Section 3.2 "写入流程" 步骤 ⑥(~line 109
**问题描述**:
写入流程步骤 ⑤(WAL encode/write)成功后,步骤 ⑥ 写入 MemTable 可能因为 Arena 满而失败。此时 WAL bytes 已持久化(或已在 page cache),但 MemTable 中没有对应 entry。设计文档未覆盖此场景。
**建议修复**:
```text
方案 A(推荐):在 WAL write 之前保证 MemTable 有足够容量。
写入前检查 Arena 剩余空间,不足时先冻结 MemTable 并创建新 MemTable。
Arena 预留必须考虑最大可能的 batch size。
方案 BMemTable 写入失败后按 ErrCommitUnknown + write-stopped 处理。
因为 WAL bytes 可能已持久化,不能按普通错误处理。
```
---
### I2. `CURRENT` 文件权威性与实际恢复模型不一致
**严重程度**: Important
**位置**: Section 3.2 "CURRENT / MANIFEST 权威性"~line 420)与 "WAL 元数据持久化协议"~line 181
**问题描述**:
设计明确声明 `CURRENT` 只是写入侧辅助文件,recovery 权威源是 `MANIFEST + 目录扫描`。但 durable-ready 协议要求在 segment 可承载写入前更新 `CURRENT` 并 fsync(步骤 6-7)。这意味着 `CURRENT` 更新是 batch 确认成功的前提之一,但 recovery 又不依赖它。
**建议修复**:
选择一种并保持一致:
```text
方案 A(推荐):简化 durable-ready 协议,移除 CURRENT 更新作为 batch 确认前提。
Recovery 通过 MANIFEST + 目录扫描发现 segment,CURRENT 仅作为写入侧快速定位优化。
新 segment 只需 rename + WAL directory fsync 即可进入 durable-ready。
方案 B:让 CURRENT 成为 recovery 的必要组件。
这样需要处理 CURRENT 损坏/缺失的 fallback,增加恢复复杂度。不推荐。
```
---
### I3. 尾部截断后缺少持久化步骤
**严重程度**: Important
**位置**: Section 3.2 "Physical Record 解析规则" 尾部损坏处理(~line 495
**问题描述**:
Recovery 允许截断最后一个 segment 的尾部损坏。但截断操作本身(`ftruncate` + 删除后续空 segment)需要 fsync 才能在再次崩溃时保持一致性。设计文档未说明截断后的持久化步骤。
**建议修复**:
在 "恢复完成状态" 之后或 "Recovery 扫描流程" 末尾添加:
```text
截断持久化步骤:
1. ftruncate active segment 到 lastCompleteBatchEnd
2. fsync truncated segment
3. 删除 startSequence == expectedSequence 但无 complete batch 的后续空 segment
4. fsync WAL directory
5. 更新 MANIFEST 记录恢复终点
6. fsync metadata directory
以上完成后,引擎才能开始接受新写入。
```
---
### I4. Batch 校验缺少资源上限
**严重程度**: Important
**位置**: Section 3.2 "WAL Batch 校验与重放"~line 530
**问题描述**:
Recovery 校验 `entryCount``entriesSize` 和 entry 边界,但未定义任何资源上限。恶意或损坏的 WAL 可能包含极大的 `entryCount``entriesSize`,导致 recovery OOM 或无限循环。
**建议修复**:
在 Section 3.2 添加硬性限制:
```text
WAL Batch 资源上限(可配置,建议默认值):
- entryCount: 最大 10,000
- entriesSize: 最大 4MB
- 单个 keyLen: 最大 4KB(不含 value
- 单个 valLen (Inline): 最大 4KB(超过走 ValueLogPointer
- fragment buffer: 最大 entriesSize 上限
- varint: 最大 5 bytesu64 varint 上限)
Recovery 解析时,超过任何上限即视为 WAL 损坏。
写入侧也必须遵守这些限制,超出拒绝写入。
```
---
### I5. `publishedSequence` 需要明确的内存序约束
**严重程度**: Important
**位置**: Section 3.2 "可见性语义"~line 148
**问题描述**:
`publishedSequence` 作为普通变量描述其语义,但未说明其在 Go 内存模型中的操作类型。多 goroutine 并发读写需要明确的 happens-before 关系。
**建议修复**:
在 "可见性语义" 小节补充:
```text
publishedSequence 的内存序约束:
1. 类型:atomic.Uint64(或等效原子变量)
2. 写入侧:Store 只在 WAL durability 和所有 MemTable 节点原子发布都完成后执行
3. 读取侧:Load 在遍历 MemTable 前执行,获得可见性 high-water mark
4. Happens-before 关系:
WAL fsync 完成 → MemTable 节点原子发布 → publishedSequence.Store
→ 读者 publishedSequence.Load → 遍历 MemTable 筛选可见 entry
```
---
## 变更追踪
| Issue | 类型 | 优先级 | 状态 |
|-------|------|--------|------|
| C1 | 语义正确性 | Critical | Open |
| C2 | 格式完整性 | Critical | Open |
| C3 | 恢复正确性 | Critical | Open |
| C4 | API 语义 | Critical | Open |
| C5 | 内存安全 | Critical | Open |
| I1 | 错误处理完整性 | Important | Open |
| I2 | 设计一致性 | Important | Open |
| I3 | 持久化完整性 | Important | Open |
| I4 | 安全性/鲁棒性 | Important | Open |
| I5 | 内存序正确性 | Important | Open |
+640
View File
@@ -0,0 +1,640 @@
# Phase 1: WAL 子系统开发方案
基于 `docs/design.md` §3.2 设计文档。
## 目标
实现完整的 WAL(预写日志)子系统,使其能够支撑单 key autocommit 的写入、崩溃恢复和读可见性语义。
## 开发阶段总览
```
Phase 1A: 项目骨架 + WAL 编码格式层
Phase 1B: WAL 文件写入 + Segment 管理
Phase 1C: WAL WriterGroup Commit
Phase 1D: WAL Recovery
Phase 1E: MemTableSkipList + Arena
Phase 1F: 写入路径集成(WAL → MemTable 完整流水线)
Phase 1G: 读路径 + 嵌入式 API
Phase 1H: MANIFEST + 文件管理
Phase 1I: 集成测试 + Benchmark
```
---
## Phase 1A: 项目骨架 + WAL 编码格式层
**目标**: 建立 Go 项目结构,实现 WAL 物理格式(Block / Physical Record / WAL Batch / Entry)的编码与解码。
### 任务
#### 1A-1: 项目初始化
- `go.mod` 初始化(模块名 `github.com/dailz/go-kv`
- 目录结构:
```
go-kv/
├── go.mod
├── wal/ # WAL 子系统
│ ├── wal.go # 公共类型、常量、配置
│ ├── record.go # Physical Record 编解码
│ ├── batch.go # WAL Batch 编解码
│ ├── entry.go # Entry 编解码
│ ├── header.go # WAL File Header 编解码
│ └── wal_test.go
├── memtable/ # MemTablePhase 1E
├── config/ # 全局配置
├── errors.go # 公共错误类型
└── db.go # DB 入口
```
- `.golangci.yml` 配置(参考 golang-lint skill
#### 1A-2: 公共错误类型 (`errors.go`)
- `ErrCommitUnknown` — maybe committed 语义
- `ErrWriteStopped` — 引擎 write-stopped
- `ErrSequenceExhausted` — sequence 耗尽
- `ErrWALCorrupted` — WAL 损坏
- `ErrInvalidConfig` — 配置不合法
#### 1A-3: WAL 常量与配置 (`wal/wal.go`)
```go
const (
WalMagic uint32 = 0x... // 待定
WalFormatVersion uint16 = 1
WalFileHeaderSize = 32
WalBlockSize = 32 * 1024 // 32KB
PhysicalRecordHeaderSize = 7
WalBatchHeaderSize = 18
MaxWalBatchEntryCount = 10_000
MaxWalBatchEntriesSize = 4 * 1024 * 1024 // 4MB
MaxWalKeyBytes = 4 * 1024 // 4KB
MaxWalInlineValueBytes = 4 * 1024 // 4KB
MaxWalVarintBytes = 5
DefaultMaxWalSegmentSize = 64 * 1024 * 1024 // 64MB
DefaultImmutableCount = 2
)
// Fragment types
const (
RecInvalid uint8 = 0
RecFull uint8 = 1
RecFirst uint8 = 2
RecMiddle uint8 = 3
RecLast uint8 = 4
)
// OpType
const (
OpInvalid uint8 = 0
OpPut uint8 = 1
OpDelete uint8 = 2
)
// ValueKind
const (
VKNone uint8 = 0
VKInline uint8 = 1
VKValueLogPointer uint8 = 2
)
```
WAL 配置结构体:
```go
type WalConfig struct {
MaxSegmentSize uint64 // default 64MB
BlockSize uint32 // default 32KB
SyncMode SyncMode // Always/Periodic/Never
PeriodicSyncMs uint32 // Periodic 模式的 fsync 间隔
MaxBatchEntries uint32 // default 10000
MaxBatchSize uint32 // default 4MB
MaxKeyBytes uint32 // default 4KB
MaxInlineValue uint32 // default 4KB
}
```
配置校验函数 — 必须在 DB 打开时验证不变量:
```text
maxWalSegmentPayload >= maxEncodedWalBatchSize + worstCasePhysicalRecordOverhead + worstCaseBlockPadding
```
#### 1A-4: WAL File Header 编解码 (`wal/header.go`)
- `WalFileHeader` 结构体:magic, formatVersion, headerSize, blockSize, segmentID, startSequence, headerCRC
- `EncodeWalHeader(h *WalFileHeader) [WalFileHeaderSize]byte`
- `DecodeWalHeader(data []byte) (*WalFileHeader, error)` — 校验 magic、formatVersion、headerSize、headerCRC
- CRC 覆盖范围:magic 到 startSequence,不包含 headerCRC 自身
- 字节序:little-endian
#### 1A-5: Physical Record 编解码 (`wal/record.go`)
- `PhysicalRecord` 结构体:CRC, Length, Type, Payload
- `EncodePhysicalRecord(recType uint8, payload []byte) []byte` — 返回编码后的 bytes
- `DecodePhysicalRecord(data []byte) (*PhysicalRecord, error)` — CRC 校验
- Block 边界处理辅助函数:
- `PaddingNeeded(blockOffset, blockSize uint32) int` — 剩余空间 <= 7 时返回需要 padding 的字节数
- `CanFitRecord(blockOffset, blockSize, payloadLen uint32) bool`
#### 1A-6: WAL Batch 编解码 (`wal/batch.go`)
- `WalBatch` 结构体:Flags, BaseSequence, EntryCount, EntriesSize, Entries
- `EncodeWalBatch(batch *WalBatch) ([]byte, error)` — 编码 Batch Header + Entries
- `DecodeWalBatch(data []byte) (*WalBatch, error)` — 校验 flags、entryCount、entriesSize
- Batch 分片:`SplitIntoRecords(encodedBatch []byte, blockSize uint32) [][]byte` — 将编码后的 Batch 拆分为 Physical Record payloads
- Batch 重组:`FragmentCollector` — 收集 fragments 并重组成完整 Batch
FragmentCollector 状态机:
```
Idle → 收到 Full → 重放 batch → Idle
Idle → 收到 First → CollectingFragments
CollectingFragments → 收到 Middle → 追加
CollectingFragments → 收到 Last → 重组 → 重放 → Idle
```
#### 1A-7: Entry 编解码 (`wal/entry.go`)
- `WalEntry` 结构体:OpType, ValueKind, Key, Value
- `EncodeEntry(e *WalEntry) ([]byte, error)` — 编码为 varint 长度 + bytes
- `DecodeEntry(data []byte) (*WalEntry, int, error)` — 解码,返回 entry 和 consumed bytes
- 校验规则:
- keyLen > 0 && keyLen <= maxKeyBytes
- Put 要求 valueKind ∈ {Inline, ValueLogPointer}
- Put + Inline: valLen <= maxInlineValueBytes (允许 valLen = 0)
- Put + ValueLogPointer: valLen > 0
- Delete: valueKind == None, valLen == 0
#### 1A-8: WAL Batch 资源校验
- `ValidateBatchLimits(entries []*WalEntry) error` — 在 sequence 分配之前检查:
- entryCount <= maxBatchEntries
- 每个 keyLen <= maxKeyBytes
- 每个 inline valLen <= maxInlineValueBytes
- entries 编码后总大小 <= maxBatchSize
- 单个 Batch 的最坏 Physical Record overhead 不超过 segment capacity
### 验收标准
- [ ] 所有编解码函数有 table-driven test
- [ ] CRC 校验正确
- [ ] Fragment 分片/重组 round-trip 正确
- [ ] 资源限制校验覆盖所有边界条件
- [ ] `go vet` / `golangci-lint` 通过
---
## Phase 1B: WAL 文件写入 + Segment 管理
**目标**: 实现 WAL segment 文件的写入、轮转和持久化协议。
### 任务
#### 1B-1: Segment 文件格式写入器 (`wal/segment_writer.go`)
- `SegmentWriter` — 封装 WAL segment 文件的追加写入
- 状态:当前 segment fd、当前 block offset、当前 segmentID、payload written bytes
- `NewSegmentWriter(dir string, segmentID uint64, startSequence uint64, cfg *WalConfig) (*SegmentWriter, error)`
- 创建 segment-N.wal.tmp
- 写入 WAL File Header
- fsync
- rename → segment-N.wal
- fsync directory
- 进入 durable-ready 状态
- `AppendBatch(batch *WalBatch) error` — 编码 batch → split into records → 按 block 边界写入
- `Sync() error` — fsync 当前 segment 文件
- `Close() error`
- `RemainingPayload() uint64` — 当前 segment 剩余可用 payload 空间
- `CurrentOffset() uint64` — 当前写入偏移
#### 1B-2: Block 写入缓冲 (`wal/block_writer.go`)
- 管理 32KB block 的填充和 padding
- `BlockWriter` — 封装 block 内的 Physical Record 写入
- 自动处理 block 边界:剩余 <= 7 bytes 时 padding
- 跨 block 的 batch fragment 自动拆分
#### 1B-3: Segment 轮转逻辑
- 写入 batch 前检查 `RemainingPayload()` 是否足够容纳整个 batch
- 不足时:当前 segment 完成(在 batch 边界)、创建新 segment
- 新 segment 的 durable-ready 协议:
1. create segment-N+1.wal.tmp
2. write WAL File Header(含 startSequence = nextExpectedSequence
3. fsync segment-N+1.wal.tmp
4. rename → segment-N+1.wal
5. fsync WAL directory
6. segment-N+1 进入 durable-ready
- 旧的 active segment 密封
#### 1B-4: CURRENT 文件管理
- best-effort 更新 CURRENT 文件
- temp + rename 模式
- 更新失败不影响已 durable-ready 的 segment
#### 1B-5: WAL 目录管理工具
- 扫描 WAL 目录中的 segment 文件
- 按 segmentID 排序
- 解析文件名中的 segmentID
- 文件名格式:`segment-{id}.wal`
### 验收标准
- [ ] Segment 创建遵循 durable-ready 协议
- [ ] Batch 不跨 segment
- [ ] Block padding 正确
- [ ] Segment 轮转在 batch 边界发生
- [ ] 多 segment 写入后,每个 segment 的 header 可以正确解析
- [ ] 测试覆盖:正常写入、跨 block batch、segment 轮转触发
---
## Phase 1C: WAL WriterGroup Commit
**目标**: 实现完整的 WAL 写入路径,包括 group commit、sequence 管理、fsync 策略和错误分类。
### 任务
#### 1C-1: Sequence 管理器 (`wal/sequence.go`)
- `SequenceManager` — 管理 WAL 物理 mutation sequence
- `atomic.Uint64` 存储 nextSequence、publishedSequence、durableSequence
- `AllocateBatch(count uint32) (baseSequence uint64, err error)` — checked arithmetic 检查溢出
- `Publish(sequence uint64)` — release 语义 store publishedSequence
- `MarkDurable(snapshot SegmentEndState)` — 推进 durableSequence
- `Published() uint64` — load publishedSequence
- `Durable() uint64` — load durableSequence
#### 1C-2: Commit Queue (`wal/commit_queue.go`)
- 写请求进入的队列
- 每个写请求关联一个 `*sync.Cond` 或 channel 用于等待/唤醒
- `CommitBatch` 结构体:entries、完成 channel、错误结果、baseSequence
#### 1C-3: WAL Writer 主循环 (`wal/writer.go`)
核心写入循环:
```
loop:
1. 从 commit queue 收集一批写入
2. 等待触发条件(500µs 或 32KB)或 queue 非空
3. 组装 WAL Batch
4. 校验 batch 资源限制
5. 预留 MemTable Arena 容量
6. 分配 sequencebaseSequence
7. 在私有缓冲区编码 WAL Batch
8. 检查/触发 segment 轮转
9. Append WAL Batch 到 segment 文件
10. 写入 MemTablepending/unpublished
11. fsyncAlways 模式)
12. 发布 publishedSequence
13. 唤醒所有等待的调用方
```
错误分类逻辑:
- 步骤 4-7 失败(未分配 sequence)→ 普通错误,可继续
- 步骤 6 后失败(sequence 已分配)→ write-stopped
- 步骤 9 后失败(WAL write 已尝试)→ ErrCommitUnknown + write-stopped
- 步骤 11 失败(fsync)→ ErrCommitUnknown + write-stopped
#### 1C-4: Fsync 策略实现 (`wal/fsync.go`)
- `SyncMode` 类型:Always / Periodic / Never
- `Always`: 每次 batch fsync 后再 publish
- `Periodic`: 后台 goroutine 定期 fsyncwrite 成功即可 publish
- `Never`: 不主动 fsync
- `Periodic` 的 fsync worker
- 快照当前 append high-water mark: (segmentID, endOffset, endSequence)
- fsync 成功后按连续 batch 推进 durableSequence
- fsync 失败 → write-stopped
#### 1C-5: durableSequence 推进逻辑
- 每个 batch 记录 `(segmentID, endOffset, endSequence)`
- fsync snapshot 后只推进满足条件的最大连续 batch
- 跨 segment 推进需要 segment 已 durable-ready
#### 1C-6: Write-Stopped 状态管理
- `atomic.Bool` 存储 writeStopped
- 进入 write-stopped 后拒绝新写入
- 已存在的 MemTable / Immutable MemTable 可继续后台处理
- 提供 `IsWriteStopped() bool` 查询接口
### 验收标准
- [ ] Group commit 正确合并多个写请求
- [ ] 双触发(时间/大小)工作正常
- [ ] Sequence 分配无溢出
- [ ] Always 模式下 publish 在 fsync 之后
- [ ] 错误分类准确(普通错误 / write-stopped / ErrCommitUnknown
- [ ] 并发写入正确(多 goroutine 同时 Put
- [ ] Write-stopped 后新写入被拒绝
---
## Phase 1D: WAL Recovery
**目标**: 实现 WAL 崩溃恢复,包括 segment 扫描、fragment 重组、batch 校验和尾部截断。
### 任务
#### 1D-1: Segment 扫描器 (`wal/scanner.go`)
- 从 WAL 目录扫描 segment 文件
- 按 segmentID 排序
- 从 MANIFEST 指定的 recoverySegmentID 开始
- 过滤掉 segmentID < recoverySegmentID 的旧 segment
- 校验连续性:segmentID 和 startSequence 都必须连续
#### 1D-2: Physical Record 解析器 (`wal/record_parser.go`)
- Block 级别的顺序解析
- 处理 padding(全 0 校验)
- Physical Record header 解析和 CRC 校验
- 错误分类:尾部 vs 中间损坏
#### 1D-3: Fragment 重组器 (`wal/fragment_collector.go`)
- 实现 Idle / CollectingFragments 状态机
- 收集 First / Middle / Last fragments
- Buffer 大小限制(Batch Header 长度 + entriesSize 上限)
- Fragment 顺序合法性检查
#### 1D-4: Batch 校验与重放 (`wal/recovery.go`)
- Batch Header 校验:flags、entryCount、entriesSize
- Batch sequence 连续性:batch.baseSequence == expectedSequence
- Entry 逐条校验:opType、valueKind、keyLen、valLen
- 重放回调:对每个合法 entry 调用 replay 函数
- Sequence 推进:expectedSequence += entryCount
#### 1D-5: 尾部截断持久化 (`wal/truncation.go`)
- 识别最后一个完整 batch 的结束位置
- ftruncate segment 文件
- fsync 被截断的 segment
- 删除不含任何 complete batch 的后续空 segment
- fsync WAL directory
- 任一步失败 → recovery 报错
#### 1D-6: Recovery 主流程 (`wal/recovery.go`)
```
1. 读取 MANIFEST → recoverySegmentID
2. 扫描 WAL 目录 → 过滤出 recovery segments
3. 排序并校验连续性
4. 逐 segment 扫描:
a. 校验 File Header
b. 逐 Block 解析 Physical Records
c. Fragment 重组 → 完整 Batch
d. Batch 校验 → 重放
e. 更新 expectedSequence
5. 处理尾部异常
6. 持久化截断(如需要)
7. 返回恢复结果:recoveredSequence, nextSequence, publishedSequence
```
### 验收标准
- [ ] 正常 WAL 完整恢复
- [ ] 尾部 partial write 正确截断
- [ ] 中间损坏正确报错
- [ ] 跨 segment 恢复正确
- [ ] Fragment 重组 round-trip 正确
- [ ] Segment 连续性校验
- [ ] Sequence 溢出检测
- [ ] 资源限制校验(recovery 侧)
---
## Phase 1E: MemTableSkipList + Arena
**目标**: 实现基于 Arena 的 SkipList,支持 pending/unpublished/aborted 状态,容量预留,和原子发布。
### 任务
#### 1E-1: Arena 分配器 (`memtable/arena.go`)
- 固定大小 Arena(默认 64MB
- 线程安全的内存分配
- 对齐分配
- 剩余容量查询
- 支持预留(reserve)操作
#### 1E-2: SkipList (`memtable/skiplist.go`)
- 最大 20 层
- Mutex 写 + 无锁读
- `atomic.Pointer` 发布 next 指针(release 语义)
- 有序遍历(Iterator
- key 比较(bytes comparison
#### 1E-3: Entry 状态管理 (`memtable/entry.go`)
- Entry 结构:key、value、sequence、pending/aborted 标记
- 原子发布:`atomic.Pointer` store-release
- 可见性判断:`entry.sequence <= loadedPublishedSequence && !aborted`
#### 1E-4: MemTable (`memtable/memtable.go`)
- 封装 SkipList + Arena
- `Put(key, value, sequence) error` — 写入 pending entry
- `PublishEntries(upToSequence)` — 批量发布 pending entries
- `AbortEntries(fromSequence)` — 标记 aborted
- `Get(key, publishedSequence) (GetResult, error)` — 无锁读,只返回 sequence <= publishedSequence 且非 aborted 的 entry
- `NewIterator(publishedSequence) Iterator` — 无锁有序遍历
- `ApproximateSize() uint64` — 近似内存使用量
- `IsFull() bool`
- `Reserve(entries []ReserveEntry) (uint64, error)` — 容量预留(最坏情况计算)
#### 1E-5: 容量预留计算
- 每个 entry 的预留大小 = key bytes + value bytes + skiplist node overhead + next 指针数组(最大层高)+ arena 对齐 padding
- 批量预留必须覆盖整个 batch
- checked arithmetic 检查单个 batch 是否超过空 MemTable 容量
#### 1E-6: Immutable MemTable 管理
- Freeze 流程:当前 MemTable → Immutable
- Immutable 队列(上限 2
- 队列满时阻塞
### 验收标准
- [ ] SkipList 正确性:插入、查找、有序遍历
- [ ] Arena 分配无泄漏
- [ ] 并发读写正确(racetest
- [ ] pending/unpublished entry 对读不可见
- [ ] 发布后 entry 可见
- [ ] aborted entry 对读不可见
- [ ] 容量预留准确
- [ ] 内存序正确(go test -race 通过)
---
## Phase 1F: 写入路径集成
**目标**: 将 WAL Writer 和 MemTable 连通,实现完整的写入流水线。
### 任务
#### 1F-1: DB 写入 API (`db.go`)
```go
type DB struct { ... }
func Open(dir string, opts ...Option) (*DB, error)
func (db *DB) Close() error
func (db *DB) Put(key, value []byte) error
func (db *DB) Delete(key []byte) error
```
#### 1F-2: 写入路径集成
完整写入流程:
```
Put(key, value)
→ commit queue
→ group commit 组装 batch
→ 校验 batch limits
→ 预留 MemTable Arena
→ 分配 sequence
→ 私有缓冲编码
→ 检查 segment 轮转
→ WAL append
→ MemTable pending write
→ fsyncAlways 模式)
→ 原子发布 MemTable entries
→ 推进 publishedSequence
→ 唤醒调用方
```
#### 1F-3: MemTable Freeze + Switch
- 写入前检查容量,不足时 freeze + switch
- Immutable 队列满时阻塞等待
- Freeze 时确保当前 MemTable 已完成所有 pending 发布
#### 1F-4: 恢复启动集成
- Open 时执行 recovery
- 恢复的 entries 写入 MemTable 并标记为 published
- 设置 nextSequence、publishedSequence
### 验收标准
- [ ] 单条 Put 写入成功
- [ ] 并发 Put 正确
- [ ] 写入后读取可见(Always 模式)
- [ ] WAL crash recovery 后数据完整
- [ ] MemTable freeze/switch 正确
- [ ] Sequence 连续无间隙
- [ ] `go test -race` 通过
---
## Phase 1G: 读路径 + 嵌入式 API
**目标**: 实现完整的读路径和嵌入式 API。
### 任务
#### 1G-1: Get API
```go
type GetResult struct {
Value []byte
Found bool
}
func (db *DB) Get(key []byte) (GetResult, error)
```
#### 1G-2: 读路径实现
- 读取 publishedSequenceatomic load
- 遍历 MemTable → Immutable MemTables
- 只返回 `sequence <= publishedSequence` 且非 aborted 的 entry
- Delete (tombstone) 返回 `Found=false`
#### 1G-3: 辅助 API
```go
func (db *DB) GetDurableSequence() uint64
func (db *DB) IsWriteStopped() bool
```
### 验收标准
- [ ] Put 后 Get 返回正确值
- [ ] Delete 后 Get 返回 Found=false
- [ ] 空 value 正确区分(Found=true, Value=[]byte{}
- [ ] 并发读写正确
- [ ] 未发布 entry 对 Get 不可见
---
## Phase 1H: MANIFEST + 文件管理
**目标**: 实现 MANIFEST 持久化和 WAL segment 生命周期管理。
### 任务
#### 1H-1: MANIFEST 格式
- 记录 recovery 起点的 recoverySegmentID
- temp + rename 原子更新
- MANIFEST 只在 checkpointMemTable flush)后推进
#### 1H-2: 首次创建 DB 流程
- 创建目录结构
- 创建初始 MANIFESTrecoverySegmentID=0
- 创建初始 WAL segment
#### 1H-3: WAL Segment 生命周期
- 旧 segment 删除条件:已被 MANIFEST checkpoint 覆盖
- 删除顺序:先删除文件,再 fsync directory
### 验收标准
- [ ] 首次创建 DB 成功
- [ ] 重复打开 DB 正确恢复
- [ ] MANIFEST 原子更新
- [ ] 旧 WAL segment 正确清理
---
## Phase 1I: 集成测试 + Benchmark
**目标**: 端到端测试和性能基准。
### 任务
#### 1I-1: 集成测试
- 正常写入 + 读取 round-trip
- 并发写入 + 读取一致性
- 崩溃恢复(kill -9 模拟)
- WAL 尾部损坏恢复
- Write-stopped 后的行为
- Sequence 耗尽处理
- 配置校验拒绝非法配置
- 空 value 写入/读取
- 大量数据写入(触发 segment 轮转)
#### 1I-2: Benchmark
- 单线程 Put 吞吐
- 多线程 Put 吞吐
- 单线程 Get 延迟
- 多线程 Get 延迟
- WAL Recovery 时间
- 写入放大测量
#### 1I-3: Race Condition 测试
- `go test -race -count=100`
- 并发 Put + Get
- 并发 Put + Close
### 验收标准
- [ ] 所有集成测试通过
- [ ] Benchmark 数字可作为后续优化基线
- [ ] Race test 无 data race
---
## 依赖关系与并行度
```
1A ─────┐
├── 1B ─────┐
│ ├── 1C ─────┐
│ │ ├── 1F ── 1G ── 1I
│ │ │
1A ── 1E ──────────┘ │
│ │
├── 1D ─────────────────┘
└── 1H ────────────────────────────── 1I
```
可并行开发的模块:
- 1A 完成后,1B/1D/1E/1H 可以并行开发
- 1B 完成后,1C 可以开始
- 1C + 1D + 1E 完成后,1F 可以集成
- 1F + 1H 完成后,1G 可以集成
- 所有完成后,1I 集成测试
## 技术要点备忘
### 内存序(最关键)
- skiplist next 指针:`atomic.Pointer` store-release
- `publishedSequence``atomic.Uint64` store(在所有 entry 节点发布后)
- 读者先 load publishedSequenceacquire),再遍历 skiplist
### WAL 副作用边界
- 未调用 `write()` → 普通错误
- 已调用 `write()` → ErrCommitUnknown + write-stopped
- 私有缓冲区编码,不共享 bufio.Writer
### ErrCommitUnknown 语义
- maybe committed,不是 definitely failed
- 不盲目重试
- 第一阶段为弱确认
### WAL Batch 资源前置校验
- sequence 分配之前完成所有可失败校验
- 减少 write-stopped 触发机会
+22
View File
@@ -0,0 +1,22 @@
package go_kv
import "errors"
// ErrCommitUnknown indicates the commit result is indeterminate: the WAL write
// was attempted but the caller cannot assume the write definitely failed or
// succeeded. The caller must query the commit state before retrying.
var ErrCommitUnknown = errors.New("go-kv: commit result unknown")
// ErrWriteStopped indicates the engine has entered a write-stopped state and
// rejects all subsequent writes.
var ErrWriteStopped = errors.New("go-kv: write stopped")
// ErrSequenceExhausted indicates the sequence number space is exhausted. The
// engine enters a terminal state requiring database migration or rebuild.
var ErrSequenceExhausted = errors.New("go-kv: sequence exhausted")
// ErrWALCorrupted indicates WAL data corruption was detected.
var ErrWALCorrupted = errors.New("go-kv: WAL corrupted")
// ErrInvalidConfig indicates an invalid configuration was provided.
var ErrInvalidConfig = errors.New("go-kv: invalid config")
+42
View File
@@ -0,0 +1,42 @@
package go_kv
import (
"errors"
"fmt"
"testing"
)
func TestErrorTypes(t *testing.T) {
allErrors := []error{
ErrCommitUnknown,
ErrWriteStopped,
ErrSequenceExhausted,
ErrWALCorrupted,
ErrInvalidConfig,
}
// Each error must match itself via errors.Is.
for _, err := range allErrors {
if !errors.Is(err, err) {
t.Errorf("errors.Is(%v, %v) = false, want true", err, err)
}
}
// Each error must NOT match any other error.
for i, a := range allErrors {
for j, b := range allErrors {
if i == j {
continue
}
if errors.Is(a, b) {
t.Errorf("errors.Is(%v, %v) = true, want false (distinct errors)", a, b)
}
}
}
// Wrapped errors must still be identifiable via errors.Is.
wrapped := fmt.Errorf("operation failed: %w", ErrCommitUnknown)
if !errors.Is(wrapped, ErrCommitUnknown) {
t.Errorf("errors.Is(wrapped ErrCommitUnknown, ErrCommitUnknown) = false, want true")
}
}
+11
View File
@@ -0,0 +1,11 @@
module github.com/dailz/go-kv
go 1.26.3
require github.com/stretchr/testify v1.11.1
require (
github.com/davecgh/go-spew v1.1.1 // indirect
github.com/pmezard/go-difflib v1.0.0 // indirect
gopkg.in/yaml.v3 v3.0.1 // indirect
)
+10
View File
@@ -0,0 +1,10 @@
github.com/davecgh/go-spew v1.1.1 h1:vj9j/u1bqnvCEfJOwUhtlOARqs3+rkHYY13jYWTU97c=
github.com/davecgh/go-spew v1.1.1/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
github.com/pmezard/go-difflib v1.0.0 h1:4DBwDE0NGyQoBHbLQYPwSUPoCMWR5BEzIk/f1lZbAQM=
github.com/pmezard/go-difflib v1.0.0/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4=
github.com/stretchr/testify v1.11.1 h1:7s2iGBzp5EwR7/aIZr8ao5+dra3wiQyKjjFuvgVKu7U=
github.com/stretchr/testify v1.11.1/go.mod h1:wZwfW3scLgRK+23gO65QZefKpKQRnfz6sD981Nm4B6U=
gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405 h1:yhCVgyC4o1eVCa2tZl7eS0r+SDo693bJlVdllGtEeKM=
gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0=
gopkg.in/yaml.v3 v3.0.1 h1:fxVm/GzAzEWqLHuvctI91KS9hhNmmWOoWu0XTYJS7CA=
gopkg.in/yaml.v3 v3.0.1/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM=
+43
View File
@@ -0,0 +1,43 @@
package manifest
import (
"fmt"
"os"
"strconv"
"strings"
)
// WriteCurrent writes the CURRENT file to dir with a best-effort atomic rename.
// The file contains the active WAL segment filename (e.g. "segment-5.wal").
// CURRENT is only a write-side hint; it may be missing or stale after a crash.
func WriteCurrent(dir string, segmentID uint64) error {
content := fmt.Sprintf("segment-%d.wal\n", segmentID)
tmpPath := dir + "/CURRENT.tmp"
if err := os.WriteFile(tmpPath, []byte(content), 0o644); err != nil {
return fmt.Errorf("write current tmp: %w", err)
}
if err := os.Rename(tmpPath, dir+"/CURRENT"); err != nil {
return fmt.Errorf("rename current: %w", err)
}
return nil
}
// ReadCurrent reads the CURRENT file from dir and returns the segment ID.
// If the file does not exist or cannot be parsed, it returns 0, false with no error.
func ReadCurrent(dir string) (segmentID uint64, ok bool) {
data, err := os.ReadFile(dir + "/CURRENT")
if err != nil {
return 0, false
}
line := strings.TrimSpace(string(data))
// Expected format: "segment-N.wal"
if !strings.HasPrefix(line, "segment-") || !strings.HasSuffix(line, ".wal") {
return 0, false
}
idStr := strings.TrimSuffix(strings.TrimPrefix(line, "segment-"), ".wal")
id, err := strconv.ParseUint(idStr, 10, 64)
if err != nil {
return 0, false
}
return id, true
}
+2
View File
@@ -0,0 +1,2 @@
// Package manifest manages database metadata and checkpoint information.
package manifest
+52
View File
@@ -0,0 +1,52 @@
package manifest
import (
"fmt"
"os"
"strconv"
"strings"
)
// Manifest holds database metadata used for recovery.
// The MANIFEST file stores the recovery checkpoint so that
// recovery knows which segments are already confirmed durable.
type Manifest struct {
RecoverySegmentID uint64
}
// Load reads the MANIFEST file from dir.
// If the file does not exist, it returns a zero-value Manifest with no error (fresh DB).
func Load(dir string) (*Manifest, error) {
path := dir + "/MANIFEST"
data, err := os.ReadFile(path)
if err != nil {
if os.IsNotExist(err) {
return &Manifest{RecoverySegmentID: 0}, nil
}
return nil, fmt.Errorf("read manifest: %w", err)
}
line := strings.TrimSpace(string(data))
if !strings.HasPrefix(line, "recovery_segment_id:") {
return nil, fmt.Errorf("manifest: invalid format: %q", line)
}
idStr := strings.TrimPrefix(line, "recovery_segment_id:")
id, err := strconv.ParseUint(idStr, 10, 64)
if err != nil {
return nil, fmt.Errorf("manifest: parse recovery_segment_id: %w", err)
}
return &Manifest{RecoverySegmentID: id}, nil
}
// Save atomically writes the MANIFEST file to dir with the given recoverySegmentID.
func Save(dir string, recoverySegmentID uint64) error {
content := fmt.Sprintf("recovery_segment_id:%d\n", recoverySegmentID)
tmpPath := dir + "/MANIFEST.tmp"
if err := os.WriteFile(tmpPath, []byte(content), 0o644); err != nil {
return fmt.Errorf("write manifest tmp: %w", err)
}
if err := os.Rename(tmpPath, dir+"/MANIFEST"); err != nil {
return fmt.Errorf("rename manifest: %w", err)
}
return nil
}
+45
View File
@@ -0,0 +1,45 @@
package manifest
import (
"testing"
"github.com/stretchr/testify/assert"
"github.com/stretchr/testify/require"
)
func TestManifestRoundtrip(t *testing.T) {
dir := t.TempDir()
err := Save(dir, 42)
require.NoError(t, err)
m, err := Load(dir)
require.NoError(t, err)
assert.Equal(t, uint64(42), m.RecoverySegmentID)
}
func TestManifestMissing(t *testing.T) {
dir := t.TempDir()
m, err := Load(dir)
require.NoError(t, err)
assert.Equal(t, uint64(0), m.RecoverySegmentID)
}
func TestCurrentRoundtrip(t *testing.T) {
dir := t.TempDir()
err := WriteCurrent(dir, 5)
require.NoError(t, err)
segmentID, ok := ReadCurrent(dir)
assert.True(t, ok)
assert.Equal(t, uint64(5), segmentID)
}
func TestCurrentMissing(t *testing.T) {
dir := t.TempDir()
_, ok := ReadCurrent(dir)
assert.False(t, ok)
}
+2
View File
@@ -0,0 +1,2 @@
// Package memtable implements an in-memory sorted key-value store.
package memtable
+80
View File
@@ -0,0 +1,80 @@
package wal
// WAL file format constants.
const (
// WalMagic is the file type identifier for WAL segment files ("WALK").
WalMagic uint32 = 0x57414C4B
// WalFormatVersion is the WAL file format version. First version is 1.
WalFormatVersion uint16 = 1
// WalFileHeaderSize is the size of the WAL file header in bytes.
// Fields: magic(4) + formatVersion(2) + headerSize(2) + blockSize(4) +
// segmentID(8) + startSequence(8) + headerCRC(4) = 32.
WalFileHeaderSize = 32
// WalBlockSize is the fixed size of each WAL block in bytes (32 KB).
WalBlockSize = 32 * 1024
// PhysicalRecordHeaderSize is the size of a physical record header in bytes.
// Fields: crc32c(4) + length(2) + type(1) = 7.
PhysicalRecordHeaderSize = 7
// WalBatchHeaderSize is the size of a WAL batch header in bytes.
// Fields: flags(2) + baseSequence(8) + entryCount(4) + entriesSize(4) = 18.
WalBatchHeaderSize = 18
// MaxWalBatchEntryCount limits the number of entries in a single batch.
MaxWalBatchEntryCount uint32 = 10000
// MaxWalBatchEntriesSize limits the total size of the entries region in bytes (4 MB).
MaxWalBatchEntriesSize uint32 = 4 * 1024 * 1024
// MaxWalKeyBytes limits the size of a single key in bytes (4 KB).
MaxWalKeyBytes uint32 = 4 * 1024
// MaxWalInlineValueBytes limits the size of an inline value in bytes (4 KB).
// Values exceeding this must use ValueLogPointer.
MaxWalInlineValueBytes uint32 = 4 * 1024
// MaxWalVarintBytes is the maximum encoded length of a varint field.
MaxWalVarintBytes = 5
// DefaultMaxWalSegmentSize is the default maximum size of a WAL segment file (64 MB).
DefaultMaxWalSegmentSize uint64 = 64 * 1024 * 1024
)
// Fragment types for physical records.
const (
// RecInvalid is an illegal fragment type used for corruption detection.
RecInvalid uint8 = 0
// RecFull indicates a complete WAL batch in a single physical record.
RecFull uint8 = 1
// RecFirst is the first fragment of a multi-record WAL batch.
RecFirst uint8 = 2
// RecMiddle is a middle fragment (may appear zero or more times).
RecMiddle uint8 = 3
// RecLast is the last fragment of a multi-record WAL batch.
RecLast uint8 = 4
)
// OpType represents the operation type of a WAL entry.
const (
// OpInvalid is an illegal operation type used for corruption detection.
OpInvalid uint8 = 0
// OpPut represents a key-value put operation.
OpPut uint8 = 1
// OpDelete represents a key deletion operation.
OpDelete uint8 = 2
)
// ValueKind represents how the value field is encoded in a WAL entry.
const (
// VKNone indicates no value (used with Delete operations).
VKNone uint8 = 0
// VKInline indicates the value field contains inline user bytes.
VKInline uint8 = 1
// VKValueLogPointer indicates the value field contains an encoded Value Log pointer.
VKValueLogPointer uint8 = 2
)
+72
View File
@@ -0,0 +1,72 @@
package wal
import "testing"
func TestConstantValues(t *testing.T) {
tests := []struct {
name string
got interface{}
expected interface{}
}{
{"WalBlockSize", WalBlockSize, 32 * 1024},
{"WalFileHeaderSize", WalFileHeaderSize, 32},
{"WalBatchHeaderSize", WalBatchHeaderSize, 18},
{"PhysicalRecordHeaderSize", PhysicalRecordHeaderSize, 7},
{"MaxWalBatchEntriesSize", MaxWalBatchEntriesSize, uint32(4 * 1024 * 1024)},
{"MaxWalBatchEntryCount", MaxWalBatchEntryCount, uint32(10000)},
{"MaxWalKeyBytes", MaxWalKeyBytes, uint32(4 * 1024)},
{"MaxWalInlineValueBytes", MaxWalInlineValueBytes, uint32(4 * 1024)},
{"MaxWalVarintBytes", MaxWalVarintBytes, 5},
{"DefaultMaxWalSegmentSize", DefaultMaxWalSegmentSize, uint64(64 * 1024 * 1024)},
{"WalMagic", WalMagic, uint32(0x57414C4B)},
{"WalFormatVersion", WalFormatVersion, uint16(1)},
}
for _, tt := range tests {
if tt.got != tt.expected {
t.Errorf("%s = %v, want %v", tt.name, tt.got, tt.expected)
}
}
}
func TestFragmentTypes(t *testing.T) {
if RecInvalid != uint8(0) {
t.Errorf("RecInvalid = %d, want 0", RecInvalid)
}
if RecFull != uint8(1) {
t.Errorf("RecFull = %d, want 1", RecFull)
}
if RecFirst != uint8(2) {
t.Errorf("RecFirst = %d, want 2", RecFirst)
}
if RecMiddle != uint8(3) {
t.Errorf("RecMiddle = %d, want 3", RecMiddle)
}
if RecLast != uint8(4) {
t.Errorf("RecLast = %d, want 4", RecLast)
}
}
func TestOpTypes(t *testing.T) {
if OpInvalid != uint8(0) {
t.Errorf("OpInvalid = %d, want 0", OpInvalid)
}
if OpPut != uint8(1) {
t.Errorf("OpPut = %d, want 1", OpPut)
}
if OpDelete != uint8(2) {
t.Errorf("OpDelete = %d, want 2", OpDelete)
}
}
func TestValueKinds(t *testing.T) {
if VKNone != uint8(0) {
t.Errorf("VKNone = %d, want 0", VKNone)
}
if VKInline != uint8(1) {
t.Errorf("VKInline = %d, want 1", VKInline)
}
if VKValueLogPointer != uint8(2) {
t.Errorf("VKValueLogPointer = %d, want 2", VKValueLogPointer)
}
}
+2
View File
@@ -0,0 +1,2 @@
// Package wal implements the Write-Ahead Log subsystem.
package wal
+134
View File
@@ -0,0 +1,134 @@
package wal
import (
"bytes"
"encoding/binary"
"errors"
"fmt"
)
// WalEntry represents a single WAL record.
type WalEntry struct {
OpType uint8
ValueKind uint8
Key []byte
Value []byte
}
// Validate checks that the entry fields are consistent with the design rules.
func (e *WalEntry) Validate() error {
keyLen := len(e.Key)
if keyLen == 0 || keyLen > int(MaxWalKeyBytes) {
return fmt.Errorf("wal: invalid key length %d", keyLen)
}
valLen := len(e.Value)
switch e.OpType {
case OpPut:
switch e.ValueKind {
case VKInline:
if valLen > int(MaxWalInlineValueBytes) {
return fmt.Errorf("wal: inline value length %d out of range [0, %d]", valLen, MaxWalInlineValueBytes)
}
case VKValueLogPointer:
if valLen == 0 {
return errors.New("wal: value log pointer requires non-empty value")
}
default:
return fmt.Errorf("wal: put requires valueKind Inline(1) or ValueLogPointer(2), got %d", e.ValueKind)
}
case OpDelete:
if e.ValueKind != VKNone {
return fmt.Errorf("wal: delete requires valueKind None(0), got %d", e.ValueKind)
}
if valLen != 0 {
return fmt.Errorf("wal: delete requires empty value, got length %d", valLen)
}
default:
return fmt.Errorf("wal: invalid opType %d", e.OpType)
}
return nil
}
// EncodeEntry serializes a WalEntry into a byte slice.
func EncodeEntry(e *WalEntry) ([]byte, error) {
if err := e.Validate(); err != nil {
return nil, err
}
keyLen := uint64(len(e.Key))
valLen := uint64(len(e.Value))
// Size: 1 (opType) + 1 (valueKind) + varint(keyLen) + varint(valLen) + key + value
size := 2 + MaxWalVarintBytes + MaxWalVarintBytes + len(e.Key) + len(e.Value)
buf := make([]byte, size)
buf[0] = e.OpType
buf[1] = e.ValueKind
n := 2
n += binary.PutUvarint(buf[n:], keyLen)
n += binary.PutUvarint(buf[n:], valLen)
n += copy(buf[n:], e.Key)
n += copy(buf[n:], e.Value)
return buf[:n], nil
}
// DecodeEntry deserializes a WalEntry from a byte slice.
// Returns the decoded entry and the number of bytes consumed.
func DecodeEntry(data []byte) (entry *WalEntry, consumed int, err error) {
if len(data) < 2 {
return nil, 0, errors.New("wal: data too short for entry header")
}
opType := data[0]
valueKind := data[1]
r := bytes.NewReader(data[2:])
keyLen, err := binary.ReadUvarint(r)
if err != nil {
return nil, 0, fmt.Errorf("wal: reading key length: %w", err)
}
valLen, err := binary.ReadUvarint(r)
if err != nil {
return nil, 0, fmt.Errorf("wal: reading value length: %w", err)
}
// Calculate consumed so far: 2 header bytes + bytes read from reader
consumed = 2 + (len(data) - 2 - r.Len())
// Read key
remaining := len(data) - consumed
if uint64(remaining) < keyLen {
return nil, 0, fmt.Errorf("wal: data truncated: need %d bytes for key, have %d", keyLen, remaining)
}
key := make([]byte, keyLen)
copy(key, data[consumed:consumed+int(keyLen)])
consumed += int(keyLen)
// Read value
remaining = len(data) - consumed
if uint64(remaining) < valLen {
return nil, 0, fmt.Errorf("wal: data truncated: need %d bytes for value, have %d", valLen, remaining)
}
value := make([]byte, valLen)
copy(value, data[consumed:consumed+int(valLen)])
consumed += int(valLen)
e := &WalEntry{
OpType: opType,
ValueKind: valueKind,
Key: key,
Value: value,
}
if err := e.Validate(); err != nil {
return nil, 0, err
}
return e, consumed, nil
}
+114
View File
@@ -0,0 +1,114 @@
package wal
import (
"bytes"
"encoding/binary"
"testing"
)
func TestEntryRoundtrip(t *testing.T) {
maxKey := bytes.Repeat([]byte("k"), int(MaxWalKeyBytes))
cases := []struct {
name string
e *WalEntry
}{
{"put_inline", &WalEntry{OpPut, VKInline, []byte("key1"), []byte("val1")}},
{"put_inline_empty_value", &WalEntry{OpPut, VKInline, []byte("key2"), []byte{}}},
{"delete", &WalEntry{OpDelete, VKNone, []byte("key3"), nil}},
{"put_max_key", &WalEntry{OpPut, VKInline, maxKey, []byte("v")}},
}
for _, tc := range cases {
t.Run(tc.name, func(t *testing.T) {
encoded, err := EncodeEntry(tc.e)
if err != nil {
t.Fatalf("encode: %v", err)
}
got, consumed, err := DecodeEntry(encoded)
if err != nil {
t.Fatalf("decode: %v", err)
}
if consumed != len(encoded) {
t.Fatalf("consumed %d != encoded len %d", consumed, len(encoded))
}
if got.OpType != tc.e.OpType {
t.Errorf("OpType: got %d, want %d", got.OpType, tc.e.OpType)
}
if got.ValueKind != tc.e.ValueKind {
t.Errorf("ValueKind: got %d, want %d", got.ValueKind, tc.e.ValueKind)
}
if !bytes.Equal(got.Key, tc.e.Key) {
t.Errorf("Key: got %q, want %q", got.Key, tc.e.Key)
}
if !bytes.Equal(got.Value, tc.e.Value) {
t.Errorf("Value: got %q, want %q", got.Value, tc.e.Value)
}
})
}
}
func TestEntryValidation(t *testing.T) {
bigKey := bytes.Repeat([]byte("k"), int(MaxWalKeyBytes)+1)
bigVal := bytes.Repeat([]byte("v"), int(MaxWalInlineValueBytes)+1)
cases := []struct {
name string
e *WalEntry
wantErr bool
}{
{"op_invalid", &WalEntry{OpInvalid, VKNone, []byte("k"), nil}, true},
{"put_vk_none", &WalEntry{OpPut, VKNone, []byte("k"), nil}, true},
{"delete_vk_inline", &WalEntry{OpDelete, VKInline, []byte("k"), nil}, true},
{"key_empty", &WalEntry{OpPut, VKInline, []byte{}, []byte("v")}, true},
{"key_too_big", &WalEntry{OpPut, VKInline, bigKey, []byte("v")}, true},
{"put_inline_val_too_big", &WalEntry{OpPut, VKInline, []byte("k"), bigVal}, true},
{"put_vlptr_val_empty", &WalEntry{OpPut, VKValueLogPointer, []byte("k"), []byte{}}, true},
}
for _, tc := range cases {
t.Run(tc.name, func(t *testing.T) {
_, err := EncodeEntry(tc.e)
if (err != nil) != tc.wantErr {
t.Errorf("EncodeEntry() error = %v, wantErr %v", err, tc.wantErr)
}
})
}
}
func TestEntryDecodeTruncated(t *testing.T) {
// Build a valid encoded entry, then truncate mid-varint.
e := &WalEntry{OpPut, VKInline, []byte("key1"), []byte("value1")}
full, err := EncodeEntry(e)
if err != nil {
t.Fatal(err)
}
// Truncate to just 1 byte — not enough for header.
_, _, err = DecodeEntry(full[:1])
if err == nil {
t.Error("expected error for 1-byte data")
}
// Build data with an incomplete varint: opType + valueKind + start of varint (0xFF means more bytes follow).
trunc := []byte{OpPut, VKInline, 0xFF}
_, _, err = DecodeEntry(trunc)
if err == nil {
t.Error("expected error for truncated varint")
}
// Also test: varint specifies more bytes than available.
// Encode a large key length varint but don't provide the key bytes.
varintBuf := make([]byte, binary.MaxVarintLen64)
n := binary.PutUvarint(varintBuf, 1000) // keyLen = 1000
data := []byte{OpPut, VKInline}
data = append(data, varintBuf[:n]...)
data = append(data, varintBuf[:n]...) // valLen varint (also 1000)
// Don't append any key/value bytes.
_, _, err = DecodeEntry(data)
if err == nil {
t.Error("expected error for missing key/value bytes")
}
}
+98
View File
@@ -0,0 +1,98 @@
package wal
import (
"encoding/binary"
"errors"
"hash/crc32"
)
const (
walFileHeaderSize = 32
walMagic = 0x57414C4B // "WALK" in ASCII
walFormatVersion = 1
)
// WalFileHeader is the 32-byte header written at the start of every WAL segment file.
type WalFileHeader struct {
Magic uint32
FormatVersion uint16
HeaderSize uint16
BlockSize uint32
SegmentID uint64
StartSequence uint64
HeaderCRC uint32
}
// EncodeWalHeader serializes h into a fixed-size 32-byte array using little-endian byte order.
// The HeaderCRC field is computed over bytes 027 (everything except the CRC itself).
func EncodeWalHeader(h *WalFileHeader) [walFileHeaderSize]byte {
h.HeaderSize = walFileHeaderSize
h.Magic = walMagic
h.FormatVersion = walFormatVersion
var buf [walFileHeaderSize]byte
le := binary.LittleEndian
le.PutUint32(buf[0:4], h.Magic)
le.PutUint16(buf[4:6], h.FormatVersion)
le.PutUint16(buf[6:8], h.HeaderSize)
le.PutUint32(buf[8:12], h.BlockSize)
le.PutUint64(buf[12:20], h.SegmentID)
le.PutUint64(buf[20:28], h.StartSequence)
// CRC32 IEEE over bytes 027 (excludes the CRC field itself)
h.HeaderCRC = crc32.ChecksumIEEE(buf[0:28])
le.PutUint32(buf[28:32], h.HeaderCRC)
return buf
}
var (
errBadMagic = errors.New("wal: bad magic number")
errBadVersion = errors.New("wal: unsupported format version")
errBadHeaderSize = errors.New("wal: bad header size")
errCRCMismatch = errors.New("wal: header CRC mismatch")
errHeaderTooShort = errors.New("wal: header data too short")
)
// DecodeWalHeader parses a 32-byte little-endian header and validates magic,
// format version, header size, and CRC.
func DecodeWalHeader(data []byte) (*WalFileHeader, error) {
if len(data) < walFileHeaderSize {
return nil, errHeaderTooShort
}
le := binary.LittleEndian
magic := le.Uint32(data[0:4])
if magic != walMagic {
return nil, errBadMagic
}
version := le.Uint16(data[4:6])
if version != walFormatVersion {
return nil, errBadVersion
}
hdrSize := le.Uint16(data[6:8])
if hdrSize != walFileHeaderSize {
return nil, errBadHeaderSize
}
// Verify CRC before trusting any other fields
gotCRC := crc32.ChecksumIEEE(data[0:28])
storedCRC := le.Uint32(data[28:32])
if gotCRC != storedCRC {
return nil, errCRCMismatch
}
return &WalFileHeader{
Magic: magic,
FormatVersion: version,
HeaderSize: hdrSize,
BlockSize: le.Uint32(data[8:12]),
SegmentID: le.Uint64(data[12:20]),
StartSequence: le.Uint64(data[20:28]),
HeaderCRC: storedCRC,
}, nil
}
+87
View File
@@ -0,0 +1,87 @@
package wal
import (
"errors"
"testing"
)
func TestHeaderRoundtrip(t *testing.T) {
orig := &WalFileHeader{
BlockSize: 32 * 1024, // 32 KB
SegmentID: 5,
StartSequence: 1000,
}
encoded := EncodeWalHeader(orig)
decoded, err := DecodeWalHeader(encoded[:])
if err != nil {
t.Fatalf("DecodeWalHeader returned error: %v", err)
}
if decoded.Magic != walMagic {
t.Errorf("Magic = %x, want %x", decoded.Magic, walMagic)
}
if decoded.FormatVersion != walFormatVersion {
t.Errorf("FormatVersion = %d, want %d", decoded.FormatVersion, walFormatVersion)
}
if decoded.HeaderSize != walFileHeaderSize {
t.Errorf("HeaderSize = %d, want %d", decoded.HeaderSize, walFileHeaderSize)
}
if decoded.BlockSize != orig.BlockSize {
t.Errorf("BlockSize = %d, want %d", decoded.BlockSize, orig.BlockSize)
}
if decoded.SegmentID != orig.SegmentID {
t.Errorf("SegmentID = %d, want %d", decoded.SegmentID, orig.SegmentID)
}
if decoded.StartSequence != orig.StartSequence {
t.Errorf("StartSequence = %d, want %d", decoded.StartSequence, orig.StartSequence)
}
if decoded.HeaderCRC != orig.HeaderCRC {
t.Errorf("HeaderCRC = %x, want %x", decoded.HeaderCRC, orig.HeaderCRC)
}
}
func TestHeaderCRC(t *testing.T) {
h := &WalFileHeader{
BlockSize: 32 * 1024,
SegmentID: 1,
StartSequence: 0,
}
encoded := EncodeWalHeader(h)
// Flip a byte in the magic field (bytes 0-3)
encoded[0] ^= 0xFF
_, err := DecodeWalHeader(encoded[:])
if !errors.Is(err, errCRCMismatch) && !errors.Is(err, errBadMagic) {
// Flipping magic may fail on magic check first or CRC check
// Either way, decoding must fail
t.Fatalf("expected CRC or magic error, got: %v", err)
}
// Restore magic and flip a byte in the payload instead
encoded[0] = byte(walMagic & 0xFF)
encoded[12] ^= 0x01 // flip byte in SegmentID
_, err = DecodeWalHeader(encoded[:])
if !errors.Is(err, errCRCMismatch) {
t.Fatalf("expected errCRCMismatch, got: %v", err)
}
}
func TestHeaderBadMagic(t *testing.T) {
data := make([]byte, 32)
// All zeros — magic won't match
_, err := DecodeWalHeader(data)
if !errors.Is(err, errBadMagic) {
t.Fatalf("expected errBadMagic, got: %v", err)
}
}
func TestHeaderShortData(t *testing.T) {
data := make([]byte, 16) // too short
_, err := DecodeWalHeader(data)
if !errors.Is(err, errHeaderTooShort) {
t.Fatalf("expected errHeaderTooShort, got: %v", err)
}
}
+126
View File
@@ -0,0 +1,126 @@
package wal
import (
"encoding/binary"
"errors"
"hash/crc32"
)
// PhysicalRecord represents a single physical record in the WAL.
type PhysicalRecord struct {
CRC uint32
Length uint16
Type uint8
Payload []byte
}
// EncodePhysicalRecord encodes a physical record with the given type and payload.
// Format: [crc32 u32 LE][length u16 LE][type u8][payload bytes]
// CRC covers length + type + payload.
func EncodePhysicalRecord(recType uint8, payload []byte) []byte {
length := uint16(len(payload))
buf := make([]byte, PhysicalRecordHeaderSize+len(payload))
// Write length and type first so we can compute CRC.
binary.LittleEndian.PutUint16(buf[4:6], length)
buf[6] = recType
copy(buf[7:], payload)
// CRC covers bytes [4:] = length + type + payload.
crc := crc32.ChecksumIEEE(buf[4:])
binary.LittleEndian.PutUint32(buf[0:4], crc)
return buf
}
// DecodePhysicalRecord decodes a physical record from data.
// Returns the record, number of bytes consumed, and any error.
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]
if int(length) > len(data)-PhysicalRecordHeaderSize {
return nil, 0, errors.New("record: data too short for payload")
}
payload := make([]byte, length)
copy(payload, data[7:7+length])
// Verify CRC: covers length + type + payload.
expectedCRC := crc32.ChecksumIEEE(data[4 : 7+length])
if crc != expectedCRC {
return nil, 0, errors.New("record: CRC mismatch")
}
consumed = PhysicalRecordHeaderSize + int(length)
return &PhysicalRecord{
CRC: crc,
Length: length,
Type: recType,
Payload: payload,
}, consumed, nil
}
// PaddingNeeded returns the number of padding bytes needed at blockOffset.
// If the remaining space in the current block is <= PhysicalRecordHeaderSize (7),
// that remaining space must be zero-padded.
func PaddingNeeded(blockOffset uint32) int {
remaining := WalBlockSize - (blockOffset % WalBlockSize)
if remaining <= PhysicalRecordHeaderSize {
return int(remaining)
}
return 0
}
// CanFitRecord reports whether a physical record with the given payload length
// can fit in the current block starting at blockOffset.
func CanFitRecord(blockOffset uint32, payloadLen uint32) bool {
remaining := WalBlockSize - (blockOffset % WalBlockSize)
return int(remaining) >= PhysicalRecordHeaderSize+int(payloadLen)
}
// SplitIntoRecords splits an encoded WAL batch into physical record payloads
// respecting 32 KB block boundaries.
// Each returned byte slice is the full encoded physical record (header + payload).
func SplitIntoRecords(encodedBatch []byte) [][]byte {
maxPayload := WalBlockSize - PhysicalRecordHeaderSize
total := len(encodedBatch)
if total == 0 {
return nil
}
// Single record fits entirely.
if total <= maxPayload {
return [][]byte{EncodePhysicalRecord(RecFull, encodedBatch)}
}
var records [][]byte
offset := 0
for offset < total {
chunkLen := min(total-offset, maxPayload)
var recType uint8
switch {
case offset == 0 && offset+chunkLen == total:
recType = RecFull
case offset == 0:
recType = RecFirst
case offset+chunkLen == total:
recType = RecLast
default:
recType = RecMiddle
}
records = append(records, EncodePhysicalRecord(recType, encodedBatch[offset:offset+chunkLen]))
offset += chunkLen
}
return records
}
+143
View File
@@ -0,0 +1,143 @@
package wal
import (
"bytes"
"testing"
)
func TestRecordRoundtrip(t *testing.T) {
payload := []byte("hello world")
encoded := EncodePhysicalRecord(RecFull, payload)
rec, consumed, err := DecodePhysicalRecord(encoded)
if err != nil {
t.Fatalf("DecodePhysicalRecord failed: %v", err)
}
if rec.Type != RecFull {
t.Errorf("expected type RecFull(%d), got %d", RecFull, rec.Type)
}
if string(rec.Payload) != "hello world" {
t.Errorf("expected payload 'hello world', got %q", string(rec.Payload))
}
if consumed != 7+len(payload) {
t.Errorf("expected consumed %d, got %d", 7+len(payload), consumed)
}
}
func TestSplitSmallPayload(t *testing.T) {
payload := make([]byte, 100)
for i := range payload {
payload[i] = byte(i)
}
records := SplitIntoRecords(payload)
if len(records) != 1 {
t.Fatalf("expected 1 record, got %d", len(records))
}
rec, _, err := DecodePhysicalRecord(records[0])
if err != nil {
t.Fatalf("DecodePhysicalRecord failed: %v", err)
}
if rec.Type != RecFull {
t.Errorf("expected RecFull, got %d", rec.Type)
}
if !bytes.Equal(rec.Payload, payload) {
t.Error("payload mismatch")
}
}
func TestSplitIntoRecords(t *testing.T) {
// 40 KB payload → needs to split across blocks.
payload := make([]byte, 40*1024)
for i := range payload {
payload[i] = byte(i % 256)
}
records := SplitIntoRecords(payload)
if len(records) < 2 {
t.Fatalf("expected at least 2 records, got %d", len(records))
}
// Verify fragment sequence.
types := make([]uint8, len(records))
var concatenated []byte
for i, enc := range records {
rec, _, err := DecodePhysicalRecord(enc)
if err != nil {
t.Fatalf("DecodePhysicalRecord record %d failed: %v", i, err)
}
types[i] = rec.Type
concatenated = append(concatenated, rec.Payload...)
}
// First record must be RecFirst.
if types[0] != RecFirst {
t.Errorf("first record type: expected RecFirst(%d), got %d", RecFirst, types[0])
}
// Last record must be RecLast.
if types[len(types)-1] != RecLast {
t.Errorf("last record type: expected RecLast(%d), got %d", RecLast, types[len(types)-1])
}
// Middle records must be RecMiddle.
for i := 1; i < len(types)-1; i++ {
if types[i] != RecMiddle {
t.Errorf("record %d type: expected RecMiddle(%d), got %d", i, RecMiddle, types[i])
}
}
// Concatenated payloads must equal original.
if !bytes.Equal(concatenated, payload) {
t.Error("concatenated payloads do not match original")
}
}
func TestBlockPadding(t *testing.T) {
// blockOffset = WalBlockSize - 5 → remaining = 5, which is <= 7 → padding needed = 5.
blockOffset := uint32(WalBlockSize - 5)
padding := PaddingNeeded(blockOffset)
if padding != 5 {
t.Errorf("PaddingNeeded(%d): expected 5, got %d", blockOffset, padding)
}
// Cannot fit a record.
if CanFitRecord(blockOffset, 1) {
t.Error("CanFitRecord should return false when remaining <= 7")
}
// blockOffset = WalBlockSize - 8 → remaining = 8, which is > 7 → no padding needed.
blockOffset2 := uint32(WalBlockSize - 8)
padding2 := PaddingNeeded(blockOffset2)
if padding2 != 0 {
t.Errorf("PaddingNeeded(%d): expected 0, got %d", blockOffset2, padding2)
}
// Can fit a 1-byte payload: remaining=8, header=7, payload=1 → 8 >= 8.
if !CanFitRecord(blockOffset2, 1) {
t.Error("CanFitRecord should return true when remaining=8 and payloadLen=1")
}
// Cannot fit a 2-byte payload: remaining=8, header=7, payload=2 → 8 < 9.
if CanFitRecord(blockOffset2, 2) {
t.Error("CanFitRecord should return false when remaining=8 and payloadLen=2")
}
}
func TestCRCMismatch(t *testing.T) {
encoded := EncodePhysicalRecord(RecFull, []byte("test"))
// Corrupt a payload byte.
encoded[8] ^= 0xFF
_, _, err := DecodePhysicalRecord(encoded)
if err == nil {
t.Error("expected CRC mismatch error")
}
}
func TestDataTooShort(t *testing.T) {
_, _, err := DecodePhysicalRecord([]byte{1, 2, 3})
if err == nil {
t.Error("expected error for data too short")
}
}
+87
View File
@@ -0,0 +1,87 @@
package wal
import (
"sync/atomic"
"github.com/dailz/go-kv"
)
// SequenceManager manages monotonic sequence number allocation for the WAL.
// Invariant: durableSequence <= publishedSequence <= nextSequence
type SequenceManager struct {
nextSequence atomic.Uint64
publishedSequence atomic.Uint64
durableSequence atomic.Uint64
exhausted atomic.Bool
}
// NewSequenceManager creates a SequenceManager initialised from a recovered
// sequence number. All three watermarks start at recoveredSequence.
func NewSequenceManager(recoveredSequence uint64) *SequenceManager {
sm := &SequenceManager{}
sm.nextSequence.Store(recoveredSequence)
sm.publishedSequence.Store(recoveredSequence)
sm.durableSequence.Store(recoveredSequence)
return sm
}
// AllocateBatch atomically reserves [base, base+count-1] sequence numbers.
// Returns ErrSequenceExhausted if count is 0 or the allocation would overflow uint64.
func (sm *SequenceManager) AllocateBatch(count uint32) (baseSequence uint64, err error) {
if count == 0 {
return 0, go_kv.ErrSequenceExhausted
}
for {
if sm.exhausted.Load() {
return 0, go_kv.ErrSequenceExhausted
}
base := sm.nextSequence.Load()
last := base + uint64(count) - 1
if last < base {
return 0, go_kv.ErrSequenceExhausted
}
newNext := last + 1
if !sm.nextSequence.CompareAndSwap(base, newNext) {
continue
}
if newNext == 0 {
sm.exhausted.Store(true)
}
return base, nil
}
}
// Publish advances the publishedSequence watermark to seq (only forward).
func (sm *SequenceManager) Publish(seq uint64) {
for {
current := sm.publishedSequence.Load()
if seq <= current {
return
}
if sm.publishedSequence.CompareAndSwap(current, seq) {
return
}
}
}
// MarkDurable advances the durableSequence watermark to seq (only forward).
func (sm *SequenceManager) MarkDurable(seq uint64) {
for {
current := sm.durableSequence.Load()
if seq <= current {
return
}
if sm.durableSequence.CompareAndSwap(current, seq) {
return
}
}
}
// Published returns the current publishedSequence watermark.
func (sm *SequenceManager) Published() uint64 { return sm.publishedSequence.Load() }
// Durable returns the current durableSequence watermark.
func (sm *SequenceManager) Durable() uint64 { return sm.durableSequence.Load() }
// NextSequence returns the next sequence number to be allocated.
func (sm *SequenceManager) NextSequence() uint64 { return sm.nextSequence.Load() }
+154
View File
@@ -0,0 +1,154 @@
package wal
import (
"errors"
"math"
"sync"
"sync/atomic"
"testing"
"github.com/dailz/go-kv"
)
func TestSequenceAllocation(t *testing.T) {
sm := NewSequenceManager(0)
base, err := sm.AllocateBatch(5)
if err != nil {
t.Fatalf("AllocateBatch(5): %v", err)
}
if base != 0 {
t.Fatalf("expected base=0, got %d", base)
}
base, err = sm.AllocateBatch(3)
if err != nil {
t.Fatalf("AllocateBatch(3): %v", err)
}
if base != 5 {
t.Fatalf("expected base=5, got %d", base)
}
base, err = sm.AllocateBatch(1)
if err != nil {
t.Fatalf("AllocateBatch(1): %v", err)
}
if base != 8 {
t.Fatalf("expected base=8, got %d", base)
}
if sm.NextSequence() != 9 {
t.Fatalf("expected NextSequence=9, got %d", sm.NextSequence())
}
}
func TestSequenceOverflow(t *testing.T) {
nearMax := uint64(math.MaxUint64 - 2)
sm := NewSequenceManager(nearMax)
// Remaining: MaxUint64-2, MaxUint64-1, MaxUint64 = 3 slots.
// Asking for 5 should overflow.
_, err := sm.AllocateBatch(5)
if !errors.Is(err, go_kv.ErrSequenceExhausted) {
t.Fatalf("expected ErrSequenceExhausted, got %v", err)
}
// 3 should still succeed.
base, err := sm.AllocateBatch(3)
if err != nil {
t.Fatalf("AllocateBatch(3): %v", err)
}
if base != nearMax {
t.Fatalf("expected base=%d, got %d", nearMax, base)
}
// Now any further allocation should fail.
_, err = sm.AllocateBatch(1)
if !errors.Is(err, go_kv.ErrSequenceExhausted) {
t.Fatalf("expected ErrSequenceExhausted after exhaustion, got %v", err)
}
}
func TestPublishAdvance(t *testing.T) {
sm := NewSequenceManager(0)
sm.Publish(10)
if sm.Published() != 10 {
t.Fatalf("expected Published=10, got %d", sm.Published())
}
// Publishing a lower value must not decrease the watermark.
sm.Publish(5)
if sm.Published() != 10 {
t.Fatalf("expected Published=10 (no decrease), got %d", sm.Published())
}
sm.Publish(15)
if sm.Published() != 15 {
t.Fatalf("expected Published=15, got %d", sm.Published())
}
}
func TestMarkDurable(t *testing.T) {
sm := NewSequenceManager(0)
sm.MarkDurable(8)
if sm.Durable() != 8 {
t.Fatalf("expected Durable=8, got %d", sm.Durable())
}
// Lower value must not decrease.
sm.MarkDurable(3)
if sm.Durable() != 8 {
t.Fatalf("expected Durable=8 (no decrease), got %d", sm.Durable())
}
}
func TestZeroCountRejected(t *testing.T) {
sm := NewSequenceManager(0)
_, err := sm.AllocateBatch(0)
if !errors.Is(err, go_kv.ErrSequenceExhausted) {
t.Fatalf("expected ErrSequenceExhausted for count=0, got %v", err)
}
}
func TestConcurrentAllocation(t *testing.T) {
const goroutines = 16
const batchSize uint32 = 100
sm := NewSequenceManager(0)
var totalAllocated atomic.Uint64
var wg sync.WaitGroup
wg.Add(goroutines)
for i := 0; i < goroutines; i++ {
go func() {
defer wg.Done()
for j := 0; j < 50; j++ {
base, err := sm.AllocateBatch(batchSize)
if err != nil {
t.Errorf("AllocateBatch failed: %v", err)
return
}
totalAllocated.Add(uint64(batchSize))
// Verify no overlap: base must be aligned to batchSize increments
// and within valid range. The key property is no gaps.
_ = base
}
}()
}
wg.Wait()
expected := uint64(goroutines) * 50 * uint64(batchSize)
if totalAllocated.Load() != expected {
t.Fatalf("expected total allocated=%d, got %d", expected, totalAllocated.Load())
}
if sm.NextSequence() != expected {
t.Fatalf("expected NextSequence=%d, got %d", expected, sm.NextSequence())
}
}