K2
Serverless, durable event streams
K2 enables applications to produce events to a durable, ordered stream. Subscribe to the stream with one or more consumers.
Truly serverless
Long-term retention
Flexible consumption
Produce from anywhere. Consume at your own pace.
Append records from a Worker or over HTTP, then read them through subscriptions that track their own position in the stream.
interface Env { ORDERS: K2Producer}
export default { async fetch(request, env): Promise<Response> { const event = await request.json() const encoder = new TextEncoder()
// Append a batch of records: either all are stored, or none are const result = await env.ORDERS.send([ { content: encoder.encode(JSON.stringify(event)), headers: { "event-type": "order.created" }, }, ])
// send() returns a result instead of throwing if (!result.success) { return new Response(result.error.message, { status: result.error.retryable ? 503 : 500, }) }
return new Response("Event recorded") },} satisfies ExportedHandler<Env>const K2_ENDPOINT = "https://<STREAM_ID>.k2.cloudflarestorage.com"const SUBSCRIPTION = `${K2_ENDPOINT}/subscriptions/<SUBSCRIPTION_ID>`const WORKER_ID = "worker-1"
async function post(path: string, body: object) { const res = await fetch(`${SUBSCRIPTION}${path}`, { method: "POST", headers: { Authorization: `Bearer ${process.env.CLOUDFLARE_API_TOKEN}`, "Content-Type": "application/json", }, body: JSON.stringify(body), }) return (await res.json()).result}
const decoder = new TextDecoder()
while (true) { // Lease up to 100 records to this worker for five minutes const batch = await post("/consume", { worker_id: WORKER_ID, max_records: 100 })
if (!batch?.batch_id) { // No new records: back off before polling again await new Promise((resolve) => setTimeout(resolve, 1000)) continue }
try { for (const record of batch.records) { const bytes = Uint8Array.fromBase64(record.content) await processOrder(JSON.parse(decoder.decode(bytes))) } // Mark the batch processed and move the subscription forward await post(`/batches/${batch.batch_id}/ack`, { worker_id: WORKER_ID }) } catch { // Release the lease so the records are delivered again await post(`/batches/${batch.batch_id}/nack`, { worker_id: WORKER_ID }) }}How K2 works
Create a stream
Create a stream from the dashboard, Wrangler, or the REST API, and set the retention period.
Produce records
Append batches of records from a Workers binding or over HTTP. Once a batch is accepted, it is stored durably.
Subscribe
Create a subscription to track progress. Share one subscription across many consumers to split the work, or create several to fan out.
Consume and acknowledge
Consumers lease a batch of records and acknowledge it once processed. Records that fail processing, or whose lease expires, are delivered again, giving you at-least-once delivery.
Real-time analytics pipelines
Microservice event buses
Build without boundaries
Join thousands of developers who've eliminated infrastructure complexity and deployed globally with Cloudflare. Start building for free — no credit card required.