Reports refused while report files cannot be written (closes #123)
check / check (push) Waiting to run

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.

Model: opus-5-5
This commit is contained in:
2026-10-14 09:12:19 +00:00
parent b1eefe0c29
commit 0675b7b65c
5 changed files with 200 additions and 34 deletions
+8
View File
@@ -22,6 +22,14 @@ Decide whether the repo moves to the layout `REPO_POLICIES.md` gives, with
# Completed Steps # 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
- 2026-10-07: the files shared from `sneak/prompts` are its copies at commit - 2026-10-07: the files shared from `sneak/prompts` are its copies at commit
`dd4027b` ([#113](https://git.eeqj.de/sneak/netwatch/issues/113)), with this `dd4027b` ([#113](https://git.eeqj.de/sneak/netwatch/issues/113)), with this
repository's own entries after the shared content in `.gitignore`, repository's own entries after the shared content in `.gitignore`,
+18 -10
View File
@@ -140,6 +140,14 @@ toward `DATA_DIR_MAX_BYTES` from the next start, only if removing it fails too.
Each file contains one JSON object per line, compressed with zstd. Files are Each file contains one JSON object per line, compressed with zstd. Files are
created with `O_EXCL` to prevent overwrites. 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 ### Report limits
`POST /api/v1/reports` takes reports from anyone who can reach it, without `POST /api/v1/reports` takes reports from anyone who can reach it, without
@@ -158,16 +166,16 @@ credentials, so it is bounded instead. Both refusals below answer with the same
`X-RateLimit-Reset` headers. `X-RateLimit-Reset` headers.
- **Size cap.** The report files in `DATA_DIR` may total at most - **Size cap.** The report files in `DATA_DIR` may total at most
`DATA_DIR_MAX_BYTES`, counting the files already there at start. Reports `DATA_DIR_MAX_BYTES`, counting the files already there at start. Reports
waiting in memory count at their uncompressed size until they are written; 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 those of a failed write included; the part of its file written is removed.
is removed. When a report would take the total past the cap, the oldest report When a report would take the total past the cap, the oldest report files are
files are deleted to make room, and each deletion is logged with the file's deleted to make room, and each deletion is logged with the file's name and
name and size; a file still being written is never deleted. A report is size; a file still being written is never deleted. A report is refused with
refused with 507, and nothing of it is stored, only when the reports waiting 507, and nothing of it is stored, only when the reports waiting to be written
to be written fill the cap on their own, and then no file is deleted. At fill the cap on their own, and then no file is deleted. At start, report files
start, report files past the cap, as after lowering it, are deleted the same past the cap, as after lowering it, are deleted the same way. So the cap is
way. So the cap is how much of the newest reports is kept: the default of 1 how much of the newest reports is kept: the default of 1 GiB is small enough
GiB is small enough for any host; set it to the space you can give `DATA_DIR`. for any host; set it to the space you can give `DATA_DIR`.
### CORS ### CORS
+1 -1
View File
@@ -87,7 +87,7 @@ func (s *Handlers) decodeErrorStatus(err error) int {
// appendErrorStatus logs a failure to store a report and returns // appendErrorStatus logs a failure to store a report and returns
// the status to send: 507 when the reports waiting to be written fill // 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 { func (s *Handlers) appendErrorStatus(err error) int {
if errors.Is(err, reportbuf.ErrFull) { if errors.Is(err, reportbuf.ErrFull) {
s.log.Warn("report refused: " + s.log.Warn("report refused: " +
+33 -11
View File
@@ -91,6 +91,11 @@ type Buffer struct {
// of the report files in dataDir, plus the reports waiting to // of the report files in dataDir, plus the reports waiting to
// be written to one at their uncompressed size. // be written to one at their uncompressed size.
usedBytes int64 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 // 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 // Append marshals v as a single JSON line and appends it to
// the buffer. If the line would take usedBytes past maxBytes, the // the buffer. It stores nothing and returns an error if the last
// oldest report files are deleted to make room; it stores nothing // write of a report file failed, until a write succeeds again. If the
// and returns ErrFull if that cannot make room. If the buffer // line would take usedBytes past maxBytes, the oldest report files are
// reaches the size threshold, it is drained and written to disk // deleted to make room; it stores nothing and returns ErrFull if that
// asynchronously. // 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 { func (b *Buffer) Append(v any) error {
line, err := json.Marshal(v) line, err := json.Marshal(v)
if err != nil { if err != nil {
@@ -182,6 +188,13 @@ func (b *Buffer) Append(v any) error {
b.mu.Lock() 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) b.deleteOldestFiles(lineBytes)
if b.usedBytes+lineBytes > b.maxBytes { 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 // 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 { func (b *Buffer) writeFile(data []byte) error {
// The timestamp comes first, so the names sort by time; the number // The timestamp comes first, so the names sort by time; the number
// after it tells apart files named in the same millisecond. // 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) 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() b.mu.Lock()
defer b.mu.Unlock() defer b.mu.Unlock()
b.usedBytes -= int64(len(data)) b.writeErr = err
if err != nil { 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 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.usedBytes += size
b.filesBytes += size b.filesBytes += size
+140 -12
View File
@@ -5,6 +5,8 @@ import (
"errors" "errors"
"fmt" "fmt"
"io/fs" "io/fs"
"net/http"
"net/http/httptest"
"os" "os"
"path/filepath" "path/filepath"
"slices" "slices"
@@ -17,6 +19,8 @@ import (
"sneak.berlin/go/netwatch/internal/config" "sneak.berlin/go/netwatch/internal/config"
"sneak.berlin/go/netwatch/internal/globals" "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/logger"
"sneak.berlin/go/netwatch/internal/reportbuf" "sneak.berlin/go/netwatch/internal/reportbuf"
@@ -507,14 +511,13 @@ func TestFileBeingWrittenIsNeverDeleted(t *testing.T) {
} }
} }
// TestFailedWriteStopsCounting makes a write fail once its file is // TestFailedWriteKeepsItsReports makes a write fail once its file is
// created. Its reports are lost, so they stop counting, and the part of // created. The part of the file written is removed, so it takes no
// the file written is removed, so it takes no room. // room, and the report waits in memory for the next flush, which
func TestFailedWriteStopsCounting(t *testing.T) { // writes it.
report := map[string]string{"id": "failed"} func TestFailedWriteKeepsItsReports(t *testing.T) {
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) buf := startBuffer(t)
@@ -527,9 +530,9 @@ func TestFailedWriteStopsCounting(t *testing.T) {
_ = f.Close() _ = f.Close()
}) })
err := buf.Append(report) err := buf.Append(map[string]string{"id": "kept"})
if err != nil { if err != nil {
t.Fatalf("report that fills the cap exactly: %v", err) t.Fatalf("append report: %v", err)
} }
err = buf.Flush() err = buf.Flush()
@@ -544,10 +547,135 @@ func TestFailedWriteStopsCounting(t *testing.T) {
// Writes from here on, the final one at stop included, succeed. // Writes from here on, the final one at stop included, succeed.
buf.OnFileCreated(func(*os.File) {}) buf.OnFileCreated(func(*os.File) {})
err = buf.Append(report) err = buf.Flush()
if err != nil { if err != nil {
t.Fatalf("report after the failed write: %v", err) 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)
}
}
// 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 // TestConcurrentAppendsStopAtCap appends from many goroutines at once