outbox: a Send compacts the file to the rows it still holds

docs/changes/unreleased/1133-the-outbox-file-is-compacted.md

@@ -0,0 +1,14 @@
+---
+kind: fixed
+title: the outbox file is compacted back to the rows it still holds
+pr: 1133
+surface: [engine]
+invalidates:
+  - "The outbox file at `<profile>/pool/outbox.jsonl` was append-only with nothing that ever rewrote it, so it carried a line for every row an install ever judged — the row, and later the marker that retired it — and grew without bound. Every `Send` and every `Pending` read the whole file back, so a long-lived install's work per judgement grew with the total it had ever judged. A `Send` now rewrites the file to exactly the rows still pending when it has grown mostly into retired ones — more lines than twice the pending rows, past a floor of 1024 lines — through a temporary file fsynced and renamed over the original, and the markers of the rows that are gone go with them. A crash before the rename leaves the old file whole and one after it leaves a file holding the same pending rows, so no row is lost and no retired row comes back to be sent twice. A second codeaf on the same profile that already holds the file keeps writing to the inode the rename unlinked and those appends are lost — the one-writer-per-profile assumption the file already rested on, now made explicit."
+---
+
+The file is still append-only between compactions, which is what keeps a crash
+mid-write from losing or double-sending a row; the rewrite happens only where
+the file's own size is the cost, at the start of a `Send`, before its first POST
+and under the same lock that serialises Sends. What stays is only the pending
+rows, because a marker means nothing once the row it retired is gone.

internal/pool/outbox/outbox.go

@@ -11,6 +11,13 @@
 // is what keeps the file append-only while a cap on pending rows still drops
 // from the old end and a mark outlives closing the outbox and opening the same
 // path again.
