How to run concurrency, throughput (TPS), and Time to First Token (TTFT) load tests against Fusion AI Gateway.

Benchmarks & Load Testing

Fusion AI Gateway is deployed globally on the Cloudflare Workers edge network, providing low Time to First Token (TTFT), high sustained tokens per second (TPS), and sub-50ms prefix cache lookups.

This guide provides standalone benchmark scripts in TypeScript (Bun / Node) and Python to verify concurrency, latency, and throughput under real load.

1. Quick Concurrency & TPS Benchmark (TypeScript / Bun)

This script runs configurable parallel worker streams against the Fusion /v1/chat/completions endpoint, measuring TTFT, TPS, and error rates.

Script: `benchmark.ts`

/**
 * Fusion AI Gateway — Concurrency & Throughput Benchmark
 *
 * Usage:
 *   FUSION_API_KEY="fc_your_api_key" bun run benchmark.ts
 */

const FUSION_API_KEY = process.env.FUSION_API_KEY || "YOUR_API_KEY_HERE"
const FUSION_URL = process.env.FUSION_URL || "https://api.fusioncode.app/v1/chat/completions"
const MODEL = process.env.MODEL || "deepseek-ai/DeepSeek-V4-Flash-0731"

const CONCURRENCY = 50      // Number of concurrent worker streams
const TOTAL_REQUESTS = 100  // Total requests to execute

interface BenchmarkResult {
  reqId: number
  ttftMs: number | null
  totalMs: number
  tokens: number
  tps: number | null
  statusCode: number
  cacheHeader: string | null
  error?: string
}

