Page the ingest drain, back off deferred frames, and write only what snapshots change

[MaxGhenis]

Summary

Two ingest volume issues from the 2026-10-09 SQL plan audit (#215, lens "callers-hotness"). Their plans were fine; the cost was volume. Both were checked against current code before changing anything.

1. Drain. MessageRepository.Unprocessed read every unprocessed frame with its payload, and Worker.drain called it after every enqueued frame. A frame that exhausts its transient retries is left unprocessed by design, so a stuck backlog was re-read, re-decoded and re-applied (four attempts, 35 ms of backoff sleeps) on every later frame.
   • The drain now pages frame positions with a (received_at_ms, inbox_id) cursor (UnprocessedAfter, a range seek on the existing partial inbox_unprocessed_idx) and loads one frame at a time (UnprocessedRecord).
   • A frame that exhausts its transient retries is backed off in-process: 1 s, doubling to 5 min. Run wakes for the earliest retry by itself.
   • No frame is ever dropped. It stays unprocessed in the inbox until it applies or is quarantined, and a restart retries everything at once.
   • New /api/status counter: v2_ingest.per_account.<account>.deferred.
2. Google snapshots. Every pushed snapshot rewrote every identity (twice per participant), the conversation row, and the whole roster (DELETE + re-INSERT). Every Google message frame also rewrote the conversation's recency row, even when the message was older.
   • UpsertIdentity and UpsertConversation now carry WHERE … IS NOT excluded.* guards.
   • ReplaceConversationParticipants diffs the roster through the new SyncConversationParticipants. It writes only the differing rows, confirms an unchanged roster without the write lock, and keeps the old validation and errors. When only the roster changed, it also advances the conversation's updated_at_ms in the same transaction.
   • The worker skips identity and conversation upserts whose content it can see is unchanged.
   • BumpConversationRecency writes only when the message is newer.
   • The reaction fence upsert skips only an unchanged sequence.

The fence write per Google message frame stays, deliberately

The audit suggested skipping the fence write when a snapshot is empty and no fence exists. That is not exact. The fence is what rejects an older snapshot, and frames do arrive out of order: a frame left unprocessed after a transient failure is retried after newer ones. An empty snapshot at time 200 must still record 200, or a retried snapshot from time 100 carrying a reaction would bring that reaction back. Two tests execute this:

• TestEmptySnapshotFenceIsLoadBearingWithoutReactions (store level). Dropping the empty snapshot's fence row lets the stale reaction come back.
• TestOlderGoogleReactionSnapshotCannotOverrideNewerEmptyOne (worker level).

The existing TestReplaceEmbeddedReactionsEmptyFirstSnapshotFencesOlderNonEmptySnapshot already pinned this. The mutation check below plants the audit's skip, and the tests catch it. So the fence still advances on every newer snapshot; only a write at the fence's own sequence is skipped.

Semantic change: updated_at_ms now records the last change

The guards are the point of the change, so identities.updated_at_ms and conversations.updated_at_ms now move only when the row's content changes. A roster change also moves the conversation's updated_at_ms. I searched cmd/ and internal/ for readers:

• Store.ListConversationsUpdatedSince feeds the id-space repair's clobber detection. It now lists the threads whose title, kind, revision or roster changed at or after --since. A re-keyed snapshot that overwrites a thread changes its roster or title, so that thread is still listed. A thread whose snapshot was merely re-sent in the window is no longer listed. Before, if such a thread's roster had changed legitimately between the reference copy and --since, the repair reverted it to the reference roster.
• Store.SelfIdentityID (repair) orders by updated_at_ms. With two or more self identities on one account it now picks the most recently changed one rather than the most recently refreshed.
• No UI, HTTP, MCP, v2read, v2wire or migration code reads either column.

Measured on copies of the live store

The live store was never opened. Its files were copied with cp, and the copy passed integrity_check. The new harness (internal/ingest/replay_harness_test.go, build tag replayharness) replays a copy's inbox through the real decoders into a fresh store, frame by frame in receipt order, with every clock pinned to the frame's receipt time. SQLite's pre-update and commit hooks count row writes and commits, and the WAL is never checkpointed during the replay, so its final size is the volume written.

Full corpus: 30,394 real frames (9,295 Google, 13,943 Signal, 7,156 WhatsApp):

	origin/main	this branch
row writes	170,660	99,305 (−42%)
identity updates	37,654	186
conversation updates	23,622	7,857
roster rows (delete / insert / update)	9,163 / 10,823 / 0	41 / 1,701 / 122
fence rows (insert / update)	2,781 / 2,438	2,781 / 2,438
commits	151,295	107,691 (−29%)
WAL written	2.00 GB	1.77 GB (−12%)
frames left unprocessed	0	0

Google frames only (9,295 frames, two alternating runs each):

	origin/main	this branch
row writes	79,685	33,772 (−58%)
commits	62,195	34,063 (−45%)
WAL written	649 MB	486 MB (−25%)
median drain per frame	0.89, 0.94 ms	0.71, 0.77 ms

The machine's load average was 86–94, so read the timings as indicative; the counts are deterministic.

Differential. I compared the two full-corpus result stores table by table, every column. They are identical except updated_at_ms on 831 identities and 264 conversations, which is always older on the branch and never newer. Fence rows are identical in every column, updated_at_ms included.

Stuck backlog. 600 Google frames, with the first 20 conversation frames selected to fail transiently on every attempt. On both commits, 6 of the 20 fail validation before the fault point and are quarantined.

	origin/main	this branch
drain cost per later frame	686 ms average (352 s over 514 frames)	1.9 ms average (0.99 s)
re-decodes of the stuck frames	30,014	118
row writes	64,109	2,459
unprocessed after the fault clears and the worker restarts	0	0

One unchanged 20-member snapshot. I counted statements from the code path:

• origin/main: about 153 statements and 85 row writes (41 identity, 1 conversation, 42 roster, plus the inbox mark).
• this branch: about 45 statements, all reads except the inbox mark.

TestRepeatedGroupSnapshotWritesNothing asserts zero identity, conversation and roster writes, counted by triggers.

Residual cost. SQLite does not treat a partial index as covering for its own WHERE column. A backed-off frame therefore still costs one inbox row read per drain to confirm processed_at_ms IS NULL, plus its overflow pages when the payload is larger than a page (about 2% of frames). It costs no decode or apply. A covering index would need a migration, and 0011 is already contested by #196, #201, #210 and #215.

Invariants (stated, and tested for every input the generators reach)

• Content. For any sequence of writes, the guarded statements leave every row equal to what the replaced statements leave, and raise errors of the same classes. The exceptions are updated_at_ms on identities and conversations, and the fence's updated_at_ms. The replaced statements are kept verbatim in write_guards_test.go as the reference. The differential properties cover:

  • UpsertIdentity and UpsertConversation, including CHECK failures and conflicts under a different primary key;
  • the roster (duplicates, cross-account and orphan identities, invalid roles and times, contradictory IDs, missing conversations);
  • the reaction fence, with out-of-order and equal sequences;
  • the recency bump.

• Update time. A guarded row's updated_at_ms is the time of its last content change (for a conversation, a roster change counts), or its creation time. It is never later than the replaced statement's value.

• No-op. An unchanged identity, conversation, roster, older recency bump, or same-sequence or stale fence write touches no row. An unchanged roster takes no write lock (tested under a held BEGIN IMMEDIATE).

• Fence. An older snapshot is rejected exactly when its sequence is below the newest accepted one, and an emptying snapshot always advances the fence.

• Paging. Walking UnprocessedAfter from the zero cursor visits every frame that is unprocessed when its page is listed, exactly once, in (received_at_ms, inbox_id) order. This holds with receipt-time ties, frames processed mid-walk and frames appended mid-walk.

• No frame lost or repeated. Under random transient failures and random interleavings of appends, drains and clock advances:

  • once the failures stop, every frame is projected exactly once, and none is quarantined;
  • each frame is decoded exactly 4 × (its failing rounds) + 1 times;
  • frames that never fail apply in receipt order.

  A deferred frame is not decoded again before its retry time, and is retried without new traffic.

• Worker model. resolveIdentityFor and refreshConversation leave the identity, conversation and roster equal to a model of the merge rules after every random snapshot. The update time equals the last change.

• Real traffic. The 30,394-frame differential above.

Mutation check. mutants.py plants 31 bugs, one at a time, and runs the relevant tests:

• each of the 12 guard terms dropped;
• the roster diff ignoring a column, never deleting, accepting a repeated identity on the fast path, or never touching the conversation;
• a recency bump on equal;
• an unguarded fence, and the audit's empty-snapshot fence skip;
• a cursor that repeats;
• no backoff, no doubling, no prune, a single page, no wake timer, an exhausted frame not deferred;
• worker skips that ignore the self flag, raw value or revision, and a refresh that never touches.

All 31 are caught at this head, each by a named failing test. A build or setup failure is reported as an error, not a catch, and the script uses its own Go cache.

How the test set got there:

• The first run caught 22 of 23. The fast-path repeat survived, and TestSyncConversationParticipantsRejectsRepeatedIdentity now catches it.
• A later run showed that random sequences reach a single-column archived_at_ms change too rarely. TestUpsertGuardsWriteEveryChangedColumn now changes every guarded column on its own, NULL transitions included.

Coordination

• No migration, so this ships without app/CLI lockstep.
• Text overlaps with open PRs touching worker.go, messages.go and repositories.go (Prune old v2 inbox payloads but keep every row and dedupe key #196, Keep same-roster Google threads distinct; re-key only threads whose id is dead #201, Leave unaddressable roster entries off live Google snapshots instead of quarantining them #211, Keep v2 reads off whole-table scans found by a SQL plan audit #215). Whichever lands second carries the other's change over.
• CI now also vets the build-tagged replay harness (go vet -tags replayharness ./internal/ingest/) so it cannot rot.

Test plan

• GOWORK=off go test ./...
• New tests under -race (sqlite and ingest packages)
• Mutation check: 31 planted bugs, all caught at d0edaec
• Full-corpus replay at origin/main and this branch, with the table-by-table comparison
• Stuck-backlog replay at both commits
• CI
• Independent review (GPT-6.1 Sol)

Evidence (local): ~/reviews/openmessage-ingest-write-volume-2026-10-09/. It holds the metrics JSON for every run, the comparison report, the mutation script and report, and the review brief.

🤖 Generated with Claude Code