+//
+// Append-only also means the file grows a line for every row ever judged, and
+// a long-lived run would read the whole of that back on every Send and every
+// Pending. So a Send whose file has grown mostly into rows it can no longer
+// need rewrites it to the pending rows alone — through a temporary file in the
+// same directory, never in place — and the markers for rows that are gone go
+// with the rows.
 package outbox
 
 import (
@@ -35,6 +42,12 @@ import (
 // or below.
 const DefaultMaxPending = 5000
 
+// compactFloor is the number of lines below which a Send does not bother to
+// rewrite the file: a handful of lines costs less to read than to copy, and
+// the markers of a run that has judged a few thousand rows are what a rewrite
+// is worth. See [Outbox.compactLocked] for what a rewrite does and when.
+const compactFloor = 1024
+
 const (
 	rowSchema  = 1
 	nonceLen   = 16
@@ -116,7 +129,7 @@ func Open(path string) (*Outbox, error) {
 	o := &Outbox{path: path, f: f}
 	o.mu.Lock()
 	defer o.mu.Unlock()
-	rows := o.load()
+	rows, _ := o.load()
 	o.pend = make([]string, 0, len(rows))
 	for _, r := range rows {
 		o.pend = append(o.pend, r.Nonce)
@@ -184,7 +197,8 @@ func (o *Outbox) Append(payload json.RawMessage) error {
 func (o *Outbox) Pending() []Row {
 	o.mu.Lock()
 	defer o.mu.Unlock()
-	return o.load()
+	rows, _ := o.load()
+	return rows
 }
 
 // Send hands the pending rows to dest and answers with how many rows it
@@ -218,12 +232,24 @@ func (o *Outbox) Send(ctx context.Context, dest string) (int, error) {
 		o.mu.Unlock()
 		return 0, errors.New("outbox: outbox is closed")
 	}
-	rows := o.load()
+	rows, lines := o.load()
 	pend := make([]string, len(rows))
 	for i, r := range rows {
 		pend[i] = r.Nonce
 	}
 	o.pend = pend
+	// The file carried every row the outbox ever judged, so a run that sends
+	// often grows it without bound and every Send reads the whole of it. Send is
+	// where that is worth undoing: it is the read that would pay for the whole
+	// file, sendMu holds every other Send out while this one runs its first
+	// POST, and the file is rewritten only once it is mostly dead — more lines
+	// than twice the rows still pending — so each rewrite is paid for by the
+	// rows it drops and a run's own path meets one at most once per floor's
+	// worth of rows. A rewrite that fails is not a failed Send: the rows stay
+	// pending and the next Send tries again.
+	if lines > compactFloor && lines > 2*len(pend) {
+		_ = o.compactLocked(rows)
+	}
 	o.mu.Unlock()
 
 	if len(rows) == 0 {
@@ -372,22 +398,26 @@ func (o *Outbox) dropPendingLocked(marks []mark) {
 }
 
 // load reads the outbox file and answers the rows still pending, in file
-// order. A line that does not parse, or that is not a row, is skipped, and a
-// file that cannot be read answers as no rows at all. The caller holds mu.
+// order, and the number of lines the file held — rows, markers and lines that
+// parsed as neither. A line that does not parse, or that is not a row, is
+// skipped, and a file that cannot be read answers as no rows and no lines at
+// all. The caller holds mu.
 //
 // EACH MARKER RETIRES ONE ROW, oldest first, wherever the marker sits — see
 // [Outbox.dropPendingLocked] for why the count rather than the name: a nonce
 // is what the caller's own [Outbox.Rand] made of sixteen bytes, and a reader
 // that took a marker as a name would answer that a file holding two rows under
 // one nonce and one marker holds nothing at all.
-func (o *Outbox) load() []Row {
+func (o *Outbox) load() ([]Row, int) {
 	b, err := os.ReadFile(o.path)
 	if err != nil {
-		return nil
+		return nil, 0
 	}
 	var rows []Row
 	retired := make(map[string]int)
+	lines := 0
 	for len(b) > 0 {
+		lines++
 		line := b
 		if i := bytes.IndexByte(b, '\n'); i >= 0 {
 			line, b = b[:i], b[i+1:]
@@ -422,7 +452,68 @@ func (o *Outbox) load() []Row {
 		}
 		kept = append(kept, r)
 	}
-	return kept
+	return kept, lines
+}
+
+// compactLocked rewrites the outbox file to hold exactly the pending rows,
+// dropping every marker and every row a marker retired with them — a marker
+// means nothing once its row is gone. It writes a temporary file in the same
+// directory, fsyncs it, and renames it over the original before reopening the
+// handle, never truncating in place: a crash before the rename leaves the old
+// file whole, and a crash after it leaves a file holding the same pending
+// rows, so a row is never lost and a retired row never comes back to be sent
+// twice. The caller holds mu, and Send calls it before its first POST, so no
+// Send of this process is between a POST and its markers. Its error is Send's
+// to ignore — the file is either untouched or already compacted.
+//
+// A codeaf that already holds the old file keeps writing to the inode the
+// rename unlinked, and those appends are lost. The profile is one writer's to
+// hold across processes, which is the assumption the append-only file already
+// rests on; nothing here reaches past it.
+func (o *Outbox) compactLocked(rows []Row) error {
+	tmp, err := os.CreateTemp(filepath.Dir(o.path), filepath.Base(o.path)+".compact-")
+	if err != nil {
+		return err
+	}
+	tmpName := tmp.Name()
+	renamed := false
+	defer func() {
+		if !renamed {
+			tmp.Close()
+			os.Remove(tmpName)
+		}
+	}()
+	for _, r := range rows {
+		line, err := rowLine(r)
+		if err != nil {
+			return err
+		}
+		if _, err := tmp.Write(line); err != nil {
+			return err
+		}
+	}
+	if err := tmp.Sync(); err != nil {
+		return err
+	}
+	if err := tmp.Close(); err != nil {
+		return err
+	}
+	if err := os.Rename(tmpName, o.path); err != nil {
+		return err
+	}
+	renamed = true
+	nf, err := os.OpenFile(o.path, os.O_CREATE|os.O_WRONLY|os.O_APPEND, fileMode)
+	if err != nil {
+		// The file on disk is the compacted one and holds every pending row, so
+		// the work is done; only the handle cannot be had. Refuse writes rather
+		// than write to the inode the rename unlinked.
+		o.f.Close()
+		o.f = nil
+		return err
+	}
+	o.f.Close()
+	o.f = nf
+	return nil
 }
 
 // sendHTTP posts the rows to dest, at most maxBatch per POST, and marks each

internal/pool/outbox/outbox_test.go

@@ -1191,6 +1191,96 @@ func TestASendWhoseMarkerCannotBeWrittenKeepsTheRowsPending(t *testing.T) {
 	}
 }
 
+// seedRetiredRows writes an outbox file holding retired rows each followed by
+// its sent marker, two pending rows and a torn last line, and answers the two
+// pending row lines in file order. The count sits well above the floor at
+// which a Send rewrites the file, so the trigger is certain.
+func seedRetiredRows(t *testing.T, path string, retired int) (string, string) {
+	t.Helper()
+	nonce := func(n int) string { return fmt.Sprintf("%032x", n) }
+	var lines []string
+	for i := 0; i < retired; i++ {
+		lines = append(lines,
+			`{"schema":1,"day":"2026-09-16","nonce":"`+nonce(i)+`","payload":{"k":`+fmt.Sprintf("%d", i)+`}}`,
+			`{"sent":"`+nonce(i)+`"}`)
+	}
+	const base = 1 << 20
+	keep1 := `{"schema":1,"day":"2026-09-17","nonce":"` + nonce(base) + `","payload":{"k":"keep one"}}`
+	keep2 := `{"schema":1,"day":"2026-09-17","nonce":"` + nonce(base+1) + `","payload":{"k":"keep two"}}`
+	torn := `{"schema":1,"day":"2026-09-17","nonce":"` + nonce(base+2) + `","payload":{"k":"torn"`
+	lines = append(lines, keep1, keep2, torn)
+	if err := os.WriteFile(path, []byte(strings.Join(lines, "\n")), 0600); err != nil {
+		t.Fatal(err)
+	}
+	return keep1, keep2
+}
+
+// A file that has grown mostly into rows a marker retired is rewritten by a
+// Send down to the rows still pending: the markers and the retired rows are
+// gone, a torn last line is dropped rather than kept, and Pending still
+// answers the rows in order.
+func TestSendCompactsAFileGrownMostlyIntoRetiredRows(t *testing.T) {
+	path := filepath.Join(t.TempDir(), "outbox.jsonl")
+	keep1, keep2 := seedRetiredRows(t, path, 1500)
+	o, err := Open(path)
+	if err != nil {
+		t.Fatalf("Open: %v", err)
+	}
+	defer o.Close()
+
+	// A destination that refuses reaches the seam — a Send compacts before its
+	// first POST — and leaves every row pending for this test to read back.
+	if _, err := o.Send(context.Background(), "http://127.0.0.1:1/"); err == nil {
+		t.Fatal("a Send to a refused destination answered no error")
+	}
+
+	got := fileLines(t, path)
+	want := []string{keep1, keep2}
+	if len(got) != len(want) {
+		t.Fatalf("the compacted file holds %d lines, want the %d pending rows: %v", len(got), len(want), got)
+	}
+	for i := range want {
+		if got[i] != want[i] {
+			t.Fatalf("line %d is %q, want the pending row %q", i, got[i], want[i])
+		}
+	}
+	rows := o.Pending()
+	if len(rows) != 2 || string(rows[0].Payload) != `{"k":"keep one"}` || string(rows[1].Payload) != `{"k":"keep two"}` {
+		t.Fatalf("Pending answered %v, want the two rows in order", rows)
+	}
+}
+
+// Rows appended after a compaction are read back across a reopen, exactly as
+// rows appended to a file that was never compacted are.
+func TestRowsAppendedAfterACompactionSurviveAReopen(t *testing.T) {
+	path := filepath.Join(t.TempDir(), "outbox.jsonl")
+	seedRetiredRows(t, path, 1500)
+	o, err := Open(path)
+	if err != nil {
+		t.Fatalf("Open: %v", err)
+	}
+	o.Rand = &stepRand{step: 40}
+	if _, err := o.Send(context.Background(), "http://127.0.0.1:1/"); err == nil {
+		t.Fatal("a Send to a refused destination answered no error")
+	}
+	if got := len(fileLines(t, path)); got != 2 {
+		t.Fatalf("the file holds %d lines after the compaction, want 2", got)
+	}
+	appendRow(t, o, `{"k":"after"}`)
+	if err := o.Close(); err != nil {
+		t.Fatalf("Close: %v", err)
+	}
+	o2, err := Open(path)
+	if err != nil {
+		t.Fatalf("reopen: %v", err)
+	}
+	defer o2.Close()
+	rows := o2.Pending()
+	if len(rows) != 3 || string(rows[2].Payload) != `{"k":"after"}` {
+		t.Fatalf("a reopen after a compaction answered %v, want the two rows and the fresh one", rows)
+	}
+}
+
 // A batch sent over http carries the outbox's install nonce as
 // X-Codeaf-Install, one header per post; an outbox with none set sends no
 // header, and a file destination is never asked for one.