README.md

@@ -271,13 +271,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
 

TODO.md

@@ -21,6 +21,19 @@ 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 `Append`
+  starts when 10 MiB are waiting waits for any write under way, and `Append`
+  starts no other while it waits. Files written since the start are deleted to
+  make room in the order they were written, after those found at start. A unit
+  test holds a write open, starts a second at 10 MiB, fails both and checks the
+  order in the next file. Others check that `Append` starts no second write at
+  10 MiB while one waits, that it starts one again the next time 10 MiB are
+  waiting, and that the first file written since the start is the first deleted
 - 2026-10-19: `DEBUG` unset or empty leaves debug logging off, and any other
   value, `false` and `0` included, turns it on, as the code style guide says
   ([#159](https://git.eeqj.de/sneak/netwatch/issues/159)). No value of it stops

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
 }

internal/reportbuf/reportbuf_test.go

@@ -581,6 +581,62 @@ func TestWrittenFilesDeletedToMakeRoom(t *testing.T) {
 	}
 }
 
+// TestWrittenFilesDeletedOldestFirst writes three report files, then
+// appends a report that fits only once one of them is deleted. The
+// first one written must be the one deleted.
+func TestWrittenFilesDeletedOldestFirst(t *testing.T) {
+	const maxBytes = 1000
+
+	report := map[string]string{"id": "written"}
+
+	t.Setenv("DATA_DIR", t.TempDir())
+	t.Setenv("DATA_DIR_MAX_BYTES", strconv.Itoa(maxBytes))
+
+	buf := startBuffer(t)
+
+	var written []string
+
+	buf.OnFileCreated(func(f *os.File) { written = append(written, f.Name()) })
+
+	for range 3 {
+		err := buf.Append(report)
+		if err != nil {
+			t.Fatalf("append report: %v", err)
+		}
+
+		err = buf.Flush()
+		if err != nil {
+			t.Fatalf("flush: %v", err)
+		}
+	}
+
+	// The three files hold the same report, so they are the same size.
+	info, err := os.Stat(written[0])
+	if err != nil {
+		t.Fatalf("stat %s: %v", written[0], err)
+	}
+
+	// A report that fits beside two of the files, so deleting one makes
+	// exactly the room it needs.
+	pad := maxBytes - 2*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, written[0]) {
+		t.Fatal("first report file written kept when room was needed")
+	}
+
+	for _, path := range written[1:] {
+		if !exists(t, path) {
+			t.Fatalf("%s deleted before the first report file written", path)
+		}
+	}
+}
+
 // TestFailedDeletionStillCounts makes deleting the oldest report file
 // fail. It is still there, so it still takes room, and the next oldest
 // is deleted in its place.
@@ -666,17 +722,16 @@ 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 under a cap with room for two reports, then appends a second
+// report and a third that needs room. The file being written is the
+// only report file, and it is not deleted to make room, so the third
+// report is refused. Once the write is complete, its file is deleted
+// when room is needed.
 func TestFileBeingWrittenIsNeverDeleted(t *testing.T) {
 	report := map[string]string{"id": "writing"}
 
 	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)))
 
 	buf := startBuffer(t)
@@ -701,36 +756,25 @@ 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)
+	if !errors.Is(err, reportbuf.ErrFull) {
+		t.Errorf("report that needs room: error = %v, want ErrFull", err)
 	}
 
 	if !exists(t, writing) {
 		t.Error("report file deleted while it was being written")
 	}
 
-	if exists(t, complete) {
-		t.Error("complete 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)
 
@@ -739,9 +783,11 @@ func TestFileBeingWrittenIsNeverDeleted(t *testing.T) {
 		t.Fatalf("flush of the first report: %v", err)
 	}
 
+	// The second report still waits to be written, so this one fits
+	// only once the file just written is deleted.
 	err = buf.Append(report)
 	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 +929,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
@@ -1150,6 +1272,106 @@ func TestFlushAtSizeThreshold(t *testing.T) {
 	}
 }
 
+// TestEachSizeThresholdStartsAWrite reaches FlushSizeThreshold twice,
+// with no call to Flush and the periodic flush a minute away. Append
+// must start a write each time, so that the reports end up in two
+// report files. The test runs in a synctest bubble, so that
+// synctest.Wait returns once each write is done.
+func TestEachSizeThresholdStartsAWrite(t *testing.T) {
+	dir := t.TempDir()
+	t.Setenv("DATA_DIR", dir)
+
+	synctest.Test(t, func(t *testing.T) {
+		buf := startBuffer(t)
+
+		var appended string
+
+		for range 2 {
+			appended = appendToThreshold(t, buf)
+
+			synctest.Wait()
+		}
+
+		files, err := readReportFiles(dir)
+		if err != nil {
+			t.Fatalf("read report files: %v", err)
+		}
+
+		if !slices.Equal(files, []string{appended, appended}) {
+			t.Fatalf("%d report files, want two, each holding the %d "+
+				"bytes appended to reach the threshold",
+				len(files), len(appended))
+		}
+	})
+}
+
+// TestAppendStartsOneWaitingWrite holds a write open, then appends two
+// reports of FlushSizeThreshold bytes each. Append starts a write at
+// the first, which waits for the held one, and none at the second,
+// since that write is still waiting. Every write fails, so a write
+// started at the second report would find the reports back in the
+// buffer and try them again: once the held write is released, two
+// writes must have been started, the held one and one more.
+func TestAppendStartsOneWaitingWrite(t *testing.T) {
+	t.Setenv("DATA_DIR", t.TempDir())
+
+	synctest.Test(t, func(t *testing.T) {
+		buf := startBuffer(t)
+
+		var writes atomic.Int64
+
+		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) {
+			writes.Add(1)
+
+			<-release
+
+			_ = f.Close()
+		})
+
+		err := buf.Append(map[string]string{"id": "one-waiting"})
+		if err != nil {
+			t.Fatalf("report of the held write: %v", err)
+		}
+
+		go func() { _ = buf.Flush() }()
+
+		// Wait returns once the flush's write is held.
+		synctest.Wait()
+
+		big := map[string]string{
+			"id": strings.Repeat("a", reportbuf.FlushSizeThreshold),
+		}
+
+		// Errorf, not Fatalf, until the held write is released, so that a
+		// failure here does not leave it held.
+		for range 2 {
+			err = buf.Append(big)
+			if err != nil {
+				t.Errorf("report of FlushSizeThreshold bytes: %v", err)
+			}
+		}
+
+		close(release)
+
+		// Wait returns once every write started is done.
+		synctest.Wait()
+
+		started := writes.Load()
+
+		// Writes from here on, the final one at stop included, succeed.
+		buf.OnFileCreated(func(*os.File) {})
+
+		if started != 2 {
+			t.Fatalf("%d writes started, want 2: the held one and one "+
+				"at the threshold", started)
+		}
+	})
+}
+
 // 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.
@@ -1295,85 +1517,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")
-	}
-}