Function admitFromQueue

web/src/server/free-session/store.ts:405–484

(params: {
  model: string
  sessionLengthMs: number
  now: Date
  health: FireworksHealth
})

 * unhealthy means "upstream unreachable or saturated").
 */
export async function admitFromQueue(params: {
  model: string
  sessionLengthMs: number
  now: Date
  health: FireworksHealth
}): Promise<{
  admitted: InternalSessionRow[]
  skipped: FireworksHealth | null
}> {
  const { model, sessionLengthMs, now, health } = params

  if (health !== 'healthy') {
    return { admitted: [], skipped: health }
  }

  return db.transaction(async (tx) => {
    // Per-model lock: hashing the model into the lock id lets distinct model
    // queues admit concurrently while still serializing within a single queue.
    const modelLockId = FREEBUFF_ADMISSION_LOCK_ID + hashStringToInt32(model)
    const lockResult = await tx.execute<{ acquired: unknown }>(
      sql`SELECT pg_try_advisory_xact_lock(${modelLockId}) AS acquired`,
    )
    if (
      !coerceBool(
        (lockResult as unknown as Array<{ acquired: unknown }>)[0]?.acquired,
      )
    ) {
      return { admitted: [], skipped: null }
    }

    const candidates = await tx
      .select({ user_id: schema.freeSession.user_id })
      .from(schema.freeSession)
      .where(
        and(
          eq(schema.freeSession.status, 'queued'),
          eq(schema.freeSession.model, model),
        ),
      )
      .orderBy(
        asc(schema.freeSession.queued_at),
        asc(schema.freeSession.user_id),
      )
      .limit(1)
      .for('update', { skipLocked: true })

    const candidate = candidates[0]
    if (!candidate) return { admitted: [], skipped: null }

    const expiresAt = new Date(now.getTime() + sessionLengthMs)
    const admitted = await tx
      .update(schema.freeSession)
      .set({
        status: 'active',
        admitted_at: now,
        expires_at: expiresAt,
        updated_at: now,
      })