Function startAbortPoller

apps/sim/lib/copilot/request/session/abort.ts:326–387

(
  streamId: string,
  abortController: AbortController,
  options?: { pollMs?: number; requestId?: string; chatId?: string }
)

const pollingStreams = new Set<string>()

export function startAbortPoller(
  streamId: string,
  abortController: AbortController,
  options?: { pollMs?: number; requestId?: string; chatId?: string }
): ReturnType<typeof setInterval> {
  const pollMs = options?.pollMs ?? DEFAULT_ABORT_POLL_MS
  const requestId = options?.requestId
  const chatId = options?.chatId

  let lastHeartbeatAt = Date.now()
  let heartbeatOwnershipLost = false

  return setInterval(() => {
    if (pollingStreams.has(streamId)) return
    pollingStreams.add(streamId)

    void (async () => {
      try {
        const shouldAbort = await hasAbortMarker(streamId)
        if (shouldAbort && !abortController.signal.aborted) {
          abortController.abort(AbortReason.RedisPoller)
          await clearAbortMarker(streamId)
        }
      } catch (error) {
        logger.warn('Failed to poll stream abort marker', {
          streamId,
          ...(requestId ? { requestId } : {}),
          error: toError(error).message,
        })
      } finally {
        pollingStreams.delete(streamId)
      }

      if (!chatId || heartbeatOwnershipLost) return
      if (Date.now() - lastHeartbeatAt < CHAT_STREAM_LOCK_HEARTBEAT_INTERVAL_MS) return

      try {
        const owned = await extendLock(
          getChatStreamLockKey(chatId),
          streamId,
          CHAT_STREAM_LOCK_TTL_SECONDS
        )
        lastHeartbeatAt = Date.now()
        if (!owned) {
          heartbeatOwnershipLost = true
          logger.warn('Lost ownership of chat stream lock — stopping heartbeat', {
            chatId,
            streamId,
            ...(requestId ? { requestId } : {}),
          })
        }
      } catch (error) {
        logger.warn('Failed to extend chat stream lock TTL', {
          chatId,
          streamId,
          ...(requestId ? { requestId } : {}),
          error: toError(error).message,
        })