From 762a299839706ebaee28ca18e4145792d559623e Mon Sep 17 00:00:00 2001 From: ChromaBeast Date: Thu, 12 Nov 2026 17:20:30 +0530 Subject: [PATCH 1/2] test(metrics): only require disk-free gauge where the platform reports it DiskFreeBytes is unix-only, so TestMetricsAuthAndFormat failed on Windows CI with a missing beastdb_disk_free_bytes line. Require the gauge only when DiskFreeBytes reports support. Co-Authored-By: Claude Sonnet 5.5 --- internal/web/handler/metrics_test.go | 11 ++++++++--- 1 file changed, 8 insertions(+), 3 deletions(-) diff --git a/internal/web/handler/metrics_test.go b/internal/web/handler/metrics_test.go index 7b04a64..5e8d651 100644 --- a/internal/web/handler/metrics_test.go +++ b/internal/web/handler/metrics_test.go @@ -119,14 +119,19 @@ func TestMetricsAuthAndFormat(t *testing.T) { } body := rec.Body.String() validateExposition(t, body) - for _, want := range []string{ + wants := []string{ `beastdb_build_info{version="v\"1",role="follower"} 1`, "beastdb_wal_size_bytes ", "beastdb_wal_current_lsn ", "beastdb_buffer_pool_dirty_pages ", "beastdb_buffer_pool_pinned_pages ", "beastdb_replica_applied_lsn 7", `beastdb_requests_total{protocol="grpc",method="/beastdb.v1.KV/Get",code="NotFound"} 1`, `beastdb_request_duration_seconds_bucket{protocol="grpc",method="/beastdb.v1.KV/Get",le="+Inf"} 2`, - "beastdb_disk_free_bytes ", - } { + } + // Disk-free is only reported on platforms that support it (unix); it is + // omitted elsewhere, e.g. Windows. + if _, ok := metrics.DiskFreeBytes("."); ok { + wants = append(wants, "beastdb_disk_free_bytes ") + } + for _, want := range wants { if !strings.Contains(body, want) { t.Errorf("missing %q in output", want) } From f851bd6063b6fbaead892d18d64ab1270afc4dac Mon Sep 17 00:00:00 2001 From: ChromaBeast Date: Thu, 12 Nov 2026 17:20:30 +0530 Subject: [PATCH 2/2] perf(engine): let the first writer lead its own commit group Instead of waking a dedicated committer goroutine for every write, the writer that finds no commit in progress runs the queued group on its own goroutine; if more writers queued meanwhile, a drain goroutine serves them. Group commit semantics are unchanged (one fsync per group, lock held across it, observer in LSN order). Aims to remove a cross-goroutine wake-up per uncontended write, which can cost up to a timer tick on Windows. Not yet verified on Windows. Co-Authored-By: Claude Sonnet 5.5 --- internal/api/engine_commit.go | 116 +++++++++++++++++++++------------- 1 file changed, 71 insertions(+), 45 deletions(-) diff --git a/internal/api/engine_commit.go b/internal/api/engine_commit.go index 77b5c55..21788d3 100644 --- a/internal/api/engine_commit.go +++ b/internal/api/engine_commit.go @@ -23,21 +23,26 @@ type writeReq struct { ok bool // run appended a WAL record (so it depends on the batch fsync) } -// committer is the group-commit queue. A single goroutine drains it: each pass -// takes every queued write, runs them back to back under the exclusive engine -// lock (appending to the WAL without fsync), issues ONE fsync for the whole -// group, notifies the replication observer in LSN order, and only then -// releases the lock and acknowledges the writers. Because the lock is held -// across the fsync, readers and the observer never see a write before it is -// durable, exactly as with fsync-per-write; the gain is that N queued writers -// pay for one fsync instead of N. +// committer is the group-commit queue. Writers enqueue a request; whichever +// writer finds no commit in progress becomes the leader and runs the queued +// group itself, on its own goroutine. This avoids a cross-goroutine wake-up +// for an uncontended write, which costs up to a timer tick on Windows. If more +// writers queued while the leader was busy, the leader hands the remaining work +// to a drain goroutine and returns to its caller. +// +// A group runs back to back under the exclusive engine lock (appending to the +// WAL without fsync), issues ONE fsync for the whole group, notifies the +// replication observer in LSN order, and only then releases the lock and +// acknowledges the writers. Because the lock is held across the fsync, readers +// and the observer never see a write before it is durable, exactly as with +// fsync-per-write; the gain is that N queued writers pay for one fsync. +// Only one group runs at a time, so queue order is LSN order. type committer struct { mu sync.Mutex queue []*writeReq - started bool + running bool // a leader or drain goroutine owns the queue closed bool - wake chan struct{} - done chan struct{} + idle *sync.Cond // signalled (on mu) when running becomes false } // commit runs fn through the group-commit pipeline and returns once its WAL @@ -53,59 +58,80 @@ func (e *Engine) commit(fn func() error) error { c.mu.Unlock() return ErrEngineClosed } - if !c.started { - c.started = true - c.wake = make(chan struct{}, 1) - c.done = make(chan struct{}) - go e.commitLoop() - } c.queue = append(c.queue, req) + lead := !c.running + if lead { + c.running = true + } c.mu.Unlock() - select { - case c.wake <- struct{}{}: - default: + if lead { + e.lead() } <-req.done return req.err } -func (e *Engine) commitLoop() { +// takeBatch removes and returns the queued writes. When the queue is empty it +// clears running (releasing leadership) and returns nil. +func (c *committer) takeBatch() []*writeReq { + c.mu.Lock() + defer c.mu.Unlock() + batch := c.queue + c.queue = nil + if len(batch) == 0 { + c.running = false + if c.idle != nil { + c.idle.Broadcast() + } + } + return batch +} + +// lead runs one group (which includes the leader's own request) on the +// caller's goroutine. If more writers arrived meanwhile, a drain goroutine +// keeps serving them so the leader's caller is not delayed. +func (e *Engine) lead() { + c := &e.cq + e.runBatch(c.takeBatch()) + c.mu.Lock() + more := len(c.queue) > 0 + if !more { + c.running = false + if c.idle != nil { + c.idle.Broadcast() + } + } + c.mu.Unlock() + if more { + go e.drain() + } +} + +// drain serves queued writes until the queue is empty, then releases leadership. +func (e *Engine) drain() { c := &e.cq - defer close(c.done) for { - <-c.wake - for { - c.mu.Lock() - batch := c.queue - c.queue = nil - closed := c.closed - c.mu.Unlock() - if len(batch) == 0 { - if closed { - return - } - break - } - e.runBatch(batch) + batch := c.takeBatch() + if len(batch) == 0 { + return } + e.runBatch(batch) } } -// stopCommitter drains outstanding writes and stops the commit goroutine. +// stopCommitter rejects new writes and waits for in-flight ones to finish. func (e *Engine) stopCommitter() { c := &e.cq c.mu.Lock() c.closed = true - started := c.started - c.mu.Unlock() - if started { - select { - case c.wake <- struct{}{}: - default: - } - <-c.done + if c.idle == nil { + c.idle = sync.NewCond(&c.mu) } + for c.running { + c.idle.Wait() + } + c.mu.Unlock() } func (e *Engine) runBatch(batch []*writeReq) {