fix(realtime): count only deltas toward the compaction byte threshold

Re-seeding the counter with the snapshot's own size left any document larger
than the ceiling permanently over it, forcing a full snapshot append on every
subsequent keystroke — the write amplification the threshold exists to prevent.
The counter measures edit churn since the last fold, so a stream settles at one
snapshot plus that much churn.

Also self-corrects the Tables byte counter whenever its buffer trims to a single
entry, so an independently evicted events key cannot leave the accumulator
over-reporting and pin the buffer at one entry.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>

Author: waleedlatif1

apps/realtime/src/handlers/file-doc-store.test.ts

@@ -436,6 +436,40 @@ describe('FileDocStore', () => {
     doc.destroy()
   })
 
+  it('does not re-compact on every publish once the document itself exceeds the byte ceiling', async () => {
+    const streamKey = `filedoc:stream:${NAME}`
+    const a = await newStore()
+    const doc = new Y.Doc()
+    await a.attachRoom(NAME, doc)
+
+    const updates: Uint8Array[] = []
+    doc.on('update', (u: Uint8Array) => updates.push(u))
+    // Grow the document past the byte ceiling so its own snapshot exceeds it, then keep editing.
+    // Counting the snapshot as appended bytes would leave the threshold permanently breached and
+    // force a full snapshot append per keystroke — the amplification the threshold exists to stop.
+    doc.getText('body').insert(0, 'x'.repeat(12 * 1024 * 1024))
+    for (let i = 0; i < 30; i++) doc.getText('body').insert(0, 'tiny')
+    for (const update of updates) {
+      await a.publishAndWait(NAME, update)
+    }
+    await vi.waitFor(() => {
+      const stream = state.backing!.streams.get(streamKey)!
+      expect(stream.some((entry) => entry.message.s === '1')).toBe(true)
+    })
+
+    const snapshots = state
+      .backing!.streams.get(streamKey)!
+      .filter((entry) => entry.message.s === '1').length
+    expect(snapshots).toBeLessThanOrEqual(2)
+
+    const rebuilt = new Y.Doc()
+    Y.applyUpdate(rebuilt, (await a.getStreamState(NAME))!)
+    expect(rebuilt.getText('body').toString().startsWith('tiny')).toBe(true)
+    expect(rebuilt.getText('body').length).toBe(12 * 1024 * 1024 + 30 * 4)
+    rebuilt.destroy()
+    doc.destroy()
+  })
+
   it('stamps a compaction snapshot of an agent-ONLY stream as an agent frame (never persisted)', async () => {
     const streamKey = `filedoc:stream:${NAME}`
     const noop = Buffer.from(Y.encodeStateAsUpdate(new Y.Doc())).toString('base64')

apps/realtime/src/handlers/file-doc-store.ts

@@ -152,6 +152,9 @@ const COMPACT_THRESHOLD = 400
  * Compaction is the only safe way to shrink one of these streams: a task attaching later
  * replays every entry to rebuild the doc, so dropping the oldest entries — what a native
  * `MAXLEN` retention bound would do — loses edits outright. A snapshot folds them first.
+ *
+ * Measured over deltas appended since the last fold, never over the resulting snapshot, so a
+ * stream settles at roughly one document snapshot plus this much churn.
  */
 const COMPACT_BYTES_THRESHOLD = 8 * 1024 * 1024
 /** Check whether compaction is due only every Nth local publish, to avoid an XLEN per keystroke. */
@@ -233,10 +236,12 @@ interface StoreRoom {
   /** Local publish count, to pace compaction checks. */
   publishes: number
   /**
-   * Bytes this task has appended since the last compaction it observed, so the byte threshold
-   * costs no extra round-trip. Locally tracked, so it under-counts a peer task's appends — it
-   * is a trigger, not an accounting, and {@link COMPACT_THRESHOLD} still covers the case where
-   * many small edits arrive from elsewhere.
+   * Delta bytes this task has appended since the last compaction it performed, so the byte
+   * threshold costs no extra round-trip. Counts deltas only — never the snapshot a compaction
+   * writes, which is a function of document size rather than of edit volume and would make a
+   * large document breach the threshold permanently. Locally tracked, so it under-counts a peer
+   * task's appends: it is a trigger, not an accounting, and {@link COMPACT_THRESHOLD} still
+   * covers many small edits arriving from elsewhere.
    */
   appendedBytes: number
   /** Set once the doc has been observed seeded, so the seed transition itself is never mistaken for an
@@ -782,10 +787,12 @@ export class FileDocStore {
         // appended snapshot id instead would silently drop those un-integrated peer entries.
         const upTo = room.lastId
         const snapshot = Buffer.from(Y.encodeStateAsUpdate(room.doc)).toString('base64')
-        // The folded deltas are about to be trimmed; what remains of this task's contribution is the
-        // snapshot. Reset before the appends so a concurrent publish's bytes are counted against the
-        // new baseline rather than the one being retired.
-        room.appendedBytes = snapshot.length
+        // Counts deltas appended SINCE this fold, so it must not carry the snapshot's own size: a
+        // document whose snapshot already exceeds the ceiling would otherwise re-breach it the instant
+        // compaction finished and force a full snapshot append on every subsequent keystroke — the
+        // write amplification this threshold exists to prevent. Reset before the appends so a
+        // concurrent publish is counted against the new baseline rather than the one being retired.
+        room.appendedBytes = 0
         // Stamp the snapshot by what it folds: a real edit → SNAPSHOT_FIELD (a fresh catch-up treats it
         // as edited content, not a bare seed). An agent-ONLY stream (no real edit yet) → AGENT_FIELD, so a
         // peer catching up applies it as REDIS_AGENT_ORIGIN and never marks the doc edited — preserving

apps/sim/lib/realtime/event-log.ts

@@ -71,6 +71,13 @@ while max_bytes > 0 and total > max_bytes and redis.call('ZCARD', KEYS[1]) > 1 d
   redis.call('ZREMRANGEBYRANK', KEYS[1], 0, 0)
 end
 if total < 0 then total = 0 end
+-- Self-correct: the counter is an accumulator, so an independently evicted events key would leave
+-- it over-reporting forever and pin the buffer at a single entry. Whenever the buffer is down to one
+-- entry its exact size is known, so drift cannot outlive a trim.
+if redis.call('ZCARD', KEYS[1]) == 1 then
+  local only = redis.call('ZRANGE', KEYS[1], 0, 0)
+  if only[1] then total = string.len(only[1]) end
+end
 
 local oldest = redis.call('ZRANGE', KEYS[1], 0, 0, 'WITHSCORES')
 if oldest[2] then