diff --git a/README.md b/README.md index 47a9495..fc993e1 100644 --- a/README.md +++ b/README.md @@ -226,6 +226,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 @@ -244,16 +252,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 diff --git a/TODO.md b/TODO.md index 76023ff..a4101d2 100644 --- a/TODO.md +++ b/TODO.md @@ -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: the repo has the layout `REPO_POLICIES.md` gives ([#30](https://git.eeqj.de/sneak/netwatch/issues/30)). The Go module is at the root (`go.mod`, `go.sum`, `.golangci.yml`, `cmd/netwatch-server/`, diff --git a/internal/handlers/report.go b/internal/handlers/report.go index 8aa5ad9..77dfe1f 100644 --- a/internal/handlers/report.go +++ b/internal/handlers/report.go @@ -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: " + diff --git a/internal/reportbuf/reportbuf.go b/internal/reportbuf/reportbuf.go index 62e6bf4..62feba6 100644 --- a/internal/reportbuf/reportbuf.go +++ b/internal/reportbuf/reportbuf.go @@ -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 diff --git a/internal/reportbuf/reportbuf_test.go b/internal/reportbuf/reportbuf_test.go index 7d679f9..4bbf2a3 100644 --- a/internal/reportbuf/reportbuf_test.go +++ b/internal/reportbuf/reportbuf_test.go @@ -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,212 @@ 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":[]}`), + ) + h.HandleReport().ServeHTTP(rec, req) + + return rec.Code } // TestConcurrentAppendsStopAtCap appends from many goroutines at once