Report file writes run one at a time (closes #150)

Two writes that failed at the same time put their reports back in the
wrong order when the older one failed first. A mutex is now held
across each write, from taking the waiting reports out of the buffer
to putting them back if the write fails, so the next write takes them
in order. The write Append starts at the 10 MiB threshold waits for
one under way, and Append starts no other while it waits. Files
written since the start are now deleted to make room in the order they
were written. A test that needed two writes at once is removed; new
tests cover the one waiting write and the order files are deleted in.

Model: opus-5-5

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")
-	}
-}