Function acquirePendingChatStream

apps/sim/lib/copilot/request/session/abort.ts:189–268

(
  chatId: string,
  streamId: string,
  timeoutMs = 5_000
)

}

export async function acquirePendingChatStream(
  chatId: string,
  streamId: string,
  timeoutMs = 5_000
): Promise<boolean> {
  // Span records wall time spent waiting for the per-chat stream lock.
  // Typical case: sub-10ms uncontested acquire. Worst case: up to
  // `timeoutMs` spent polling while a prior stream finishes. Previously
  // this time looked like "unexplained gap before llm.stream".
  return withCopilotSpan(
    TraceSpan.CopilotChatAcquirePendingStreamLock,
    {
      [TraceAttr.ChatId]: chatId,
      [TraceAttr.StreamId]: streamId,
      [TraceAttr.LockTimeoutMs]: timeoutMs,
    },
    async (span) => {
      const redis = getRedisClient()
      span.setAttribute(TraceAttr.LockBackend, redis ? AbortBackend.Redis : AbortBackend.InProcess)
      if (redis) {
        const deadline = Date.now() + timeoutMs
        for (;;) {
          try {
            const acquired = await acquireLock(
              getChatStreamLockKey(chatId),
              streamId,
              CHAT_STREAM_LOCK_TTL_SECONDS
            )
            if (acquired) {
              registerPendingChatStream(chatId, streamId)
              span.setAttribute(TraceAttr.LockAcquired, true)
              return true
            }
            if (!pendingChatStreams.has(chatId)) {
              const ownerStreamId = await redis.get(getChatStreamLockKey(chatId))
              if (ownerStreamId) {
                const settled = await waitForPendingChatStream(chatId, 0, ownerStreamId)
                if (settled) {
                  continue
                }
              }
            }
          } catch (error) {
            logger.warn('Failed to acquire chat stream lock', {
              chatId,
              streamId,
              error: toError(error).message,
            })
          }

          if (Date.now() >= deadline) {
            span.setAttribute(TraceAttr.LockAcquired, false)
            span.setAttribute(TraceAttr.LockTimedOut, true)
            return false
          }
          await sleep(200)
        }
      }