90 lines
2.2 KiB
Go
90 lines
2.2 KiB
Go
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()
|
|
}
|