diff --git a/README.md b/README.md index 1574245..299664c 100644 --- a/README.md +++ b/README.md @@ -259,13 +259,15 @@ 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. -The page sends the refused samples again in its next report, as long as it still -holds them: at the 1-second interval one refused report already loses some of -those in the history (see [Reporting](#reporting)). 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. +minute, tries again to write them. The server writes one file at a time, so a +file holds its reports in the order they arrived, those of a failed write +included. From the failed write until a write succeeds, `POST /api/v1/reports` +refuses every report with 500 and keeps nothing of it. The page sends the +refused samples again in its next report, as long as it still holds them: at the +1-second interval one refused report already loses some of those in the history +(see [Reporting](#reporting)). 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 diff --git a/TODO.md b/TODO.md index 539dfe2..1884ba2 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-19: report file writes no longer overlap + ([#150](https://git.eeqj.de/sneak/netwatch/issues/150)), so the reports of + failed writes are written in the order they arrived. Two writes that failed at + the same time put their reports back in the wrong order when the older failed + first. One write runs at a time, and each takes the waiting reports only when + it starts, after a failed write has put its own back. The write that 10 MiB + waiting starts waits for one under way, and only one such write waits at a + time. A unit test holds a write open, starts a second at 10 MiB, fails both + and checks the order in the next file. Files written since the start are + deleted to make room in the order they were written, after those found at + start - 2026-10-18: when the backend runs on its own, a `DATA_DIR` that cannot be created or read, or where no file can be created, stops the start with an error naming `DATA_DIR` and its value, as the other settings' errors do diff --git a/internal/reportbuf/reportbuf.go b/internal/reportbuf/reportbuf.go index eadf3c1..3b54eef 100644 --- a/internal/reportbuf/reportbuf.go +++ b/internal/reportbuf/reportbuf.go @@ -16,7 +16,6 @@ import ( "log/slog" "os" "path/filepath" - "slices" "strings" "sync" "sync/atomic" @@ -73,11 +72,9 @@ type Buffer struct { // except in tests that hold the write open or make it fail. fileCreated func(f *os.File) // files are the report files that may be deleted to make room, - // in name order, which is oldest first: those in dataDir at - // start, and each one this buffer writes, put in at its place by - // name once it is complete, even when an older file's write - // completes after a newer one's. A file still being written is - // not among them. filesBytes is their total size. + // oldest first: those in dataDir at start, in name order, then + // each one this buffer writes, once it is complete. A file still + // being written is not among them. filesBytes is their total size. files []reportFile filesBytes int64 log *slog.Logger @@ -90,6 +87,10 @@ type Buffer struct { // millisecond still get different names. seq atomic.Int64 stopOnce sync.Once + // thresholdWriteWaiting is set while a write Append started at + // flushSizeThreshold waits for writeMu; Append starts no other + // while it is set. + thresholdWriteWaiting atomic.Bool // usedBytes is what Append checks against maxBytes: the size // of the report files in dataDir, plus the reports waiting to // be written to one at their uncompressed size. @@ -99,6 +100,11 @@ type Buffer struct { // refuses reports, and the reports of that write wait in buf for // the next flush. writeErr error + // writeMu is held across each write of a report file, so writes + // never overlap: a failed write puts its reports back in buf before + // the next write takes them, which keeps them in the order they + // were appended. + writeMu sync.Mutex // writers are the goroutines that write report files: flushLoop, // and each one Append starts at flushSizeThreshold. OnStop waits // for them before the final flush. @@ -207,8 +213,9 @@ func New( // 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. +// cannot make room. If the buffer reaches the size threshold, it starts +// a write of the reports in the background (see flushAtThreshold), +// unless one it started is still waiting. func (b *Buffer) Append(v any) error { line, err := json.Marshal(v) if err != nil { @@ -238,18 +245,9 @@ func (b *Buffer) Append(v any) error { b.buf.Write(line) b.buf.WriteByte('\n') - if b.buf.Len() >= flushSizeThreshold { - data := b.drainBuf() - b.mu.Unlock() - - b.writers.Go(func() { - writeErr := b.writeFile(data) - if writeErr != nil { - b.log.Error("flush reports failed", "error", writeErr) - } - }) - - return nil + if b.buf.Len() >= flushSizeThreshold && + b.thresholdWriteWaiting.CompareAndSwap(false, true) { + b.writers.Go(b.flushAtThreshold) } b.mu.Unlock() @@ -307,21 +305,28 @@ func (b *Buffer) flushLoop() { } } -// flushLocked acquires the lock, drains the buffer, and -// writes the data to a compressed file. +// flushLocked waits for any write of a report file under way, then +// writes the reports waiting in the buffer to a new one. func (b *Buffer) flushLocked() error { - b.mu.Lock() + b.writeMu.Lock() + defer b.writeMu.Unlock() - if b.buf.Len() == 0 { - b.mu.Unlock() + return b.writeFile() +} - return nil +// flushAtThreshold is the write Append starts when the buffer reaches +// flushSizeThreshold. Once any write under way is done, it lets Append +// start another and writes the reports waiting in the buffer. +func (b *Buffer) flushAtThreshold() { + b.writeMu.Lock() + defer b.writeMu.Unlock() + + b.thresholdWriteWaiting.Store(false) + + err := b.writeFile() + if err != nil { + b.log.Error("flush reports failed", "error", err) } - - data := b.drainBuf() - b.mu.Unlock() - - return b.writeFile(data) } // drainBuf copies the buffer contents and resets it. @@ -334,10 +339,22 @@ func (b *Buffer) drainBuf() []byte { return data } -// writeFile writes data, reports drained from the buffer, to a new -// 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 { +// writeFile takes the reports waiting in the buffer, if any, and writes +// them to a new timestamped zstd-compressed JSONL file in the data +// directory. If the write fails, it puts the reports back in the +// buffer. The caller must hold b.writeMu. +func (b *Buffer) writeFile() error { + b.mu.Lock() + + if b.buf.Len() == 0 { + b.mu.Unlock() + + return nil + } + + data := b.drainBuf() + b.mu.Unlock() + // The timestamp comes first, so the names sort by time; the number // after it tells apart files named in the same millisecond. ts := b.now().UTC().Format("2006-01-02T15-04-05.000Z") @@ -368,14 +385,7 @@ func (b *Buffer) writeFile(data []byte) error { b.usedBytes -= int64(len(data)) b.usedBytes += size b.filesBytes += size - - // At its place by name, not at the end: another write, of a newer - // file, may have completed while this one was being written. - i, _ := slices.BinarySearchFunc(b.files, name, - func(f reportFile, target string) int { - return strings.Compare(f.name, target) - }) - b.files = slices.Insert(b.files, i, reportFile{name: name, size: size}) + b.files = append(b.files, reportFile{name: name, size: size}) return nil } diff --git a/internal/reportbuf/reportbuf_test.go b/internal/reportbuf/reportbuf_test.go index b49bff9..c8015e5 100644 --- a/internal/reportbuf/reportbuf_test.go +++ b/internal/reportbuf/reportbuf_test.go @@ -666,18 +666,23 @@ func TestFileDeletedByHandFreesRoom(t *testing.T) { } } -// TestFileBeingWrittenIsNeverDeleted holds the write of one report file -// open while a second write completes, then sends a report that needs -// room. Deleting either file would make it, and the one being written is -// the older, but only the complete one may be deleted. Once the first -// write is complete, its file is deleted when room is needed. +// TestFileBeingWrittenIsNeverDeleted holds the write of a report file +// open, then sends a report that needs room. The report file from an +// earlier run is deleted to make it, and the one being written is not. +// Once the write is complete, its file is deleted when room is needed. func TestFileBeingWrittenIsNeverDeleted(t *testing.T) { - report := map[string]string{"id": "writing"} + const fileBytes = 100 - t.Setenv("DATA_DIR", t.TempDir()) - // Room for two reports waiting to be written, but not for two - // beside a report file. - t.Setenv("DATA_DIR_MAX_BYTES", strconv.Itoa(2*lineBytes(t, report))) + report := map[string]string{"id": "writing"} + dir := t.TempDir() + earlier := reportFilePath(dir, 1) + + writeBytes(t, earlier, fileBytes) + + t.Setenv("DATA_DIR", dir) + // Room for the earlier file and one report. + t.Setenv("DATA_DIR_MAX_BYTES", + strconv.Itoa(fileBytes+lineBytes(t, report))) buf := startBuffer(t) @@ -701,24 +706,8 @@ func TestFileBeingWrittenIsNeverDeleted(t *testing.T) { writing := <-created - // Only the first write is held; the second goes through, and - // so do the writes after it, the final one at stop included. - var complete string - - buf.OnFileCreated(func(f *os.File) { complete = f.Name() }) - - // Errorf, not Fatalf, until the first write is released, so that a + // Errorf, not Fatalf, until the write is released, so that a // failure here does not leave it held. - err = buf.Append(report) - if err != nil { - t.Errorf("second report: %v", err) - } - - err = buf.Flush() - if err != nil { - t.Errorf("flush of the second report: %v", err) - } - err = buf.Append(report) if err != nil { t.Errorf("report that needs room: %v", err) @@ -728,10 +717,14 @@ func TestFileBeingWrittenIsNeverDeleted(t *testing.T) { t.Error("report file deleted while it was being written") } - if exists(t, complete) { - t.Error("complete report file kept, though room was needed") + if exists(t, earlier) { + t.Error("earlier report file kept, though room was needed") } + // Only this write is held; the writes after it, the final one at + // stop included, go through. + buf.OnFileCreated(func(*os.File) {}) + close(release) err = <-flushed @@ -739,9 +732,13 @@ func TestFileBeingWrittenIsNeverDeleted(t *testing.T) { t.Fatalf("flush of the first report: %v", err) } - err = buf.Append(report) + // As large as the earlier file, so it fits only once the file just + // written is deleted too. + pad := fileBytes - lineBytes(t, map[string]string{"id": ""}) + + err = buf.Append(map[string]string{"id": strings.Repeat("a", pad)}) if err != nil { - t.Fatalf("report after the first write was complete: %v", err) + t.Fatalf("report after the write was complete: %v", err) } if exists(t, writing) { @@ -883,6 +880,82 @@ func TestFailedWriteKeepsReportsAppendedDuringIt(t *testing.T) { } } +// TestFailedWritesKeepReportsInOrder holds a write open, then appends a +// report that takes the buffer to FlushSizeThreshold, so that Append +// starts a second write. The held write fails first, then the second, +// which waited for it and took both reports. The next flush writes them +// in the order they were appended, the held write's first. +func TestFailedWritesKeepReportsInOrder(t *testing.T) { + dir := t.TempDir() + t.Setenv("DATA_DIR", dir) + + buf := startBuffer(t) + + created := make(chan struct{}) + release := make(chan struct{}) + + // Each write is held until release is closed, then fails, as its + // file is closed under it. + buf.OnFileCreated(func(f *os.File) { + created <- struct{}{} + + <-release + + _ = f.Close() + }) + + 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() }() + + <-created + + big := strings.Repeat("a", reportbuf.FlushSizeThreshold) + + // 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": big}) + if err != nil { + t.Errorf("report that starts the second write: %v", err) + } + + close(release) + + err = <-flushed + if err == nil { + t.Error("held write succeeded, though its file was closed under it") + } + + // The second write has created its file, so it has taken the + // reports. + <-created + + // Writes from here on, the final one at stop included, succeed. + buf.OnFileCreated(func(*os.File) {}) + + // Flush waits for the second write to fail before it writes. + err = buf.Flush() + if err != nil { + t.Fatalf("flush after the failed writes: %v", err) + } + + files, err := readReportFiles(dir) + if err != nil { + t.Fatalf("read report files: %v", err) + } + + want := `{"id":"held"}` + "\n" + `{"id":"` + big + `"}` + "\n" + if len(files) != 1 || files[0] != want { + t.Fatalf("%d report files, want one holding the held write's "+ + "report, then the second one", len(files)) + } +} + // 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 @@ -1295,85 +1368,3 @@ func hasReportFile(t *testing.T, dir string) bool { return false } - -// TestFilesDeletedOldestFirstWhenWritesOverlap holds the write of an -// older report file open until a newer one's write completes, then -// releases it. When room is needed, the older file is deleted first, -// though its write was the last to complete. -func TestFilesDeletedOldestFirstWhenWritesOverlap(t *testing.T) { - const maxBytes = 1000 - - report := map[string]string{"id": "overlap"} - - t.Setenv("DATA_DIR", t.TempDir()) - t.Setenv("DATA_DIR_MAX_BYTES", strconv.Itoa(maxBytes)) - - buf := startBuffer(t) - - created := make(chan string) - release := make(chan struct{}) - - buf.OnFileCreated(func(f *os.File) { - created <- f.Name() - - <-release - }) - - err := buf.Append(report) - if err != nil { - t.Fatalf("older report: %v", err) - } - - flushed := make(chan error) - - go func() { flushed <- buf.Flush() }() - - older := <-created - - // Only the older write is held; the newer one goes through, and - // so do the writes after it, the final one at stop included. - var newer string - - buf.OnFileCreated(func(f *os.File) { newer = f.Name() }) - - // Errorf, not Fatalf, until the older write is released, so that a - // failure here does not leave it held. - err = buf.Append(report) - if err != nil { - t.Errorf("newer report: %v", err) - } - - err = buf.Flush() - if err != nil { - t.Errorf("flush of the newer report: %v", err) - } - - close(release) - - err = <-flushed - if err != nil { - t.Fatalf("flush of the older report: %v", err) - } - - info, err := os.Stat(newer) - if err != nil { - t.Fatalf("stat %s: %v", newer, err) - } - - // A report that fits beside the newer file alone, so deleting the - // older one makes exactly the room it needs. - pad := maxBytes - int(info.Size()) - lineBytes(t, map[string]string{"id": ""}) - - err = buf.Append(map[string]string{"id": strings.Repeat("a", pad)}) - if err != nil { - t.Fatalf("report that needs room: %v", err) - } - - if exists(t, older) { - t.Fatal("older report file kept when room was needed") - } - - if !exists(t, newer) { - t.Fatal("newer report file deleted before the older one") - } -}