async function sendRequest(reqId: number): Promise<BenchmarkResult> {
  const start = performance.now()
  let ttftMs: number | null = null
  let tokens = 0

  try {
    const res = await fetch(FUSION_URL, {
      method: "POST",
      headers: {
        Authorization: `Bearer ${FUSION_API_KEY}`,
        "Content-Type": "application/json",
      },
      body: JSON.stringify({
        model: MODEL,
        messages: [
          {
            role: "user",
            content: `Explain distributed cache invalidation in 2 concise paragraphs. [run:${reqId}]`,
          },
        ],
        stream: true,
        max_tokens: 150,
      }),
    })

    const cacheHeader = res.headers.get("x-fusion-cache")

    if (!res.ok) {
      const errText = await res.text().catch(() => "")
      return {
        reqId,
        ttftMs: null,
        totalMs: performance.now() - start,
        tokens: 0,
        tps: null,
        statusCode: res.status,
        cacheHeader,
        error: `HTTP ${res.status}: ${errText.slice(0, 100)}`,
      }
    }

    const reader = res.body?.getReader()
    if (!reader) throw new Error("No response stream body")

    const decoder = new TextDecoder()
    let firstChunk = false

    while (true) {
      const { done, value } = await reader.read()
      if (done) break
      if (!firstChunk) {
        ttftMs = performance.now() - start
        firstChunk = true
      }
      const chunk = decoder.decode(value)
      const matches = chunk.match(/data:\s*\{/g)
      if (matches) tokens += matches.length
    }

    const totalMs = performance.now() - start
    const genSec = (totalMs - (ttftMs ?? 0)) / 1000
    const tps = genSec > 0 ? tokens / genSec : null

    return {
      reqId,
      ttftMs,
      totalMs,
      tokens,
      tps,
      statusCode: 200,
      cacheHeader,
    }
  } catch (err: any) {
    return {
      reqId,
      ttftMs: null,
      totalMs: performance.now() - start,
      tokens: 0,
      tps: null,
      statusCode: 500,
      cacheHeader: null,
      error: err.message,
    }
  }
}

async function main() {
  console.log("==========================================================")
  console.log(`🚀 Fusion AI Gateway Benchmark: ${MODEL}`)
  console.log(`Endpoint   : ${FUSION_URL}`)
  console.log(`Concurrency: ${CONCURRENCY} parallel streams | Total: ${TOTAL_REQUESTS}`)
  console.log("==========================================================\n")

  const results: BenchmarkResult[] = []
  let completed = 0

  const queue = Array.from({ length: TOTAL_REQUESTS }, (_, i) => i + 1)
  const workers = Array.from({ length: CONCURRENCY }, async () => {
    while (queue.length > 0) {
      const reqId = queue.shift()!
      const res = await sendRequest(reqId)
      results.push(res)
      completed++
      process.stdout.write(`\rProgress: ${completed}/${TOTAL_REQUESTS} requests completed...`)
    }
  })

  await Promise.all(workers)

  const successful = results.filter((r) => r.statusCode === 200)
  const errors = results.filter((r) => r.statusCode !== 200)

  const avgTtft =
    successful.reduce((acc, r) => acc + (r.ttftMs ?? 0), 0) / (successful.length || 1)
  const avgTotal =
    successful.reduce((acc, r) => acc + r.totalMs, 0) / (successful.length || 1)
  const avgTps =
    successful.reduce((acc, r) => acc + (r.tps ?? 0), 0) / (successful.length || 1)

  console.log("\n\n📊 Benchmark Results Summary:")
  console.log("----------------------------------------------------------")
  console.log(`• Total Requests : ${TOTAL_REQUESTS}`)
  console.log(`• Success Rate   : ${successful.length}/${TOTAL_REQUESTS} (${((successful.length / TOTAL_REQUESTS) * 100).toFixed(1)}%)`)
  console.log(`• Errors / 429s  : ${errors.length}`)
  console.log(`• Avg TTFT       : ${avgTtft.toFixed(1)} ms`)
  console.log(`• Avg Gen Speed  : ${avgTps.toFixed(1)} tokens/sec`)
  console.log(`• Avg Total Latency: ${avgTotal.toFixed(1)} ms`)
  console.log("==========================================================\n")
}

main()

Running the TypeScript Benchmark

FUSION_API_KEY="fc_your_api_key_here" bun run benchmark.ts

2. Python Async Benchmark (httpx + asyncio)

If your test harness is written in Python, use httpx with asyncio to test streaming throughput:

Script: `benchmark.py`

import asyncio
import os
import time
import httpx

FUSION_API_KEY = os.getenv("FUSION_API_KEY", "fc_your_api_key_here")
FUSION_BASE_URL = os.getenv("FUSION_URL", "https://api.fusioncode.app/v1/chat/completions")
MODEL = os.getenv("MODEL", "deepseek-ai/DeepSeek-V4-Flash-0731")

CONCURRENCY = 25
TOTAL_REQUESTS = 50

async def benchmark_single(client: httpx.AsyncClient, req_id: int):
    start = time.perf_counter()
    first_token_time = None
    tokens = 0
    
    payload = {
        "model": MODEL,
        "messages": [{"role": "user", "content": f"Explain memory caching in 2 sentences. [run:{req_id}]"}],
        "stream": True,
        "max_tokens": 150
    }
    headers = {
        "Authorization": f"Bearer {FUSION_API_KEY}",
        "Content-Type": "application/json"
    }

    try:
        async with client.stream("POST", FUSION_BASE_URL, json=payload, headers=headers, timeout=30.0) as response:
            if response.status_code != 200:
                print(f"[{req_id}] Error: HTTP {response.status_code}")
                return None
            
            async for line in response.aiter_lines():
                if line.startswith("data: ") and "[DONE]" not in line:
                    if first_token_time is None:
                        first_token_time = (time.perf_counter() - start) * 1000
                    tokens += 1

        total_time = (time.perf_counter() - start) * 1000
        gen_time = (total_time - (first_token_time or 0)) / 1000
        tps = tokens / gen_time if gen_time > 0 else 0
        
        return {
            "req_id": req_id,
            "ttft_ms": first_token_time,
            "total_ms": total_time,
            "tokens": tokens,
            "tps": tps
        }
    except Exception as e:
        print(f"[{req_id}] Exception: {e}")
        return None

async def main():
    print(f"🚀 Running Python Async Benchmark against {FUSION_BASE_URL} ({MODEL})")
    sem = asyncio.Semaphore(CONCURRENCY)

    async with httpx.AsyncClient() as client:
        async def bounded_req(i):
            async with sem:
                return await benchmark_single(client, i)

        tasks = [bounded_req(i) for i in range(TOTAL_REQUESTS)]
        results = await asyncio.gather(*tasks)

    valid = [r for r in results if r is not None]
    avg_ttft = sum(r["ttft_ms"] for r in valid if r["ttft_ms"]) / len(valid) if valid else 0
    avg_tps = sum(r["tps"] for r in valid) / len(valid) if valid else 0

    print("\n📊 Summary:")
    print(f"• Success Rate : {len(valid)}/{TOTAL_REQUESTS}")
    print(f"• Avg TTFT     : {avg_ttft:.1f} ms")
    print(f"• Avg TPS      : {avg_tps:.1f} tokens/sec")

if __name__ == "__main__":
    asyncio.run(main())

Running the Python Benchmark

pip install httpx
FUSION_API_KEY="fc_your_api_key_here" python benchmark.py

3. Interpreting Benchmark Headers

Every response from Fusion AI Gateway includes performance headers:

Header	Meaning	Description
`x-fusion-cache`	`HIT` / `MISS`	Indicates whether the request was served directly from Fusion’s Edge KV cache.
`x-fusion-routed-model`	Model String	The exact upstream model executing the request.
`x-fusion-edge`	`HIT` / `MISS`	Indicates in-memory edge isolate cache hit.

Benchmarks & Load Testing

Benchmarks & Load Testing

1. Quick Concurrency & TPS Benchmark (TypeScript / Bun)

Script: benchmark.ts

Running the TypeScript Benchmark

2. Python Async Benchmark (httpx + asyncio)

Script: benchmark.py

Running the Python Benchmark

3. Interpreting Benchmark Headers

Script: `benchmark.ts`

Script: `benchmark.py`