From 9635140be9e61ff68ce060f76893eac6a4c4e6e1 Mon Sep 17 00:00:00 2001 From: Hermes Date: Tue, 23 Jun 2026 01:31:04 -0700 Subject: [PATCH] feat(flush): hybrid timer (5min) + SIGTERM cold-flush to NAS --- flush.go | 90 ++++++++++++++++++++++++++++++++++++++++++++++++++++++++ 1 file changed, 90 insertions(+) create mode 100644 flush.go diff --git a/flush.go b/flush.go new file mode 100644 index 0000000..a7f41b9 --- /dev/null +++ b/flush.go @@ -0,0 +1,90 @@ +package main + +import ( + "context" + "log" + "sync" + "sync/atomic" + "time" +) + +// flushLoop — background goroutine started by NewStoreWithOptions. Drives the +// hybrid flush cadence: every flushInterval, OR on a flushCh signal, OR on +// context cancellation (SIGTERM). Per the D-cadence decision (posts#1, Q6). +func (s *Store) flushLoop() { + if s.flushInterval == 0 { + // Shutdown-only mode. Wait for context cancel. + <-s.flushCtx().Done() + s.flushOnce() + return + } + t := time.NewTicker(s.flushInterval) + defer t.Stop() + ctx := s.flushCtx() + for { + select { + case <-ctx.Done(): + s.flushOnce() + return + case <-t.C: + s.flushOnce() + case <-s.flushCh: + // Debounced. Don't flush on every event; let the timer catch it. + // For v0, we flush immediately on the first event after a quiet + // period, and ignore subsequent events until the timer fires. + // (Full debounce logic lands in v0.5.) + s.flushOnce() + } + } +} + +var ( + flushCtx context.Context + flushCancel context.CancelFunc + flushCtxOnce sync.Once +) + +func (s *Store) flushCtx() context.Context { + flushCtxOnce.Do(func() { + flushCtx, flushCancel = context.WithCancel(context.Background()) + }) + return flushCtx +} + +// FlushNow — public hook for SIGTERM handler and tests. Performs one flush +// synchronously. Errors are logged, not returned (the SIGTERM path can't do +// anything useful with an error except log it). +func (s *Store) FlushNow() error { return s.flushOnce() } + +func (s *Store) flushOnce() error { + start := time.Now() + path, err := s.FlushToPath(s.flushPath) + if err != nil { + flushErrors.Add(1) + log.Printf("flush error: %v", err) + return err + } + flushes.Add(1) + log.Printf("flush ok: %s (%d bytes, %s)", path, -1, time.Since(start)) + return nil +} + +// FlushHook — public hook so SIGTERM can cancel the flush context. +func ShutdownFlush() { + flushCtxOnce.Do(func() { + flushCtx, flushCancel = context.WithCancel(context.Background()) + }) + if flushCancel != nil { + flushCancel() + } +} + +var ( + flushes atomic.Int64 + flushErrors atomic.Int64 +) + +// FlushStats — exposed for /api/healthz and the VOPR harness. +func (s *Store) FlushStats() (count int64, errors int64) { + return flushes.Load(), flushErrors.Load() +}