fix(redis): keep counters outliving their data, and stop a failed fold retrying hot

The copilot owner counter used a fixed one-hour window while the stream TTL is
configurable and defaults to exactly that. Raising COPILOT_STREAM_TTL_SECONDS
would have let the counter expire under live data, and the next write would see
zero reserved and grant another full ceiling. The window is now the larger of
the two, so a counter can never expire before what it accounts for.

A failed fold deliberately leaves the trigger armed, but the snapshot XADD lands
before the XTRIM — so a persistent trim failure retried immediately, appending a
full-document snapshot every time and turning a Redis blip into the write
amplification the threshold exists to prevent. A forced fold now waits out a
cooldown after a failure. The entry-count path is unaffected.

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

Author: waleedlatif1

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

@@ -134,6 +134,7 @@ interface StoreRoomInternals {
   pendingDeltas: Array<{ id: string; bytes: number }>
   realEdited: boolean
   publishes: number
+  compactRetryAfter: number
   doc: Y.Doc
   seededObserved: boolean
 }
@@ -365,6 +366,7 @@ describe('FileDocStore', () => {
       doc: new Y.Doc(),
       lastId: '400-0',
       publishes: 0,
+      compactRetryAfter: 0,
       pendingDeltas: [],
       seededObserved: true,
       realEdited: true,
@@ -507,11 +509,21 @@ describe('FileDocStore', () => {
       throw new Error('redis blip')
     }
     await internals(a).maybeCompact(NAME, true)
-    write.xTrim = original
 
     // A failed fold must not disarm the trigger — otherwise the stream stays oversized until
     // this task happens to append another full threshold's worth of deltas.
     expect(room.pendingDeltas).toEqual([{ id: '1-0', bytes: 9 * 1024 * 1024 }])
+
+    // But it must not retry immediately either: the snapshot XADD lands before the XTRIM, so a
+    // persistent trim failure would append a full-document snapshot on every attempt.
+    const snapshotsAfterFailure = state.backing!.streams.get(`filedoc:stream:${NAME}`)?.length ?? 0
+    await internals(a).maybeCompact(NAME, true)
+    await internals(a).maybeCompact(NAME, true)
+    expect(state.backing!.streams.get(`filedoc:stream:${NAME}`)?.length ?? 0).toBe(
+      snapshotsAfterFailure
+    )
+
+    write.xTrim = original
     doc.destroy()
   })
 
@@ -557,6 +569,7 @@ describe('FileDocStore', () => {
       doc: agentDoc,
       lastId: '400-0',
       publishes: 0,
+      compactRetryAfter: 0,
       pendingDeltas: [],
       seededObserved: true,
       realEdited: false,
@@ -724,6 +737,7 @@ describe('FileDocStore', () => {
       doc: docA,
       lastId: '401-0',
       publishes: 0,
+      compactRetryAfter: 0,
       pendingDeltas: [],
       seededObserved: true,
       realEdited: true,
@@ -732,6 +746,7 @@ describe('FileDocStore', () => {
       doc: new Y.Doc(),
       lastId: '400-0',
       publishes: 0,
+      compactRetryAfter: 0,
       pendingDeltas: [],
       seededObserved: true,
       realEdited: true,

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

@@ -162,6 +162,15 @@ const COMPACT_CHECK_EVERY = 64
 /** Compaction critical section (snapshot + xAdd + xTrim) is fast; a generous TTL covers a slow Redis
  * round-trip without risking expiry mid-compact. Released via compare-and-delete regardless. */
 const COMPACT_LOCK_TTL_MS = 10_000
+/**
+ * Quiet period after a failed fold before another may be forced.
+ *
+ * A failed fold deliberately leaves the trigger armed so the bytes are not forgotten, but the
+ * snapshot `XADD` lands before the `XTRIM` — so if the trim is what failed, retrying immediately
+ * appends another full-document snapshot each time, turning a Redis blip into exactly the write
+ * amplification the threshold exists to prevent. The entry-count path is unaffected.
+ */
+const COMPACT_RETRY_COOLDOWN_MS = 30_000
 /** Retry a failed stream append this many times before giving up, so a transient Redis blip doesn't
  * silently drop an edit from the shared log (which no peer would then ever see). */
 const PUBLISH_MAX_RETRIES = 3
@@ -253,6 +262,8 @@ interface StoreRoom {
   lastId: string
   /** Local publish count, to pace compaction checks. */
   publishes: number
+  /** Epoch ms before which no forced fold is attempted, after one failed. */
+  compactRetryAfter: number
   /**
    * Deltas this task has appended and not yet folded, as `{id, bytes}` pairs in append order.
    *
@@ -349,6 +360,7 @@ export class FileDocStore {
       doc,
       lastId: '0',
       publishes: 0,
+      compactRetryAfter: 0,
       pendingDeltas: [],
       seededObserved: false,
       realEdited: false,
@@ -804,6 +816,7 @@ export class FileDocStore {
     const room = this.rooms.get(name)
     if (!room) return
     try {
+      if (force && Date.now() < room.compactRetryAfter) return
       if (!force && (await this.write.xLen(streamKey(name))) < COMPACT_THRESHOLD) return
       const key = `${COMPACT_LOCK_PREFIX}${name}`
       const token = await this.acquireLock(key, COMPACT_LOCK_TTL_MS)
@@ -844,6 +857,7 @@ export class FileDocStore {
         await this.releaseLock(key, token)
       }
     } catch (error) {
+      room.compactRetryAfter = Date.now() + COMPACT_RETRY_COOLDOWN_MS
       logger.warn(`FileDocStore compaction failed for ${name}`, { error: getErrorMessage(error) })
     }
   }

apps/sim/lib/copilot/request/session/buffer.ts

@@ -260,6 +260,13 @@ export async function appendEvents(
     ...(scope?.userId ? { userId: scope.userId } : {}),
   }
   const budgetKeys = getRedisBudgetKeys(budgetScope)
+  /*
+    A counter must never expire before the data it accounts for: the next write would then
+    see zero reserved and let the stream grow by another full ceiling. `COPILOT_STREAM_TTL_SECONDS`
+    is configurable and defaults to exactly the budget window, so raising it would otherwise
+    break that invariant silently.
+  */
+  const budgetTtlSeconds = Math.max(limits.ttlSeconds, config.ttlSeconds)
 
   /*
     Redis measures a member in UTF-8 bytes, so the ceiling has to be measured the same
@@ -321,7 +328,7 @@ export async function appendEvents(
         config.eventLimit,
         limits.maxOwnerBytes,
         limits.maxUserBytes,
-        limits.ttlSeconds,
+        budgetTtlSeconds,
         String(chunk.members[chunk.members.length - 1].seq),
         ...zaddArgs
       )