Function doFlush

apps/sim/lib/execution/event-buffer.ts:785–875

(terminalStatus?: TerminalExecutionStreamStatus)

  let writeFailure: Error | null = null

  const doFlush = async (terminalStatus?: TerminalExecutionStreamStatus): Promise<boolean> => {
    if (pending.length === 0) return true
    const batch = pending
    pending = []
    try {
      const key = getEventsKey(executionId)
      const zaddArgs: (string | number)[] = []
      let batchBytes = 0
      for (const entry of batch) {
        const entryJson = getExecutionEventEntryJson(entry)
        batchBytes += Buffer.byteLength(entryJson, 'utf8')
        zaddArgs.push(entry.eventId, entryJson)
      }
      const budgetReservation: ExecutionRedisBudgetReservation = {
        executionId,
        userId: context.userId,
        category: 'event_buffer',
        operation: terminalStatus ? 'write_terminal_events' : 'write_events',
        bytes: batchBytes,
        logger,
      }
      const limits = getExecutionRedisBudgetLimits()
      if (batchBytes > limits.maxSingleWriteBytes) {
        throw new ExecutionResourceLimitError({
          resource: 'redis_key_bytes',
          attemptedBytes: batchBytes,
          limitBytes: limits.maxSingleWriteBytes,
        })
      }
      const budgetKeys = getExecutionRedisBudgetKeys(budgetReservation)
      const flushResult = getFlushScriptResult(
        await redis.eval(
          FLUSH_EVENTS_SCRIPT,
          3 + budgetKeys.length,
          key,
          getSeqKey(executionId),
          getMetaKey(executionId),
          ...budgetKeys,
          TTL_SECONDS,
          EVENT_LIMIT,
          new Date().toISOString(),
          terminalStatus ?? '',
          batchBytes,
          limits.maxExecutionBytes,
          limits.maxUserBytes,
          limits.ttlSeconds,
          ...zaddArgs
        )
      )
      if (!flushResult.allowed) {
        throw new ExecutionResourceLimitError({
          resource:
            flushResult.resource === 'user_redis_bytes'
              ? 'user_redis_bytes'
              : 'execution_redis_bytes',
          attemptedBytes: batchBytes,
          currentBytes: flushResult.currentBytes ?? 0,
          limitBytes: