Function processOutboxEvents

apps/sim/lib/core/outbox/service.ts:165–194

(
  handlers: OutboxHandlerRegistry,
  options: { batchSize?: number; maxRuntimeMs?: number; minRemainingMs?: number } = {}
)

 * multiple workers — `SELECT FOR UPDATE SKIP LOCKED` serializes claims.
 */
export async function processOutboxEvents(
  handlers: OutboxHandlerRegistry,
  options: { batchSize?: number; maxRuntimeMs?: number; minRemainingMs?: number } = {}
): Promise<ProcessOutboxResult> {
  const batchSize = options.batchSize ?? 10
  const deadline = options.maxRuntimeMs ? Date.now() + options.maxRuntimeMs : undefined
  const minRemainingMs = options.minRemainingMs ?? DEFAULT_HANDLER_TIMEOUT_MS + 5000

  const reaped = await reapStuckProcessingRows()

  let processed = 0
  let retried = 0
  let deadLettered = 0
  let leaseLost = 0

  for (let i = 0; i < batchSize; i++) {
    if (deadline && Date.now() + minRemainingMs > deadline) break

    const [event] = await claimBatch(1)
    if (!event) break

    const result = await runHandler(event, handlers)
    if (result === 'completed') processed++
    else if (result === 'dead_letter') deadLettered++
    else if (result === 'lease_lost') leaseLost++
    else retried++
  }

  return { processed, retried, deadLettered, leaseLost, reaped }
}

/**
 * Process a specific outbox event immediately after its surrounding