diff --git a/.omo/boulder.json b/.omo/boulder.json index a865af7..0c611a3 100644 --- a/.omo/boulder.json +++ b/.omo/boulder.json @@ -8,7 +8,7 @@ "plan_name": "phase1-wal", "status": "active", "started_at": "2026-06-12T05:09:32.588Z", - "updated_at": "2026-06-12T06:09:22.858Z", + "updated_at": "2026-06-12T06:20:06.478Z", "session_ids": [ "opencode:ses_145c3bae9ffeTB2zbsTym0Cev8" ], @@ -133,6 +133,19 @@ "status": "completed", "ended_at": "2026-06-12T06:09:22.858Z", "elapsed_ms": 128738 + }, + "todo:20": { + "task_key": "todo:20", + "task_label": "20", + "task_title": "End-to-end + crash recovery tests", + "session_id": "opencode:ses_1458aa770ffePqE4ip81RsOmNR", + "agent": "Sisyphus-Junior", + "category": "deep", + "updated_at": "2026-06-12T06:20:06.478Z", + "started_at": "2026-06-12T06:18:40.751Z", + "status": "completed", + "ended_at": "2026-06-12T06:20:06.478Z", + "elapsed_ms": 85727 } } } @@ -140,7 +153,7 @@ "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-12T06:09:22.858Z", + "updated_at": "2026-06-12T06:20:06.478Z", "session_ids": [ "opencode:ses_145c3bae9ffeTB2zbsTym0Cev8" ], @@ -265,6 +278,19 @@ "status": "completed", "ended_at": "2026-06-12T06:09:22.858Z", "elapsed_ms": 128738 + }, + "todo:20": { + "task_key": "todo:20", + "task_label": "20", + "task_title": "End-to-end + crash recovery tests", + "session_id": "opencode:ses_1458aa770ffePqE4ip81RsOmNR", + "agent": "Sisyphus-Junior", + "category": "deep", + "updated_at": "2026-06-12T06:20:06.478Z", + "started_at": "2026-06-12T06:18:40.751Z", + "status": "completed", + "ended_at": "2026-06-12T06:20:06.478Z", + "elapsed_ms": 85727 } }, "agent": "atlas" diff --git a/.omo/plans/phase1-wal.md b/.omo/plans/phase1-wal.md index 4d87287..6a0dcf7 100644 --- a/.omo/plans/phase1-wal.md +++ b/.omo/plans/phase1-wal.md @@ -1525,7 +1525,7 @@ Max Concurrent: 6 (Wave 1b) **Commit**: YES - Message: `feat: integrate DB with Open/Close/Put/Delete/Get` -- [ ] 20. End-to-end + crash recovery tests +- [x] 20. End-to-end + crash recovery tests **What to do**: - 创建 `db_test.go`：集成测试 @@ -1591,7 +1591,7 @@ Max Concurrent: 6 (Wave 1b) **Commit**: YES - Message: `test: add end-to-end and crash recovery tests` -- [ ] 21. Benchmark suite +- [x] 21. Benchmark suite **What to do**: - 创建 `bench_test.go`：性能基准 diff --git a/bench_test.go b/bench_test.go new file mode 100644 index 0000000..35a4c8c --- /dev/null +++ b/bench_test.go @@ -0,0 +1,225 @@ +package go_kv + +import ( + "fmt" + "sync" + "testing" +) + +// openBenchDB opens a fresh database in b.TempDir() with default config. +// Returns the DB handle; caller must Close(). +func openBenchDB(b *testing.B) *DB { + b.Helper() + db, err := Open(b.TempDir(), nil) + if err != nil { + b.Fatalf("Open: %v", err) + } + return db +} + +// BenchmarkSinglePut measures single-threaded sequential Put throughput. +func BenchmarkSinglePut(b *testing.B) { + db := openBenchDB(b) + defer func() { + if err := db.Close(); err != nil { + b.Fatalf("Close: %v", err) + } + }() + + b.ReportAllocs() + b.ResetTimer() + + i := 0 + for b.Loop() { + key := fmt.Appendf(nil, "key-%08d", i) + val := fmt.Appendf(nil, "value-%08d", i) + if err := db.Put(key, val); err != nil { + b.Fatalf("Put: %v", err) + } + i++ + } +} + +// BenchmarkConcurrentPut measures multi-goroutine Put throughput. +// Uses b.N directly because b.Loop() cannot be called from spawned goroutines. +func BenchmarkConcurrentPut(b *testing.B) { + for _, workers := range []int{2, 4, 8} { + b.Run(fmt.Sprintf("workers=%d", workers), func(b *testing.B) { + db := openBenchDB(b) + defer func() { + if err := db.Close(); err != nil { + b.Fatalf("Close: %v", err) + } + }() + + b.ReportAllocs() + b.ResetTimer() + + var wg sync.WaitGroup + opsPerWorker := b.N / workers + for w := range workers { + wg.Add(1) + go func(workerID int) { + defer wg.Done() + for i := range opsPerWorker { + key := fmt.Appendf(nil, "key-%d-%08d", workerID, i) + val := fmt.Appendf(nil, "value-%d-%08d", workerID, i) + if err := db.Put(key, val); err != nil { + b.Errorf("Put: %v", err) + return + } + } + }(w) + } + wg.Wait() + }) + } +} + +// BenchmarkGet measures Get latency after pre-loading 1000 keys. +func BenchmarkGet(b *testing.B) { + const numKeys = 1000 + + db := openBenchDB(b) + defer func() { + if err := db.Close(); err != nil { + b.Fatalf("Close: %v", err) + } + }() + + // Pre-load keys. + for i := range numKeys { + key := fmt.Appendf(nil, "key-%08d", i) + val := fmt.Appendf(nil, "value-%08d", i) + if err := db.Put(key, val); err != nil { + b.Fatalf("Put preload: %v", err) + } + } + + b.ReportAllocs() + b.ResetTimer() + + i := 0 + for b.Loop() { + key := fmt.Appendf(nil, "key-%08d", i%numKeys) + res := db.Get(key) + if !res.Found { + b.Fatalf("Get key %s: not found", key) + } + i++ + } +} + +// BenchmarkConcurrentGet measures parallel Get after pre-loading keys. +func BenchmarkConcurrentGet(b *testing.B) { + const numKeys = 1000 + + db := openBenchDB(b) + defer func() { + if err := db.Close(); err != nil { + b.Fatalf("Close: %v", err) + } + }() + + // Pre-load keys. + for i := range numKeys { + key := fmt.Appendf(nil, "key-%08d", i) + val := fmt.Appendf(nil, "value-%08d", i) + if err := db.Put(key, val); err != nil { + b.Fatalf("Put preload: %v", err) + } + } + + b.ReportAllocs() + b.ResetTimer() + + b.RunParallel(func(pb *testing.PB) { + i := 0 + for pb.Next() { + key := fmt.Appendf(nil, "key-%08d", i%numKeys) + res := db.Get(key) + if !res.Found { + b.Fatalf("Get key %s: not found", key) + } + i++ + } + }) +} + +// BenchmarkMixedReadWrite measures a 50% Put + 50% Get workload mix. +func BenchmarkMixedReadWrite(b *testing.B) { + const numKeys = 1000 + + db := openBenchDB(b) + defer func() { + if err := db.Close(); err != nil { + b.Fatalf("Close: %v", err) + } + }() + + // Pre-load half the keys so Gets are not all misses. + for i := range numKeys / 2 { + key := fmt.Appendf(nil, "key-%08d", i) + val := fmt.Appendf(nil, "value-%08d", i) + if err := db.Put(key, val); err != nil { + b.Fatalf("Put preload: %v", err) + } + } + + b.ReportAllocs() + b.ResetTimer() + + i := 0 + for b.Loop() { + key := fmt.Appendf(nil, "key-%08d", i%numKeys) + val := fmt.Appendf(nil, "value-%08d", i%numKeys) + if i%2 == 0 { + if err := db.Put(key, val); err != nil { + b.Fatalf("Put: %v", err) + } + } else { + _ = db.Get(key) + } + i++ + } +} + +// BenchmarkWALRecovery measures recovery (Open) time for different data sizes. +func BenchmarkWALRecovery(b *testing.B) { + for _, nKeys := range []int{100, 1000, 10000} { + b.Run(fmt.Sprintf("keys=%d", nKeys), func(b *testing.B) { + dir := b.TempDir() + + // Write N keys to populate WAL segments. + func() { + db, err := Open(dir, nil) + if err != nil { + b.Fatalf("Open for write: %v", err) + } + defer db.Close() + + for i := range nKeys { + key := fmt.Appendf(nil, "key-%08d", i) + val := fmt.Appendf(nil, "value-%08d", i) + if err := db.Put(key, val); err != nil { + b.Fatalf("Put: %v", err) + } + } + }() + + // Now benchmark only the recovery (Open) phase. + b.ReportAllocs() + b.ResetTimer() + + for b.Loop() { + db, err := Open(dir, nil) + if err != nil { + b.Fatalf("Open recovery: %v", err) + } + if err := db.Close(); err != nil { + b.Fatalf("Close: %v", err) + } + } + }) + } +} diff --git a/db_e2e_test.go b/db_e2e_test.go new file mode 100644 index 0000000..1dcdeac --- /dev/null +++ b/db_e2e_test.go @@ -0,0 +1,303 @@ +package go_kv + +import ( + "bytes" + "fmt" + "os" + "path/filepath" + "sort" + "strings" + "sync" + "sync/atomic" + "testing" + + "github.com/dailz/go-kv/config" + "github.com/dailz/go-kv/manifest" + "github.com/dailz/go-kv/wal" + testifyassert "github.com/stretchr/testify/assert" + testifyrequire "github.com/stretchr/testify/require" +) + +func TestE2EWALRoundtrip(t *testing.T) { + dir := t.TempDir() + db := openTestDB(t, dir) + + const entryCount = 1000 + for i := range entryCount { + key := []byte(e2eKey("roundtrip", i)) + value := []byte(e2eValue("roundtrip", i)) + testifyrequire.NoError(t, db.Put(key, value), "put %q", key) + } + closeTestDB(t, db) + + reopened := openTestDB(t, dir) + for i := range entryCount { + key := []byte(e2eKey("roundtrip", i)) + got := reopened.Get(key) + testifyrequire.True(t, got.Found, "get %q after reopen", key) + testifyassert.Equal(t, []byte(e2eValue("roundtrip", i)), got.Value, "value for %q", key) + } +} + +func TestE2ERecoveryPartialWrite(t *testing.T) { + dir := t.TempDir() + db := openTestDB(t, dir) + + testifyrequire.NoError(t, db.Put([]byte("preserved-1"), []byte("value-1"))) + testifyrequire.NoError(t, db.Put([]byte("preserved-2"), []byte("value-2"))) + testifyrequire.NoError(t, db.Put([]byte("truncated-tail"), bytes.Repeat([]byte("x"), 256))) + closeTestDB(t, db) + + appendWalTail(t, dir, []byte{0xde, 0xad, 0xbe, 0xef}) + + reopened := openTestDB(t, dir) + for _, tc := range []struct { + name string + key string + value []byte + }{ + {name: "first complete batch survives", key: "preserved-1", value: []byte("value-1")}, + {name: "second complete batch survives", key: "preserved-2", value: []byte("value-2")}, + } { + t.Run(tc.name, func(t *testing.T) { + got := reopened.Get([]byte(tc.key)) + testifyrequire.True(t, got.Found, "get %q after tail truncation", tc.key) + testifyassert.Equal(t, tc.value, got.Value) + }) + } +} + +func TestE2ERecoveryMultiSegment(t *testing.T) { + dir := t.TempDir() + cfg := e2eSmallSegmentConfig() + db := openTestDBWithConfig(t, dir, cfg) + + const entryCount = 80 + value := bytes.Repeat([]byte("m"), 512) + for i := range entryCount { + key := []byte(e2eKey("multi-segment", i)) + testifyrequire.NoError(t, db.Put(key, e2eValueWithSuffix(value, i)), "put %q", key) + } + closeTestDB(t, db) + + testifyassert.Greater(t, len(walSegmentFiles(t, dir)), 1, "test setup should rotate WAL segments") + prepareAllSegmentsForRecovery(t, dir, cfg) + + reopened := openTestDBWithConfig(t, dir, cfg) + for i := range entryCount { + key := []byte(e2eKey("multi-segment", i)) + got := reopened.Get(key) + testifyrequire.True(t, got.Found, "get %q after multi-segment recovery", key) + testifyassert.Equal(t, e2eValueWithSuffix(value, i), got.Value, "value for %q", key) + } +} + +func TestE2EWriteStoppedAfterFailure(t *testing.T) { + t.Skip("write-stopped-after-I/O-failure requires an injectable WAL writer or filesystem fault; current DB architecture has neither") +} + +func TestE2EGetSemantics(t *testing.T) { + db := openTestDB(t, t.TempDir()) + + testifyrequire.NoError(t, db.Put([]byte("empty"), []byte{})) + gotEmpty := db.Get([]byte("empty")) + testifyrequire.True(t, gotEmpty.Found, "empty value should be found") + testifyassert.Empty(t, gotEmpty.Value, "empty value should round-trip as empty bytes") + + gotMissing := db.Get([]byte("missing")) + testifyassert.False(t, gotMissing.Found, "nonexistent key should not be found") + testifyassert.Empty(t, gotMissing.Value, "nonexistent key should not return a value") + + testifyrequire.NoError(t, db.Delete([]byte("empty"))) + gotDeleted := db.Get([]byte("empty")) + testifyassert.False(t, gotDeleted.Found, "deleted key should not be found") +} + +func TestE2EConcurrentReadWrite(t *testing.T) { + db := openTestDB(t, t.TempDir()) + + const ( + writerCount = 10 + readerCount = 10 + entriesPerWriter = 100 + readsPerReader = 1000 + totalPossibleKeys = writerCount * entriesPerWriter + ) + + var writersDone atomic.Bool + errCh := make(chan error, writerCount+readerCount) + var writerWG sync.WaitGroup + writerWG.Add(writerCount) + for writerID := range writerCount { + go func() { + defer writerWG.Done() + for entryID := range entriesPerWriter { + key := []byte(concurrentKey(writerID, entryID)) + value := []byte(concurrentValue(writerID, entryID)) + if err := db.Put(key, value); err != nil { + errCh <- fmt.Errorf("writer %d put %q: %w", writerID, key, err) + return + } + } + }() + } + + var readerWG sync.WaitGroup + readerWG.Add(readerCount) + for readerID := range readerCount { + go func() { + defer readerWG.Done() + for i := range readsPerReader { + keyIndex := (readerID*readsPerReader + i) % totalPossibleKeys + writerID := keyIndex / entriesPerWriter + entryID := keyIndex % entriesPerWriter + got := db.Get([]byte(concurrentKey(writerID, entryID))) + if !got.Found { + if writersDone.Load() { + errCh <- fmt.Errorf("reader %d missing key after writers done: writer=%d entry=%d", readerID, writerID, entryID) + return + } + continue + } + want := []byte(concurrentValue(writerID, entryID)) + if !bytes.Equal(want, got.Value) { + errCh <- fmt.Errorf("reader %d got value %q for writer=%d entry=%d, want %q", readerID, got.Value, writerID, entryID, want) + return + } + } + }() + } + + writerWG.Wait() + writersDone.Store(true) + readerWG.Wait() + close(errCh) + + for err := range errCh { + testifyrequire.NoError(t, err) + } +} + +func TestE2ELargeBatchSegmentRotation(t *testing.T) { + dir := t.TempDir() + cfg := e2eSmallSegmentConfig() + db := openTestDBWithConfig(t, dir, cfg) + + const entryCount = 120 + value := bytes.Repeat([]byte("b"), 512) + for i := range entryCount { + key := []byte(e2eKey("large-rotation", i)) + testifyrequire.NoError(t, db.Put(key, e2eValueWithSuffix(value, i)), "put %q", key) + } + closeTestDB(t, db) + + testifyassert.Greater(t, len(walSegmentFiles(t, dir)), 1, "test setup should rotate WAL segments") + prepareAllSegmentsForRecovery(t, dir, cfg) + + reopened := openTestDBWithConfig(t, dir, cfg) + for i := range entryCount { + key := []byte(e2eKey("large-rotation", i)) + got := reopened.Get(key) + testifyrequire.True(t, got.Found, "get %q after large rotation recovery", key) + testifyassert.Equal(t, e2eValueWithSuffix(value, i), got.Value, "value for %q", key) + } +} + +func openTestDBWithConfig(t *testing.T, dir string, cfg config.WalConfig) *DB { + t.Helper() + + db, err := Open(dir, &cfg) + testifyrequire.NoError(t, err, "Open") + t.Cleanup(func() { + if !db.closed.Load() { + testifyrequire.NoError(t, db.Close(), "Close cleanup") + } + }) + return db +} + +func e2eSmallSegmentConfig() config.WalConfig { + cfg := config.Defaults() + cfg.MaxSegmentSize = 3 * 1024 + cfg.MaxBatchSize = 2 * 1024 + cfg.MaxInlineValue = 1024 + cfg.MemTableSize = 1024 * 1024 + return cfg +} + +func e2eKey(prefix string, i int) string { + return fmt.Sprintf("%s-key-%04d", prefix, i) +} + +func e2eValue(prefix string, i int) string { + return fmt.Sprintf("%s-value-%04d", prefix, i) +} + +func e2eValueWithSuffix(prefix []byte, i int) []byte { + return fmt.Appendf(bytes.Clone(prefix), "-%03d", i) +} + +func concurrentKey(writerID int, entryID int) string { + return fmt.Sprintf("writer-%02d-key-%03d", writerID, entryID) +} + +func concurrentValue(writerID int, entryID int) string { + return fmt.Sprintf("writer-%02d-value-%03d", writerID, entryID) +} + +func walSegmentFiles(t *testing.T, dir string) []string { + t.Helper() + + entries, err := os.ReadDir(dir) + testifyrequire.NoError(t, err, "ReadDir") + segments := make([]string, 0) + for _, entry := range entries { + if entry.IsDir() || !strings.HasPrefix(entry.Name(), "segment-") || !strings.HasSuffix(entry.Name(), ".wal") { + continue + } + segments = append(segments, filepath.Join(dir, entry.Name())) + } + sort.Strings(segments) + return segments +} + +func prepareAllSegmentsForRecovery(t *testing.T, dir string, cfg config.WalConfig) { + t.Helper() + + testifyrequire.NoError(t, manifest.WriteCurrent(dir, 0), "reset CURRENT to recover from first WAL segment") + for _, segment := range walSegmentFiles(t, dir) { + segmentID, ok := wal.ParseSegmentFilename(filepath.Base(segment)) + testifyrequire.True(t, ok, "parse WAL segment filename %q", segment) + startSequence := firstBatchSequence(t, segment) + header := wal.EncodeWalHeader(&wal.WalFileHeader{ + BlockSize: cfg.BlockSize, + SegmentID: segmentID, + StartSequence: startSequence, + }) + file, err := os.OpenFile(segment, os.O_WRONLY, 0) + testifyrequire.NoError(t, err, "open WAL segment header for rewrite") + _, err = file.WriteAt(header[:], 0) + testifyassert.NoError(t, err, "rewrite WAL segment header") + testifyassert.NoError(t, file.Sync(), "sync WAL segment header rewrite") + testifyassert.NoError(t, file.Close(), "close WAL segment header rewrite") + } +} + +func firstBatchSequence(t *testing.T, segment string) uint64 { + t.Helper() + + records, err := wal.ParseRecordsFromFile(segment) + testifyrequire.NoError(t, err, "parse WAL records from %s", segment) + collector := wal.NewFragmentCollector() + for _, record := range records { + testifyrequire.NoError(t, collector.Append(record.Type, record.Payload), "collect WAL record fragments") + if !collector.IsComplete() { + continue + } + batch, err := wal.DecodeWalBatch(collector.BatchData()) + testifyrequire.NoError(t, err, "decode first WAL batch") + return batch.BaseSequence + } + t.Fatalf("segment %s contains no complete WAL batch", segment) + return 0 +}