From df642cb1d47c7455b8dcb106296d0f83b657d9bd Mon Sep 17 00:00:00 2001 From: Claude Date: Fri, 21 Aug 2026 20:08:45 +0000 Subject: [PATCH] Move history persistence's fsync off the collector's tick goroutine persistHistory() ran synchronously inside Run's select loop: encoding and atomically writing (with fsync) the history file blocked fastTick from running on every slowTick and on shutdown. time.Ticker drops rather than queues missed ticks, so a slow write (e.g. an SD-card fsync stall) silently dropped fast-tick samples. The snapshot is still taken on the caller's goroutine for a consistent point-in-time view, but the encode-and-write now runs in a background goroutine tracked by a new persistWG. A buffered try-lock (persisting) skips an overlapping flush rather than queuing it, since the next flush writes newer data anyway. Run's ctx.Done() branch now waits on persistWG before returning, so the final flush is still guaranteed to land before shutdown completes (bounded by main.go's existing 10s shutdown context). Closes #110 --- docs/ARCHITECTURE.md | 12 +++ internal/collector/collector.go | 22 ++++- internal/collector/persist.go | 30 +++++- internal/collector/persist_test.go | 144 +++++++++++++++++++++++++++++ 4 files changed, 202 insertions(+), 6 deletions(-) diff --git a/docs/ARCHITECTURE.md b/docs/ARCHITECTURE.md index a8c86c7..5a03dce 100644 --- a/docs/ARCHITECTURE.md +++ b/docs/ARCHITECTURE.md @@ -125,6 +125,18 @@ points per series so a corrupt or malicious file cannot trigger an oversized all - **Written**: on every `slowTick` (i.e. every `updates_check_minutes`) and once more on clean shutdown (`ctx.Done()` in `Run`), so a clean stop loses at most the points collected since the last fast tick. +- **Written asynchronously**: `persistHistory` snapshots `Collector.History()` on the + calling goroutine (a consistent point-in-time view), then hands the encode-and-write off + to a background goroutine tracked by `Collector.persistWG`. On a Pi the atomic write's + `fsync` can stall the SD card for hundreds of milliseconds, and `Run`'s `time.Ticker` + drops rather than queues any fast tick that falls due meanwhile — running the write + inline on the tick goroutine would silently punch a hole in the very history being + persisted. A buffered try-lock (`Collector.persisting`) skips — rather than queues — a + flush while a previous write is still in flight, since the next flush writes newer data + anyway. `Run`'s `ctx.Done()` branch calls `persistWG.Wait()` after the final + `persistHistory()` so the last flush is guaranteed to land before `Run` returns; this is + bounded by `cmd/pimonitor/main.go`'s shutdown context (it selects on `collDone` vs. + `shutdownCtx.Done()`), so a stuck fsync cannot hang the process. - **Written atomically**: `writeFileAtomic` writes to a temp file in the same directory, `fsync`s it, then renames it over the target path — a crash mid-write can never leave a partially written history file behind. diff --git a/internal/collector/collector.go b/internal/collector/collector.go index bb9cd7e..fb972d4 100644 --- a/internal/collector/collector.go +++ b/internal/collector/collector.go @@ -118,6 +118,18 @@ type Collector struct { diskHist map[string]*RingBuffer[HistoryPoint] rxHist map[string]*RingBuffer[HistoryPoint] txHist map[string]*RingBuffer[HistoryPoint] + + // persistWG tracks in-flight persistHistory writes so Run's ctx.Done() + // branch can wait for the final flush before returning. + persistWG sync.WaitGroup + // persisting is a buffered try-lock (capacity 1): a successful send means + // no write is currently in flight. Used to skip an overlapping flush + // rather than queue it, since the next flush writes newer data anyway. + persisting chan struct{} + // writeFile performs the atomic write persistHistory hands off to a + // background goroutine. Defaults to writeFileAtomic; overridable in + // tests to control write timing without touching real disk I/O. + writeFile func(path string, data []byte) error } // New creates a Collector wired to the standard Linux metric sources. @@ -164,6 +176,9 @@ func New(cfg Config, log *slog.Logger) *Collector { diskHist: make(map[string]*RingBuffer[HistoryPoint]), rxHist: make(map[string]*RingBuffer[HistoryPoint]), txHist: make(map[string]*RingBuffer[HistoryPoint]), + + persisting: make(chan struct{}, 1), + writeFile: writeFileAtomic, } c.fastInterval = c.clampInterval(cfg.FastInterval, "FastInterval") return c @@ -199,8 +214,13 @@ func (c *Collector) Run(ctx context.Context) { select { case <-ctx.Done(): // Final flush so a clean shutdown (e.g. reboot for updates) - // loses at most the points since the last fast tick. + // loses at most the points since the last fast tick. persistHistory + // hands the write off to a background goroutine, so wait for it + // (and any still-running flush from the last slowTick) before + // returning: cmd/pimonitor/main.go bounds this by its shutdown + // context, so a stuck fsync cannot hang the process. c.persistHistory() + c.persistWG.Wait() return case <-fastTicker.C: c.fastTick(ctx) diff --git a/internal/collector/persist.go b/internal/collector/persist.go index 3bf4bcd..4cd457c 100644 --- a/internal/collector/persist.go +++ b/internal/collector/persist.go @@ -306,16 +306,36 @@ func writeFileAtomic(path string, data []byte) error { return nil } -// persistHistory snapshots the current metric history to cfg.PersistPath -// with an atomic write. Failures are logged, not fatal: persistence is -// best-effort and must never take down metric collection. +// persistHistory snapshots the current metric history and writes it to +// cfg.PersistPath asynchronously. The snapshot itself is taken on the +// caller's goroutine so it is a consistent point-in-time view, but the +// atomic write (which fsyncs) happens in a background goroutine: on a Pi an +// SD-card fsync can stall for hundreds of milliseconds, and Run's ticker +// drops (rather than queues) any fast tick that falls due meanwhile, so a +// slow flush would silently punch a hole in the very history being +// persisted. If a previous flush's write is still in flight, this call is +// skipped entirely rather than queued — the next flush writes newer data +// anyway. Failures are logged, not fatal: persistence is best-effort and +// must never take down metric collection. func (c *Collector) persistHistory() { if c.cfg.PersistPath == "" { return } - if err := writeFileAtomic(c.cfg.PersistPath, encodeHistory(c.History())); err != nil { - c.log.Warn("could not persist metric history", "path", c.cfg.PersistPath, "error", err) + select { + case c.persisting <- struct{}{}: + default: + return } + + data := encodeHistory(c.History()) + c.persistWG.Add(1) + go func() { + defer c.persistWG.Done() + defer func() { <-c.persisting }() + if err := c.writeFile(c.cfg.PersistPath, data); err != nil { + c.log.Warn("could not persist metric history", "path", c.cfg.PersistPath, "error", err) + } + }() } // loadHistory restores metric history from cfg.PersistPath, if present, diff --git a/internal/collector/persist_test.go b/internal/collector/persist_test.go index bb0fcc5..1d0f425 100644 --- a/internal/collector/persist_test.go +++ b/internal/collector/persist_test.go @@ -4,6 +4,7 @@ import ( "context" "os" "path/filepath" + "sync/atomic" "testing" "time" ) @@ -161,6 +162,7 @@ func TestCollector_PersistAndLoadHistory(t *testing.T) { c1 := newPersistTestCollector(path, time.Hour) c1.importHistory(fixtureHistory(now.Add(-time.Minute)), now) c1.persistHistory() + c1.persistWG.Wait() c2 := newPersistTestCollector(path, time.Hour) c2.loadHistory() @@ -243,8 +245,10 @@ func TestCollector_PersistHistory_OverwritesExistingFile(t *testing.T) { c := newPersistTestCollector(path, time.Hour) c.persistHistory() // first write: empty history + c.persistWG.Wait() c.importHistory(fixtureHistory(now.Add(-time.Minute)), now) c.persistHistory() // second write must atomically replace the first + c.persistWG.Wait() c2 := newPersistTestCollector(path, time.Hour) c2.loadHistory() @@ -282,3 +286,143 @@ func TestCollector_Run_PersistsOnShutdown(t *testing.T) { t.Fatal("expected history persisted on shutdown to contain points") } } + +// blockingWriteFile returns a writeFile replacement that signals started the +// first time it is invoked and then blocks until proceed is closed, before +// delegating to writeFileAtomic. calls counts every invocation. +func blockingWriteFile(started chan<- struct{}, proceed <-chan struct{}, calls *int32) func(string, []byte) error { + var signaled int32 + return func(path string, data []byte) error { + atomic.AddInt32(calls, 1) + if atomic.CompareAndSwapInt32(&signaled, 0, 1) { + close(started) + } + <-proceed + return writeFileAtomic(path, data) + } +} + +func TestCollector_PersistHistory_DoesNotBlockOnSlowWrite(t *testing.T) { + path := filepath.Join(t.TempDir(), "history.bin") + now := time.Now().Truncate(time.Millisecond) + + c := newPersistTestCollector(path, time.Hour) + c.importHistory(fixtureHistory(now.Add(-time.Minute)), now) + + started := make(chan struct{}) + proceed := make(chan struct{}) + var calls int32 + c.writeFile = blockingWriteFile(started, proceed, &calls) + + returned := make(chan struct{}) + go func() { + c.persistHistory() + close(returned) + }() + + select { + case <-returned: + case <-time.After(2 * time.Second): + t.Fatal("persistHistory did not return while the write was blocked") + } + + select { + case <-started: + case <-time.After(2 * time.Second): + t.Fatal("write function was never invoked") + } + + // The data must still land once the slow write is allowed to finish. + close(proceed) + c.persistWG.Wait() + + c2 := newPersistTestCollector(path, time.Hour) + c2.loadHistory() + historiesEqual(t, c2.History(), c.History()) +} + +func TestCollector_PersistHistory_SkipsOverlappingWrites(t *testing.T) { + path := filepath.Join(t.TempDir(), "history.bin") + c := newPersistTestCollector(path, time.Hour) + + started := make(chan struct{}) + proceed := make(chan struct{}) + var calls int32 + c.writeFile = blockingWriteFile(started, proceed, &calls) + + c.persistHistory() // starts the write, which blocks + select { + case <-started: + case <-time.After(2 * time.Second): + t.Fatal("write function was never invoked") + } + + // Both of these must be skipped: a write is already in flight. + c.persistHistory() + c.persistHistory() + + close(proceed) + c.persistWG.Wait() + + if got := atomic.LoadInt32(&calls); got != 1 { + t.Fatalf("write function invoked %d times while overlapping, want 1", got) + } +} + +func TestCollector_Run_ShutdownWaitsForFinalFlush(t *testing.T) { + path := filepath.Join(t.TempDir(), "history.bin") + c := New(Config{ + FastInterval: 10 * time.Millisecond, + SlowInterval: time.Hour, + HistoryCapacity: 100, + PersistPath: path, + HistoryWindow: time.Hour, + }, nil) + + started := make(chan struct{}) + proceed := make(chan struct{}) + var calls int32 + c.writeFile = blockingWriteFile(started, proceed, &calls) + + ctx, cancel := context.WithCancel(context.Background()) + done := make(chan struct{}) + go func() { + c.Run(ctx) + close(done) + }() + + select { + case <-started: + t.Fatal("final flush must not start before shutdown") + case <-time.After(50 * time.Millisecond): + } + + cancel() + + select { + case <-started: + case <-time.After(2 * time.Second): + t.Fatal("final flush was never started") + } + + // Run must block on the in-flight flush rather than returning early. + select { + case <-done: + t.Fatal("Run returned before the blocked final flush completed") + case <-time.After(200 * time.Millisecond): + } + + close(proceed) + + select { + case <-done: + case <-time.After(2 * time.Second): + t.Fatal("Run did not return after the final flush completed") + } + + c2 := newPersistTestCollector(path, time.Hour) + c2.loadHistory() + if len(c2.History().MemoryUsedPercent) == 0 { + t.Fatal("expected history persisted on shutdown to contain points") + } +}