From 0fe1530e25c6e8c63c3c8e15068c43dd55e56350 Mon Sep 17 00:00:00 2001 From: dailz Date: Fri, 12 Jun 2026 14:10:56 +0800 Subject: [PATCH] feat: integrate DB with Open/Close/Put/Delete/Get - Add db.go with DB struct: Open/Close/Put/Delete/Get/GetDurableSequence/IsWriteStopped - Add db_test.go with lifecycle, put/get, delete, recovery, and corruption tests - Extract sentinel errors to errkit/ leaf package to break import cycle (wal -> go_kv) - Update wal/sequence.go, wal/writer.go to import errkit instead of root package - Root errors.go re-exports from errkit for backward compatibility Phase 1 Wave 4d complete (T19). --- .omo/boulder.json | 30 ++++++- .omo/plans/phase1-wal.md | 2 +- db.go | 160 +++++++++++++++++++++++++++++++++++ db_test.go | 176 +++++++++++++++++++++++++++++++++++++++ errkit/errors.go | 27 ++++++ errors.go | 27 ++---- wal/sequence.go | 8 +- wal/sequence_test.go | 8 +- wal/writer.go | 14 ++-- 9 files changed, 415 insertions(+), 37 deletions(-) create mode 100644 db.go create mode 100644 db_test.go create mode 100644 errkit/errors.go diff --git a/.omo/boulder.json b/.omo/boulder.json index 86d0128..a865af7 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:02:11.867Z", + "updated_at": "2026-06-12T06:09:22.858Z", "session_ids": [ "opencode:ses_145c3bae9ffeTB2zbsTym0Cev8" ], @@ -120,6 +120,19 @@ "status": "completed", "ended_at": "2026-06-12T06:02:11.867Z", "elapsed_ms": 16713 + }, + "todo:19": { + "task_key": "todo:19", + "task_label": "19", + "task_title": "DB integration: Open/Close/Put/Delete/Get", + "session_id": "opencode:ses_1459293beffeLZq5Ac22xWZfBy", + "agent": "Sisyphus-Junior", + "category": "deep", + "updated_at": "2026-06-12T06:09:22.858Z", + "started_at": "2026-06-12T06:07:14.120Z", + "status": "completed", + "ended_at": "2026-06-12T06:09:22.858Z", + "elapsed_ms": 128738 } } } @@ -127,7 +140,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:02:11.867Z", + "updated_at": "2026-06-12T06:09:22.858Z", "session_ids": [ "opencode:ses_145c3bae9ffeTB2zbsTym0Cev8" ], @@ -239,6 +252,19 @@ "status": "completed", "ended_at": "2026-06-12T06:02:11.867Z", "elapsed_ms": 16713 + }, + "todo:19": { + "task_key": "todo:19", + "task_label": "19", + "task_title": "DB integration: Open/Close/Put/Delete/Get", + "session_id": "opencode:ses_1459293beffeLZq5Ac22xWZfBy", + "agent": "Sisyphus-Junior", + "category": "deep", + "updated_at": "2026-06-12T06:09:22.858Z", + "started_at": "2026-06-12T06:07:14.120Z", + "status": "completed", + "ended_at": "2026-06-12T06:09:22.858Z", + "elapsed_ms": 128738 } }, "agent": "atlas" diff --git a/.omo/plans/phase1-wal.md b/.omo/plans/phase1-wal.md index f1bc3d8..4d87287 100644 --- a/.omo/plans/phase1-wal.md +++ b/.omo/plans/phase1-wal.md @@ -1412,7 +1412,7 @@ Max Concurrent: 6 (Wave 1b) **Commit**: YES - Message: `feat: implement MANIFEST stub and CURRENT file` -- [ ] 19. DB integration: Open/Close/Put/Delete/Get +- [x] 19. DB integration: Open/Close/Put/Delete/Get **What to do**: - 创建 `db.go`：`DB` 结构体和核心 API diff --git a/db.go b/db.go new file mode 100644 index 0000000..cca34c4 --- /dev/null +++ b/db.go @@ -0,0 +1,160 @@ +package go_kv + +import ( + "fmt" + "os" + "sync" + "sync/atomic" + + "github.com/dailz/go-kv/config" + "github.com/dailz/go-kv/memtable" + "github.com/dailz/go-kv/wal" +) + +// DB is the public database handle for the storage engine. +type DB struct { + cfg *config.WalConfig + dir string + writer *wal.WalWriter + recoveryMem *memtable.MemTable + closed atomic.Bool +} + +// Open opens a database directory, recovers durable WAL entries, and starts a +// fresh WAL writer at the next recovered segment and sequence. +func Open(dir string, cfg *config.WalConfig) (*DB, error) { + if cfg == nil { + defaults := config.Defaults() + cfg = &defaults + } + if err := cfg.Validate(); err != nil { + return nil, fmt.Errorf("db: validate config: %w", err) + } + if err := os.MkdirAll(dir, 0o755); err != nil { + return nil, fmt.Errorf("db: create directory: %w", err) + } + + recoveryMem := memtable.NewMemTable(cfg.MemTableSize) + replayer := &memTableReplayer{mem: recoveryMem} + result, err := wal.Recover(dir, replayer) + if err != nil { + return nil, fmt.Errorf("db: recover WAL: %w", err) + } + if err := replayer.Err(); err != nil { + return nil, fmt.Errorf("db: replay WAL: %w", err) + } + recoveryMem.Publish(^uint64(0)) + + writer, err := wal.NewWalWriter(cfg, dir, result.NextSegmentID, result.NextSequence) + if err != nil { + return nil, fmt.Errorf("db: create WAL writer: %w", err) + } + + return &DB{ + cfg: cfg, + dir: dir, + writer: writer, + recoveryMem: recoveryMem, + }, nil +} + +// Close drains outstanding writes and releases database resources. +func (db *DB) Close() error { + if db.closed.Swap(true) { + return nil + } + if err := db.writer.Close(); err != nil { + return fmt.Errorf("db: close WAL writer: %w", err) + } + return nil +} + +// Put stores key with value. +func (db *DB) Put(key, value []byte) error { + if db.closed.Load() { + return ErrWriteStopped + } + if err := db.writer.Put(key, value); err != nil { + return fmt.Errorf("db: put: %w", err) + } + return nil +} + +// Delete removes key. +func (db *DB) Delete(key []byte) error { + if db.closed.Load() { + return ErrWriteStopped + } + if err := db.writer.Delete(key); err != nil { + return fmt.Errorf("db: delete: %w", err) + } + return nil +} + +// Get returns the newest visible value for key across recovered and new writes. +func (db *DB) Get(key []byte) *memtable.GetResult { + recovered := db.recoveryMem.Get(key) + current := db.writer.Get(key) + + if current.Found || current.Sequence > recovered.Sequence { + return current + } + if !recovered.Found { + return &memtable.GetResult{Found: false} + } + if recovered.Value == nil { + return &memtable.GetResult{Found: false, Sequence: recovered.Sequence} + } + return recovered +} + +// GetDurableSequence returns the highest sequence durably fsynced to WAL. +func (db *DB) GetDurableSequence() uint64 { + return db.writer.GetDurableSequence() +} + +// IsWriteStopped reports whether the DB rejects new writes. +func (db *DB) IsWriteStopped() bool { + return db.closed.Load() || db.writer.IsWriteStopped() +} + +type memTableReplayer struct { + mem *memtable.MemTable + mu sync.Mutex + err error +} + +func (r *memTableReplayer) ReplayPut(key, value []byte, sequence uint64) { + if err := r.mem.PutPending(cloneBytes(key), cloneBytes(value), sequence); err != nil { + r.setErr(fmt.Errorf("put pending: %w", err)) + } +} + +func (r *memTableReplayer) ReplayDelete(key []byte, sequence uint64) { + if err := r.mem.DeletePending(cloneBytes(key), sequence); err != nil { + r.setErr(fmt.Errorf("delete pending: %w", err)) + } +} + +func (r *memTableReplayer) Err() error { + r.mu.Lock() + defer r.mu.Unlock() + return r.err +} + +func (r *memTableReplayer) setErr(err error) { + r.mu.Lock() + defer r.mu.Unlock() + if r.err == nil { + r.err = err + } +} + +func cloneBytes(src []byte) []byte { + if src == nil { + return nil + } + dst := make([]byte, len(src)) + copy(dst, src) + return dst +} diff --git a/db_test.go b/db_test.go new file mode 100644 index 0000000..d2800f8 --- /dev/null +++ b/db_test.go @@ -0,0 +1,176 @@ +package go_kv + +import ( + "fmt" + "os" + "path/filepath" + "testing" +) + +func TestDBOpenClose(t *testing.T) { + db, err := Open(t.TempDir(), nil) + if err != nil { + t.Fatalf("Open: %v", err) + } + if err := db.Put([]byte("k"), []byte("v")); err != nil { + t.Fatalf("Put: %v", err) + } + if err := db.Close(); err != nil { + t.Fatalf("Close: %v", err) + } +} + +func TestDBPutGet(t *testing.T) { + db := openTestDB(t, t.TempDir()) + + if err := db.Put([]byte("k1"), []byte("v1")); err != nil { + t.Fatalf("Put: %v", err) + } + + got := db.Get([]byte("k1")) + if !got.Found { + t.Fatal("Get(k1) not found") + } + if string(got.Value) != "v1" { + t.Fatalf("Get(k1) value = %q, want %q", got.Value, "v1") + } +} + +func TestDBDelete(t *testing.T) { + db := openTestDB(t, t.TempDir()) + + if err := db.Put([]byte("k"), []byte("v")); err != nil { + t.Fatalf("Put: %v", err) + } + if err := db.Delete([]byte("k")); err != nil { + t.Fatalf("Delete: %v", err) + } + + got := db.Get([]byte("k")) + if got.Found { + t.Fatalf("Get(k) found deleted key with value %q", got.Value) + } +} + +func TestDBRecovery(t *testing.T) { + dir := t.TempDir() + db := openTestDB(t, dir) + + entries := map[string]string{ + "k1": "v1", + "k2": "v2", + "k3": "v3", + } + for key, value := range entries { + if err := db.Put([]byte(key), []byte(value)); err != nil { + t.Fatalf("Put(%q): %v", key, err) + } + } + closeTestDB(t, db) + + reopened := openTestDB(t, dir) + for key, want := range entries { + got := reopened.Get([]byte(key)) + if !got.Found { + t.Fatalf("Get(%q) not found after recovery", key) + } + if string(got.Value) != want { + t.Fatalf("Get(%q) value = %q, want %q", key, got.Value, want) + } + } +} + +func TestDBRecoveryAfterCorruption(t *testing.T) { + dir := t.TempDir() + db := openTestDB(t, dir) + + if err := db.Put([]byte("good1"), []byte("v1")); err != nil { + t.Fatalf("Put good1: %v", err) + } + if err := db.Put([]byte("good2"), []byte("v2")); err != nil { + t.Fatalf("Put good2: %v", err) + } + closeTestDB(t, db) + + appendWalTail(t, dir, []byte{0xde, 0xad, 0xbe, 0xef}) + + reopened := openTestDB(t, dir) + for key, want := range map[string]string{"good1": "v1", "good2": "v2"} { + got := reopened.Get([]byte(key)) + if !got.Found { + t.Fatalf("Get(%q) not found after corrupted-tail recovery", key) + } + if string(got.Value) != want { + t.Fatalf("Get(%q) value = %q, want %q", key, got.Value, want) + } + } +} + +func TestDBMultiplePuts(t *testing.T) { + dir := t.TempDir() + db := openTestDB(t, dir) + + for i := range 50 { + key := fmt.Appendf(nil, "k%d", i) + value := fmt.Appendf(nil, "v%d", i) + if err := db.Put(key, value); err != nil { + t.Fatalf("Put(%q): %v", key, err) + } + } + closeTestDB(t, db) + + reopened := openTestDB(t, dir) + for i := range 50 { + key := fmt.Appendf(nil, "k%d", i) + want := fmt.Sprintf("v%d", i) + got := reopened.Get(key) + if !got.Found { + t.Fatalf("Get(%q) not found after recovery", key) + } + if string(got.Value) != want { + t.Fatalf("Get(%q) value = %q, want %q", key, got.Value, want) + } + } +} + +func openTestDB(t *testing.T, dir string) *DB { + t.Helper() + + db, err := Open(dir, nil) + if err != nil { + t.Fatalf("Open: %v", err) + } + t.Cleanup(func() { + if !db.closed.Load() { + if err := db.Close(); err != nil { + t.Fatalf("Close cleanup: %v", err) + } + } + }) + return db +} + +func closeTestDB(t *testing.T, db *DB) { + t.Helper() + if err := db.Close(); err != nil { + t.Fatalf("Close: %v", err) + } +} + +func appendWalTail(t *testing.T, dir string, tail []byte) { + t.Helper() + + path := filepath.Join(dir, "segment-0.wal") + f, err := os.OpenFile(path, os.O_APPEND|os.O_WRONLY, 0) + if err != nil { + t.Fatalf("OpenFile(%s): %v", path, err) + } + defer func() { + if err := f.Close(); err != nil { + t.Fatalf("Close corrupted WAL file: %v", err) + } + }() + if _, err := f.Write(tail); err != nil { + t.Fatalf("Write corrupted WAL tail: %v", err) + } +} diff --git a/errkit/errors.go b/errkit/errors.go new file mode 100644 index 0000000..ecde56f --- /dev/null +++ b/errkit/errors.go @@ -0,0 +1,27 @@ +// Package errkit defines sentinel errors shared across packages in the go-kv module. +// +// These errors live in a leaf package to avoid import cycles: both the root +// package and sub-packages (wal, memtable, etc.) can import errkit without +// creating circular dependencies. +package errkit + +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") diff --git a/errors.go b/errors.go index 461c8ce..22f3169 100644 --- a/errors.go +++ b/errors.go @@ -1,22 +1,11 @@ package go_kv -import "errors" +import "github.com/dailz/go-kv/errkit" -// 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") +var ( + ErrCommitUnknown = errkit.ErrCommitUnknown + ErrWriteStopped = errkit.ErrWriteStopped + ErrSequenceExhausted = errkit.ErrSequenceExhausted + ErrWALCorrupted = errkit.ErrWALCorrupted + ErrInvalidConfig = errkit.ErrInvalidConfig +) diff --git a/wal/sequence.go b/wal/sequence.go index c2317e9..a6843f7 100644 --- a/wal/sequence.go +++ b/wal/sequence.go @@ -3,7 +3,7 @@ package wal import ( "sync/atomic" - "github.com/dailz/go-kv" + "github.com/dailz/go-kv/errkit" ) // SequenceManager manages monotonic sequence number allocation for the WAL. @@ -29,16 +29,16 @@ func NewSequenceManager(recoveredSequence uint64) *SequenceManager { // 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 + return 0, errkit.ErrSequenceExhausted } for { if sm.exhausted.Load() { - return 0, go_kv.ErrSequenceExhausted + return 0, errkit.ErrSequenceExhausted } base := sm.nextSequence.Load() last := base + uint64(count) - 1 if last < base { - return 0, go_kv.ErrSequenceExhausted + return 0, errkit.ErrSequenceExhausted } newNext := last + 1 if !sm.nextSequence.CompareAndSwap(base, newNext) { diff --git a/wal/sequence_test.go b/wal/sequence_test.go index 36c0381..d63340e 100644 --- a/wal/sequence_test.go +++ b/wal/sequence_test.go @@ -7,7 +7,7 @@ import ( "sync/atomic" "testing" - "github.com/dailz/go-kv" + "github.com/dailz/go-kv/errkit" ) func TestSequenceAllocation(t *testing.T) { @@ -49,7 +49,7 @@ func TestSequenceOverflow(t *testing.T) { // Remaining: MaxUint64-2, MaxUint64-1, MaxUint64 = 3 slots. // Asking for 5 should overflow. _, err := sm.AllocateBatch(5) - if !errors.Is(err, go_kv.ErrSequenceExhausted) { + if !errors.Is(err, errkit.ErrSequenceExhausted) { t.Fatalf("expected ErrSequenceExhausted, got %v", err) } @@ -64,7 +64,7 @@ func TestSequenceOverflow(t *testing.T) { // Now any further allocation should fail. _, err = sm.AllocateBatch(1) - if !errors.Is(err, go_kv.ErrSequenceExhausted) { + if !errors.Is(err, errkit.ErrSequenceExhausted) { t.Fatalf("expected ErrSequenceExhausted after exhaustion, got %v", err) } } @@ -108,7 +108,7 @@ func TestZeroCountRejected(t *testing.T) { sm := NewSequenceManager(0) _, err := sm.AllocateBatch(0) - if !errors.Is(err, go_kv.ErrSequenceExhausted) { + if !errors.Is(err, errkit.ErrSequenceExhausted) { t.Fatalf("expected ErrSequenceExhausted for count=0, got %v", err) } } diff --git a/wal/writer.go b/wal/writer.go index 7de8261..ca79047 100644 --- a/wal/writer.go +++ b/wal/writer.go @@ -6,7 +6,7 @@ import ( "sync/atomic" "time" - go_kv "github.com/dailz/go-kv" + "github.com/dailz/go-kv/errkit" "github.com/dailz/go-kv/config" "github.com/dailz/go-kv/memtable" ) @@ -131,7 +131,7 @@ func (ww *WalWriter) processRemaining() { func (ww *WalWriter) processBatch(requests []*CommitRequest) { if ww.writeStopped.Load() { - ww.sendError(requests, go_kv.ErrWriteStopped) + ww.sendError(requests, errkit.ErrWriteStopped) return } @@ -159,20 +159,20 @@ func (ww *WalWriter) processBatch(requests []*CommitRequest) { } if err := ww.segManager.AppendBatch(encoded); err != nil { - ww.stopWithError(requests, go_kv.ErrCommitUnknown) + ww.stopWithError(requests, errkit.ErrCommitUnknown) return } active := ww.memTables.GetActive() if err := writePending(active, entries, baseSequence); err != nil { abortRange(active, baseSequence, lastSequence) - ww.stopWithError(requests, go_kv.ErrCommitUnknown) + ww.stopWithError(requests, errkit.ErrCommitUnknown) return } if err := ww.segManager.Sync(); err != nil { abortRange(active, baseSequence, lastSequence) - ww.stopWithError(requests, go_kv.ErrCommitUnknown) + ww.stopWithError(requests, errkit.ErrCommitUnknown) return } @@ -263,7 +263,7 @@ func (ww *WalWriter) sendSuccess(requests []*CommitRequest, baseSequence uint64, // Put stores key with value. func (ww *WalWriter) Put(key, value []byte) error { if ww.writeStopped.Load() { - return go_kv.ErrWriteStopped + return errkit.ErrWriteStopped } req := ww.queue.Submit([]*WalEntry{{ @@ -279,7 +279,7 @@ func (ww *WalWriter) Put(key, value []byte) error { // Delete removes key. func (ww *WalWriter) Delete(key []byte) error { if ww.writeStopped.Load() { - return go_kv.ErrWriteStopped + return errkit.ErrWriteStopped } req := ww.queue.Submit([]*WalEntry{{