fix(execution): measure pressure at write time and keep terminal status last

Pressure was read from bytes counted once a flush succeeded, but a burst is
compacted long before the scheduled flush runs — so the very batch that
exhausts the budget went through at the loose ceiling and was dropped instead
of offloaded. Count bytes as each event is compacted.

Separately, the terminal-alone retry stamped terminal status while entries
queued ahead of it were still unwritten. Terminal status is the reader's
end-of-run signal: a reconnecting client drains what is in Redis and closes, so
those entries were stranded behind a stream it had already finished with. Drain
the backlog first, then publish the terminal event.

Author: waleedlatif1

apps/sim/lib/execution/event-buffer.test.ts

@@ -671,6 +671,77 @@ describe('execution event buffer', () => {
     expect(persisted).not.toContain(payload)
   })
 
+  /**
+   * Terminal status is the reader's end-of-run signal: once it lands, a
+   * reconnecting client drains what is in Redis and closes. Stamping it while
+   * lower event ids are still queued strands those events behind a stream the
+   * reader has already finished with.
+   *
+   * Needs a backlog past the single-write cap so chunking leaves a remainder
+   * behind the terminal entry — the only shape where that ordering can invert.
+   */
+  it('does not stamp terminal status while earlier events are still queued', async () => {
+    mockRedis.incrby.mockResolvedValue(100000)
+    const idsAtStamp: number[] = []
+    mockRedis.eval.mockImplementation(async (script: string, ...args: unknown[]) => {
+      if (!isFlushScript(script)) return [1, 'ok', 0, 0]
+      const { terminalStatus, zaddArgs } = parseFlushEvalArgs(args)
+      // Reject any multi-entry batch, forcing the terminal-alone retry path.
+      if (zaddArgs.length > 2) return [0, 'execution_redis_bytes', 64 * 1024 * 1024]
+      for (let i = 0; i < zaddArgs.length; i += 2) {
+        persistedEntries.push(JSON.parse(zaddArgs[i + 1] as string) as ExecutionEventEntry)
+      }
+      if (terminalStatus && idsAtStamp.length === 0) {
+        idsAtStamp.push(...persistedEntries.map((e) => e.eventId))
+      }
+      return [1, 1, 0]
+    })
+
+    // ~3MB per event, so three of them exceed the 8MiB single-write cap and the
+    // chunk boundary leaves a remainder queued behind the terminal entry.
+    const payload = 'x'.repeat(1_500_000)
+    const writer = createExecutionEventWriter('exec-1', {
+      workspaceId: 'ws-1',
+      workflowId: 'wf-1',
+    })
+    for (let i = 0; i < 3; i++) {
+      await writer.write(makeEvent(payload)).catch(() => {})
+    }
+    await writer.writeTerminal(makeEvent('terminal'), 'complete').catch(() => {})
+    await writer.close().catch(() => {})
+
+    const terminalId = Math.max(...persistedEntries.map((e) => e.eventId))
+    const strandedAtStamp = persistedEntries
+      .map((e) => e.eventId)
+      .filter((id) => id < terminalId && !idsAtStamp.includes(id))
+    expect(strandedAtStamp).toEqual([])
+  })
+
+  /**
+   * Pressure has to be measured as events are produced, not once a flush
+   * succeeds. A burst is compacted long before the scheduled flush runs, so
+   * flush-time accounting would let the very batch that exhausts the budget
+   * through at the loose ceiling and drop it instead of offloading it.
+   */
+  it('engages pressure within a burst that has not flushed yet', async () => {
+    mockRedis.incrby.mockResolvedValue(100000)
+    const payload = 'x'.repeat(2 * 1024 * 1024)
+
+    const writer = createExecutionEventWriter('exec-1', {
+      workspaceId: 'ws-1',
+      workflowId: 'wf-1',
+    })
+    // No flush between writes: everything stays pending while the burst builds.
+    for (let i = 0; i < 20; i++) {
+      await writer.write(makeEvent(payload)).catch(() => {})
+    }
+    await writer.flush().catch(() => {})
+
+    // The later events in the burst must have been offloaded, not left inline.
+    const persisted = JSON.stringify(persistedEntries)
+    expect(persisted).toContain(LARGE_VALUE_REF_MARKER)
+  })
+
   it('preserves requested UserFile base64 when buffering terminal events', async () => {
     mockRedis.incrby.mockResolvedValue(100)
     const base64 = Buffer.from('hello').toString('base64')

apps/sim/lib/execution/event-buffer.ts

@@ -765,9 +765,12 @@ export function createExecutionEventWriter(
   let flushTimer: ReturnType<typeof setTimeout> | null = null
   let consecutiveFlushFailures = 0
   /**
-   * Bytes this execution has successfully buffered. Counted gross rather than
-   * net of ring-buffer pruning, so it reaches the pressure mark early — erring
-   * toward offloading sooner is the safe direction.
+   * Bytes this execution has produced, counted as each event is compacted
+   * rather than once a flush succeeds. A burst can be compacted long before the
+   * scheduled flush runs, so flush-time accounting would let the very batch that
+   * exhausts the budget through at the loose ceiling. Counted gross of
+   * ring-buffer pruning too, so the mark is reached early — erring toward
+   * offloading sooner is the safe direction.
    */
   let bufferedBytes = 0
 
@@ -947,7 +950,6 @@ export function createExecutionEventWriter(
       }
       consecutiveFlushFailures = 0
       lastResourceLimitError = null
-      bufferedBytes += batchBytes
       if (chunkTerminalStatus) pendingTerminalStatus = undefined
       return true
     } catch (error) {
@@ -1032,6 +1034,7 @@ export function createExecutionEventWriter(
       valueThresholdBytes: getValueThresholdBytes(),
     })
     const entry: ExecutionEventEntry = { eventId, executionId, event: compactEvent }
+    bufferedBytes += getJsonSize(entry) ?? 0
     pending.push(entry)
     if (pending.length >= FLUSH_MAX_BATCH) {
       await flushPending()
@@ -1079,6 +1082,7 @@ export function createExecutionEventWriter(
         valueThresholdBytes: getValueThresholdBytes(),
       })
       const entry: ExecutionEventEntry = { eventId, executionId, event: compactEvent }
+      bufferedBytes += getJsonSize(entry) ?? 0
       pending.push(entry)
       let ok = false
       try {
@@ -1091,9 +1095,16 @@ export function createExecutionEventWriter(
           // budget rejection specifically: a transient Redis error leaves the batch
           // queued for retry, and clearing it here would turn that into data loss.
           const remaining = pending.filter((pendingEntry) => pendingEntry !== entry)
+          // Drain what is queued ahead of the terminal event first. Terminal
+          // status is the reader's end-of-run signal: stamping it while lower
+          // event ids are still queued strands them behind a stream the reader
+          // has already drained and closed.
+          if (remaining.length > 0) {
+            pending = remaining
+            await flushPending(false)
+          }
           pending = [entry]
           ok = await flushPending(false)
-          pending = pending.concat(remaining)
         }
       } catch (error) {
         discardTerminalEntry(entry)