Reports refused while report files cannot be written (closes #123 )

A failed write of a report file used to drop its reports, and POST /api/v1/reports kept answering 200 for as long as writes failed. Now a failed write puts its reports back in the buffer, still counted against DATA_DIR_MAX_BYTES, and Append refuses every report, which the handler answers 500, until a write succeeds. The minute flush tries the held reports again, so storing resumes with no restart once writing works, and the buffer takes nothing new while writes fail. Tests also cover a report appended while a write that fails runs, and the cap once a failed write's reports are written. Model: opus-5-5
2026-10-14 10:41:02 +00:00
5 changed files with 281 additions and 29 deletions
@@ -228,6 +228,14 @@ the numbers: its file, if it was created, is removed. The file stays, counted
 toward `DATA_DIR_MAX_BYTES` from the next start, only if removing it fails too.
 Files are created with `O_EXCL` to prevent overwrites.

+The reports of a failed write stay in memory, and the next flush, within a
+minute, tries again to write them. From the failed write until a write succeeds,
+`POST /api/v1/reports` refuses every report with 500 and keeps nothing of it, so
+the page keeps its samples and sends them again at its next report. Reports
+already answered 200 are lost only if the server stops before a write succeeds;
+the final flush at stop tries them too, and if it fails, the process exits
+non-zero.
+
 #### Report limits

 `POST /api/v1/reports` takes reports from anyone who can reach it, without
@@ -246,16 +254,16 @@ credentials, so it is bounded instead. Both refusals below answer with the same
  `X-RateLimit-Reset` headers.
 - **Size cap.** The report files in `DATA_DIR` may total at most
  `DATA_DIR_MAX_BYTES`, counting the files already there at start. Reports
-  waiting in memory count at their uncompressed size until they are written;
-  those lost to a failed write stop counting, and the part of its file written
-  is removed. When a report would take the total past the cap, the oldest report
-  files are deleted to make room, and each deletion is logged with the file's
-  name and size; a file still being written is never deleted. A report is
-  refused with 507, and nothing of it is stored, only when the reports waiting
-  to be written fill the cap on their own, and then no file is deleted. At
-  start, report files past the cap, as after lowering it, are deleted the same
-  way. So the cap is how much of the newest reports is kept: the default of 1
-  GiB is small enough for any host; set it to the space you can give `DATA_DIR`.
+  waiting in memory count at their uncompressed size until they are written,
+  those of a failed write included; the part of its file written is removed.
+  When a report would take the total past the cap, the oldest report files are
+  deleted to make room, and each deletion is logged with the file's name and
+  size; a file still being written is never deleted. A report is refused with
+  507, and nothing of it is stored, only when the reports waiting to be written
+  fill the cap on their own, and then no file is deleted. At start, report files
+  past the cap, as after lowering it, are deleted the same way. So the cap is
+  how much of the newest reports is kept: the default of 1 GiB is small enough
+  for any host; set it to the space you can give `DATA_DIR`.

 #### CORS

@@ -21,6 +21,17 @@ Run `make frontend-viewport-test` in CI as its own step; it is not part of

 # Completed Steps

+- 2026-10-14: while report files cannot be written, `POST /api/v1/reports`
+  answers 500, and storing recovers once writing works again, with no restart
+  ([#123](https://git.eeqj.de/sneak/netwatch/issues/123)). A failed write puts
+  its reports back in the report buffer, still counted against
+  `DATA_DIR_MAX_BYTES`, where they used to be lost; the next flush, within a
+  minute, tries them again. Until a write succeeds, every report is refused and
+  nothing of it is kept, so the buffer does not grow while writes fail. A test
+  posts reports through the report handler before, during and after a failure;
+  others check that a report appended while a write runs is kept, behind that
+  write's own, when the write fails, and that the cap still holds exactly once a
+  failed write's reports are written
 - 2026-10-14: changing the interval no longer drops the samples not yet reported
  to the backend ([#124](https://git.eeqj.de/sneak/netwatch/issues/124)). The
  interval menu resets each host with the new `HostState.reset`, which moves the
@@ -87,7 +87,7 @@ func (s *Handlers) decodeErrorStatus(err error) int {

 // appendErrorStatus logs a failure to store a report and returns
 // the status to send: 507 when the reports waiting to be written fill
-// the size cap, otherwise 500.
+// the size cap, otherwise 500, as when report files cannot be written.
 func (s *Handlers) appendErrorStatus(err error) int {
 	if errors.Is(err, reportbuf.ErrFull) {
 		s.log.Warn("report refused: " +
@@ -91,6 +91,11 @@ type Buffer struct {
 	// of the report files in dataDir, plus the reports waiting to
 	// be written to one at their uncompressed size.
 	usedBytes int64
+	// writeErr is the error of the last write of a report file to
+	// finish, nil if it succeeded or none has. While it is set, Append
+	// refuses reports, and the reports of that write wait in buf for
+	// the next flush.
+	writeErr error
 }

 // New creates a Buffer and registers lifecycle hooks to
@@ -167,11 +172,12 @@ func New(
 }

 // Append marshals v as a single JSON line and appends it to
-// the buffer. If the line would take usedBytes past maxBytes, the
-// oldest report files are deleted to make room; it stores nothing
-// and returns ErrFull if that cannot make room. If the buffer
-// reaches the size threshold, it is drained and written to disk
-// asynchronously.
+// the buffer. It stores nothing and returns an error if the last
+// write of a report file failed, until a write succeeds again. If the
+// line would take usedBytes past maxBytes, the oldest report files are
+// deleted to make room; it stores nothing and returns ErrFull if that
+// cannot make room. If the buffer reaches the size threshold, it is
+// drained and written to disk asynchronously.
 func (b *Buffer) Append(v any) error {
 	line, err := json.Marshal(v)
 	if err != nil {
@@ -182,6 +188,13 @@ func (b *Buffer) Append(v any) error {

 	b.mu.Lock()

+	if b.writeErr != nil {
+		err = fmt.Errorf("report files cannot be written: %w", b.writeErr)
+		b.mu.Unlock()
+
+		return err
+	}
+
 	b.deleteOldestFiles(lineBytes)

 	if b.usedBytes+lineBytes > b.maxBytes {
@@ -291,7 +304,8 @@ func (b *Buffer) drainBuf() []byte {
 }

 // writeFile writes data, reports drained from the buffer, to a new
-// timestamped zstd-compressed JSONL file in the data directory.
+// timestamped zstd-compressed JSONL file in the data directory. If the
+// write fails, it puts the reports back in the buffer.
 func (b *Buffer) writeFile(data []byte) error {
 	// The timestamp comes first, so the names sort by time; the number
 	// after it tells apart files named in the same millisecond.
@@ -300,19 +314,27 @@ func (b *Buffer) writeFile(data []byte) error {

 	size, err := b.createFile(filepath.Join(b.dataDir, name), data)

-	// The reports no longer wait to be written, so they stop counting
-	// at their uncompressed size. If the write failed they are lost;
-	// otherwise they count as the file, which from here on may be
-	// deleted to make room.
 	b.mu.Lock()
 	defer b.mu.Unlock()

-	b.usedBytes -= int64(len(data))
+	b.writeErr = err

 	if err != nil {
+		// The reports go back in front of any appended since, for the
+		// next flush to write, and still count at their uncompressed
+		// size.
+		appended := bytes.Clone(b.buf.Bytes())
+		b.buf.Reset()
+		b.buf.Write(data)
+		b.buf.Write(appended)
+
 		return err
 	}

+	// The reports no longer wait to be written, so they stop counting
+	// at their uncompressed size and count as the file, which from here
+	// on may be deleted to make room.
+	b.usedBytes -= int64(len(data))
 	b.usedBytes += size
 	b.filesBytes += size

@@ -5,6 +5,8 @@ import (
 	"errors"
 	"fmt"
 	"io/fs"
+	"net/http"
+	"net/http/httptest"
 	"os"
 	"path/filepath"
 	"slices"
@@ -17,6 +19,8 @@ import (

 	"sneak.berlin/go/netwatch/internal/config"
 	"sneak.berlin/go/netwatch/internal/globals"
+	"sneak.berlin/go/netwatch/internal/handlers"
+	"sneak.berlin/go/netwatch/internal/healthcheck"
 	"sneak.berlin/go/netwatch/internal/logger"
 	"sneak.berlin/go/netwatch/internal/reportbuf"

@@ -507,13 +511,17 @@ func TestFileBeingWrittenIsNeverDeleted(t *testing.T) {
 	}
 }

-// TestFailedWriteStopsCounting makes a write fail once its file is
-// created. Its reports are lost, so they stop counting, and the part of
-// the file written is removed, so it takes no room.
-func TestFailedWriteStopsCounting(t *testing.T) {
-	report := map[string]string{"id": "failed"}
+// TestFailedWriteKeepsItsReports makes a write fail once its file is
+// created, under a cap with room for one report. The part of the file
+// written is removed, so it takes no room, and the report waits in
+// memory for the next flush, which writes it. Once written, it counts
+// as its file and no longer as a report waiting, so the cap still holds
+// exactly: one more report is taken and the one after it is refused.
+func TestFailedWriteKeepsItsReports(t *testing.T) {
+	report := map[string]string{"id": "kept"}
+	dir := t.TempDir()

-	t.Setenv("DATA_DIR", t.TempDir())
+	t.Setenv("DATA_DIR", dir)
 	t.Setenv("DATA_DIR_MAX_BYTES", strconv.Itoa(lineBytes(t, report)))

 	buf := startBuffer(t)
@@ -544,10 +552,213 @@ func TestFailedWriteStopsCounting(t *testing.T) {
 	// Writes from here on, the final one at stop included, succeed.
 	buf.OnFileCreated(func(*os.File) {})

+	err = buf.Flush()
+	if err != nil {
+		t.Fatalf("flush after the failed write: %v", err)
+	}
+
+	files, err := readReportFiles(dir)
+	if err != nil {
+		t.Fatalf("read report files: %v", err)
+	}
+
+	want := `{"id":"kept"}` + "\n"
+	if len(files) != 1 || files[0] != want {
+		t.Fatalf("report files = %q, want one holding %q", files, want)
+	}
+
+	// The file is deleted to make room if need be.
 	err = buf.Append(report)
 	if err != nil {
-		t.Fatalf("report after the failed write: %v", err)
+		t.Fatalf("report after the written one: %v", err)
 	}
+
+	err = buf.Append(report)
+	if !errors.Is(err, reportbuf.ErrFull) {
+		t.Fatalf("report past the cap: error = %v, want ErrFull", err)
+	}
+}
+
+// TestFailedWriteKeepsReportsAppendedDuringIt holds a write open,
+// appends a report while it is held, then makes the held write fail.
+// Both reports are kept, and the next flush writes them in the order
+// they were appended, the held write's first.
+func TestFailedWriteKeepsReportsAppendedDuringIt(t *testing.T) {
+	dir := t.TempDir()
+	t.Setenv("DATA_DIR", dir)
+
+	buf := startBuffer(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": "held"})
+	if err != nil {
+		t.Fatalf("report of the held write: %v", err)
+	}
+
+	flushed := make(chan error)
+
+	go func() { flushed <- buf.Flush() }()
+
+	held := <-created
+
+	// Errorf, not Fatalf, until the held write is released, so that a
+	// failure here does not leave it held.
+	err = buf.Append(map[string]string{"id": "appended"})
+	if err != nil {
+		t.Errorf("report appended while the write was held: %v", err)
+	}
+
+	// Closing the file under the write makes the write fail.
+	_ = held.Close()
+
+	close(release)
+
+	err = <-flushed
+	if err == nil {
+		t.Fatal("held write succeeded, though its file was closed under it")
+	}
+
+	// Writes from here on, the final one at stop included, succeed.
+	buf.OnFileCreated(func(*os.File) {})
+
+	err = buf.Flush()
+	if err != nil {
+		t.Fatalf("flush after the failed write: %v", err)
+	}
+
+	files, err := readReportFiles(dir)
+	if err != nil {
+		t.Fatalf("read report files: %v", err)
+	}
+
+	want := `{"id":"held"}` + "\n" + `{"id":"appended"}` + "\n"
+	if len(files) != 1 || files[0] != want {
+		t.Fatalf("report files = %q, want one holding %q", files, want)
+	}
+}
+
+// TestReportsRefusedWhileWritesFail posts reports through the report
+// handler, as the page does. Once a write has failed, a report is
+// refused with 500 until a write succeeds again; the reports taken
+// before and after are stored, and the one refused is not.
+func TestReportsRefusedWhileWritesFail(t *testing.T) {
+	dir := t.TempDir()
+	t.Setenv("DATA_DIR", dir)
+
+	buf, h := startBufferWithHandlers(t)
+
+	if code := postReport(t, h, "before"); code != http.StatusOK {
+		t.Fatalf("report before the failed write: status = %d, want %d",
+			code, http.StatusOK)
+	}
+
+	// With the data directory gone, the write has nowhere to go.
+	err := os.Remove(dir)
+	if err != nil {
+		t.Fatalf("remove data dir: %v", err)
+	}
+
+	err = buf.Flush()
+	if err == nil {
+		t.Fatal("flush succeeded with the data directory gone")
+	}
+
+	if code := postReport(t, h, "refused"); code != http.StatusInternalServerError {
+		t.Fatalf("report after the failed write: status = %d, want %d",
+			code, http.StatusInternalServerError)
+	}
+
+	err = os.Mkdir(dir, 0o750)
+	if err != nil {
+		t.Fatalf("make data dir: %v", err)
+	}
+
+	err = buf.Flush()
+	if err != nil {
+		t.Fatalf("flush with the data directory back: %v", err)
+	}
+
+	if code := postReport(t, h, "after"); code != http.StatusOK {
+		t.Fatalf("report after a write succeeded: status = %d, want %d",
+			code, http.StatusOK)
+	}
+
+	err = buf.Flush()
+	if err != nil {
+		t.Fatalf("flush of the report after: %v", err)
+	}
+
+	files, err := readReportFiles(dir)
+	if err != nil {
+		t.Fatalf("read report files: %v", err)
+	}
+
+	stored := strings.Join(files, "")
+
+	for _, clientID := range []string{"before", "after"} {
+		if !strings.Contains(stored, `"clientId":"`+clientID+`"`) {
+			t.Errorf("report %q not stored", clientID)
+		}
+	}
+
+	if strings.Contains(stored, `"clientId":"refused"`) {
+		t.Error("refused report stored")
+	}
+}
+
+// startBufferWithHandlers starts a Buffer and the Handlers that store
+// reports in it through fx, as main does, with the DATA_DIR the calling
+// test has set.
+func startBufferWithHandlers(
+	t *testing.T,
+) (*reportbuf.Buffer, *handlers.Handlers) {
+	t.Helper()
+
+	var (
+		buf *reportbuf.Buffer
+		h   *handlers.Handlers
+	)
+
+	app := fxtest.New(t,
+		fx.Provide(
+			config.New,
+			globals.New,
+			handlers.New,
+			healthcheck.New,
+			logger.New,
+			reportbuf.New,
+		),
+		fx.Populate(&buf, &h),
+	)
+
+	app.RequireStart()
+	t.Cleanup(app.RequireStop)
+
+	return buf, h
+}
+
+// postReport posts a report from clientID to the report handler, as the
+// page does, and returns the status of the answer.
+func postReport(t *testing.T, h *handlers.Handlers, clientID string) int {
+	t.Helper()
+
+	rec := httptest.NewRecorder()
+	req := httptest.NewRequestWithContext(t.Context(),
+		http.MethodPost, "/api/v1/reports",
+		strings.NewReader(`{"clientId":"`+clientID+`","hosts":[]}`),
+	)
+	req.Header.Set("Content-Type", "application/json")
+	h.HandleReport().ServeHTTP(rec, req)
+
+	return rec.Code
 }

 // TestConcurrentAppendsStopAtCap appends from many goroutines at once