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() }