From e6b4e648ce73f3fe7cc5e6a8efc862be50205abb Mon Sep 17 00:00:00 2001 From: sneak Date: Wed, 14 Oct 2026 11:59:00 +0000 Subject: [PATCH] Stopping the server waits for report file writes under way (closes #122) The report buffer's stop now waits, through a sync.WaitGroup, for flushLoop and for the write Append starts at 10 MiB before its final flush. It used to return at once, so the process exiting cut such a write off and its reports were lost. A write under way that fails puts its reports back in the buffer, so the final flush writes them; the stop fails only if that flush fails too. Tests hold the periodic write and the one at 10 MiB open in a synctest bubble and find the stop waiting until each is released; a third makes the held write fail and finds its report written by the final flush. Model: opus-5-5 --- TODO.md | 12 ++ internal/reportbuf/reportbuf.go | 17 +- internal/reportbuf/reportbuf_test.go | 225 ++++++++++++++++++++++++--- 3 files changed, 232 insertions(+), 22 deletions(-) diff --git a/TODO.md b/TODO.md index a12f8da..302faf8 100644 --- a/TODO.md +++ b/TODO.md @@ -21,6 +21,18 @@ Run `make frontend-viewport-test` in CI as its own step; it is not part of # Completed Steps +- 2026-10-14: stopping the server waits for a report file write already under + way, the periodic flush's or the one `Append` starts when 10 MiB are waiting, + before the final flush + ([#122](https://git.eeqj.de/sneak/netwatch/issues/122)). It used to return at + once, and the process exiting cut the write off, losing its reports. A write + under way that fails puts its reports back in memory, so the final flush + writes them; the stop fails, and the process exits non-zero, only if that + flush fails too. Two tests hold the periodic write and the one at 10 MiB open + and find the stop waiting until each is released; a third makes the held write + fail and finds its report written by the final flush. They run in a + `testing/synctest` bubble, so the periodic flush comes without waiting a + minute - 2026-10-14: `make test` no longer runs the frontend's production build, which the tests do not need ([#118](https://git.eeqj.de/sneak/netwatch/issues/118)). `yarn build` moved out of the stage that runs the frontend's unit tests, now diff --git a/internal/reportbuf/reportbuf.go b/internal/reportbuf/reportbuf.go index 62feba6..924a7a2 100644 --- a/internal/reportbuf/reportbuf.go +++ b/internal/reportbuf/reportbuf.go @@ -96,6 +96,10 @@ type Buffer struct { // refuses reports, and the reports of that write wait in buf for // the next flush. writeErr error + // writers are the goroutines that write report files: flushLoop, + // and each one Append starts at flushSizeThreshold. OnStop waits + // for them before the final flush. + writers sync.WaitGroup } // New creates a Buffer and registers lifecycle hooks to @@ -147,7 +151,7 @@ func New( b.mu.Unlock() - go b.flushLoop() + b.writers.Go(b.flushLoop) return nil }, @@ -158,7 +162,14 @@ func New( var err error b.stopOnce.Do(func() { + // Closing done ends flushLoop after any flush it is + // running. Wait for it and for the writes Append + // started, which the process exiting would cut off; + // one that fails puts its reports back in buf, for the + // final flush to write. close(b.done) + b.writers.Wait() + err = b.flushLocked() }) @@ -211,12 +222,12 @@ func (b *Buffer) Append(v any) error { data := b.drainBuf() b.mu.Unlock() - go func() { + b.writers.Go(func() { writeErr := b.writeFile(data) if writeErr != nil { b.log.Error("flush reports failed", "error", writeErr) } - }() + }) return nil } diff --git a/internal/reportbuf/reportbuf_test.go b/internal/reportbuf/reportbuf_test.go index eba53f4..9e277f4 100644 --- a/internal/reportbuf/reportbuf_test.go +++ b/internal/reportbuf/reportbuf_test.go @@ -15,6 +15,7 @@ import ( "sync" "sync/atomic" "testing" + "testing/synctest" "time" "sneak.berlin/go/netwatch/internal/config" @@ -107,9 +108,174 @@ func TestFailedFinalFlushFailsStop(t *testing.T) { } } -// startBuffer starts a Buffer through fx, as main does, with the -// DATA_DIR and DATA_DIR_MAX_BYTES the calling test has set. -func startBuffer(t *testing.T) *reportbuf.Buffer { +// TestStopWaitsForPeriodicWrite holds open the write the periodic flush +// starts, then stops the app. The stop must wait until the write is +// released, and the report must then be in its file. The test runs in +// a synctest bubble, whose clock moves on at once when every goroutine +// in it is waiting, so the periodic flush does not take a minute. +func TestStopWaitsForPeriodicWrite(t *testing.T) { + dir := t.TempDir() + t.Setenv("DATA_DIR", dir) + + synctest.Test(t, func(t *testing.T) { + buf, app := startApp(t) + + created := make(chan struct{}) + release := make(chan struct{}) + + buf.OnFileCreated(func(*os.File) { + created <- struct{}{} + + <-release + }) + + err := buf.Append(map[string]string{"id": "periodic"}) + if err != nil { + t.Fatalf("append report: %v", err) + } + + <-created + + err = stopWhileHeld(t, app, release) + if err != nil { + t.Fatalf("stop: %v", err) + } + + files, err := readReportFiles(dir) + if err != nil { + t.Fatalf("read report files: %v", err) + } + + want := `{"id":"periodic"}` + "\n" + if len(files) != 1 || files[0] != want { + t.Fatalf("report files = %q, want one holding %q", files, want) + } + }) +} + +// TestStopWaitsForWriteAtSizeThreshold holds open the write Append +// starts when the buffer reaches FlushSizeThreshold, then stops the +// app. The stop must wait until the write is released, and the reports +// must then be in its file. +func TestStopWaitsForWriteAtSizeThreshold(t *testing.T) { + dir := t.TempDir() + t.Setenv("DATA_DIR", dir) + + synctest.Test(t, func(t *testing.T) { + buf, app := startApp(t) + + created := make(chan struct{}) + release := make(chan struct{}) + + buf.OnFileCreated(func(*os.File) { + created <- struct{}{} + + <-release + }) + + appended := appendToThreshold(t, buf) + + <-created + + err := stopWhileHeld(t, app, release) + if err != nil { + t.Fatalf("stop: %v", err) + } + + files, err := readReportFiles(dir) + if err != nil { + t.Fatalf("read report files: %v", err) + } + + if len(files) != 1 || files[0] != appended { + t.Fatalf("%d report files, want one holding the %d bytes "+ + "appended", len(files), len(appended)) + } + }) +} + +// TestFinalFlushWritesReportsOfFailedWrite holds open the write the +// periodic flush starts, makes it fail, then stops the app. The stop +// waits for the write, which puts its report back in the buffer, and +// the final flush writes it, so the stop succeeds with nothing lost. +func TestFinalFlushWritesReportsOfFailedWrite(t *testing.T) { + dir := t.TempDir() + t.Setenv("DATA_DIR", dir) + + synctest.Test(t, func(t *testing.T) { + buf, app := startApp(t) + + created := make(chan *os.File) + release := make(chan struct{}) + + buf.OnFileCreated(func(f *os.File) { + created <- f + + <-release + }) + + err := buf.Append(map[string]string{"id": "failed"}) + if err != nil { + t.Fatalf("append report: %v", err) + } + + held := <-created + + // Closing the file under the write makes the write fail. + _ = held.Close() + + // The final flush's write goes through. + buf.OnFileCreated(func(*os.File) {}) + + err = stopWhileHeld(t, app, release) + if err != nil { + t.Fatalf("stop: %v", err) + } + + files, err := readReportFiles(dir) + if err != nil { + t.Fatalf("read report files: %v", err) + } + + want := `{"id":"failed"}` + "\n" + if len(files) != 1 || files[0] != want { + t.Fatalf("report files = %q, want one holding %q", files, want) + } + }) +} + +// stopWhileHeld stops app while a report file write is held open until +// release is closed, and fails the test if the stop returns before +// that. It then closes release and returns the stop's error. It must +// be called in a synctest bubble. +func stopWhileHeld( + t *testing.T, + app *fxtest.App, + release chan<- struct{}, +) error { + t.Helper() + + stopped := make(chan error, 1) + + go func() { stopped <- app.Stop(t.Context()) }() + + // Wait returns once every other goroutine in the bubble is waiting: + // the stop has got as far as it can while the write is held. + synctest.Wait() + + if len(stopped) != 0 { + t.Error("stop returned while a report file write was held open") + } + + close(release) + + return <-stopped +} + +// startApp starts a Buffer through fx, as main does, with the DATA_DIR +// and DATA_DIR_MAX_BYTES the calling test has set, and returns it with +// the app, for tests that stop the app themselves. +func startApp(t *testing.T) (*reportbuf.Buffer, *fxtest.App) { t.Helper() var buf *reportbuf.Buffer @@ -127,6 +293,16 @@ func startBuffer(t *testing.T) *reportbuf.Buffer { app.RequireStart() t.Cleanup(app.RequireStop) + return buf, app +} + +// startBuffer starts a Buffer through fx, as main does, with the +// DATA_DIR and DATA_DIR_MAX_BYTES the calling test has set. +func startBuffer(t *testing.T) *reportbuf.Buffer { + t.Helper() + + buf, _ := startApp(t) + return buf } @@ -892,6 +1068,32 @@ func TestFlushAtSizeThreshold(t *testing.T) { buf := startBuffer(t) + appended := appendToThreshold(t, buf) + + // Append writes the file in the background, so wait for it. + deadline := time.Now().Add(10 * time.Second) + + for { + files, err := readReportFiles(dir) + if err == nil && len(files) == 1 && files[0] == appended { + return + } + + if time.Now().After(deadline) { + t.Fatalf("%d report files (error: %v), want one holding the "+ + "%d bytes appended", len(files), err, len(appended)) + } + + time.Sleep(10 * time.Millisecond) + } +} + +// appendToThreshold appends reports to buf until it holds +// FlushSizeThreshold bytes, so that the last append starts writing them +// to a report file, and returns the lines appended. +func appendToThreshold(t *testing.T, buf *reportbuf.Buffer) string { + t.Helper() + // Large reports, so the threshold takes a few hundred appends. pad := strings.Repeat("a", 64<<10) @@ -914,22 +1116,7 @@ func TestFlushAtSizeThreshold(t *testing.T) { appended.WriteByte('\n') } - // Append writes the file in the background, so wait for it. - deadline := time.Now().Add(10 * time.Second) - - for { - files, err := readReportFiles(dir) - if err == nil && len(files) == 1 && files[0] == appended.String() { - return - } - - if time.Now().After(deadline) { - t.Fatalf("%d report files (error: %v), want one holding the "+ - "%d bytes appended", len(files), err, appended.Len()) - } - - time.Sleep(10 * time.Millisecond) - } + return appended.String() } // reportFilesBytes returns the total size of the report files in dir. -- 2.54.0