|
| 1 | +--- |
| 2 | +title: Distributed Abort Controller |
| 3 | +description: A distributed AbortController that uses durable workflows for cross-process cancellation signaling. |
| 4 | +type: guide |
| 5 | +summary: Build a distributed abort controller that uses workflow streams and hooks to propagate cancellation signals across process boundaries. |
| 6 | +--- |
| 7 | + |
| 8 | +Use this pattern when you need an `AbortController`-like interface that works across distributed systems. The controller uses a durable workflow to coordinate cancellation — calling `.abort()` on one machine triggers the `.signal` on any other machine. |
| 9 | + |
| 10 | +## When to use this |
| 11 | + |
| 12 | +- **Cross-process cancellation** — Cancel a long-running operation from a different server, worker, or edge function |
| 13 | +- **Durable cancellation** — The abort signal persists even if the process that created it crashes |
| 14 | +- **UI stop buttons** — Let users cancel operations running on the server from the browser |
| 15 | +- **Timeout coordination** — The built-in TTL auto-expires stale controllers |
| 16 | + |
| 17 | +## Pattern |
| 18 | + |
| 19 | +The `DistributedAbortController` class encapsulates a workflow that: |
| 20 | +1. Accepts a user-provided unique ID (like a chat ID or task ID) |
| 21 | +2. Creates or reconnects to an existing workflow using that ID |
| 22 | +3. Waits for a hook signal OR TTL expiration |
| 23 | +4. Writes a cancellation message to the run's stream when triggered |
| 24 | + |
| 25 | +### Core Implementation |
| 26 | + |
| 27 | +```typescript lineNumbers |
| 28 | +import { defineHook, getWritable, sleep } from "workflow"; |
| 29 | +import { start, getRun, getHookByToken } from "workflow/api"; |
| 30 | + |
| 31 | +// Default TTL: 24 hours |
| 32 | +const DEFAULT_TTL_MS = 24 * 60 * 60 * 1000; |
| 33 | +// Default grace period: 1 hour (keeps hook alive after abort for late subscribers) |
| 34 | +const DEFAULT_GRACE_MS = 60 * 60 * 1000; |
| 35 | + |
| 36 | +// Hook to trigger the abort signal |
| 37 | +export const abortHook = defineHook<{ reason?: string }>(); |
| 38 | + |
| 39 | +// The abort message written to the stream |
| 40 | +export type AbortMessage = { |
| 41 | + type: "abort"; |
| 42 | + reason?: string; |
| 43 | + expired?: boolean; |
| 44 | +}; |
| 45 | + |
| 46 | +// Helper to create a consistent hook token from the user ID |
| 47 | +function getAbortToken(id: string): string { |
| 48 | + return `abort:${id}`; |
| 49 | +} |
| 50 | + |
| 51 | +// Step function that writes the abort message to the stream |
| 52 | +async function writeAbortSignal(reason?: string, expired?: boolean) { |
| 53 | + "use step"; |
| 54 | + |
| 55 | + const writable = getWritable<AbortMessage>(); |
| 56 | + const writer = writable.getWriter(); |
| 57 | + try { |
| 58 | + await writer.write({ type: "abort", reason, expired }); |
| 59 | + } finally { |
| 60 | + writer.releaseLock(); |
| 61 | + } |
| 62 | + await writable.close(); |
| 63 | +} |
| 64 | + |
| 65 | +// Workflow that waits for abort or TTL expiration |
| 66 | +export async function abortControllerWorkflow( |
| 67 | + id: string, |
| 68 | + ttlMs: number, |
| 69 | + graceMs: number |
| 70 | +) { |
| 71 | + "use workflow"; |
| 72 | + |
| 73 | + const startTime = Date.now(); |
| 74 | + const hook = abortHook.create({ token: getAbortToken(id) }); |
| 75 | + |
| 76 | + // Race: manual abort OR TTL expiration // [!code highlight] |
| 77 | + const result = await Promise.race([ |
| 78 | + hook.then((payload) => ({ |
| 79 | + reason: payload.reason, |
| 80 | + expired: false, |
| 81 | + })), |
| 82 | + sleep(`${ttlMs}ms`).then(() => ({ |
| 83 | + reason: "Controller expired", |
| 84 | + expired: true, |
| 85 | + })), |
| 86 | + ]); |
| 87 | + |
| 88 | + await writeAbortSignal(result.reason, result.expired); |
| 89 | + |
| 90 | + // Only sleep through grace period on TTL expiration (keeps hook alive for late subscribers). // [!code highlight] |
| 91 | + // Manual aborts complete immediately. |
| 92 | + if (result.expired) { |
| 93 | + const elapsed = Date.now() - startTime; |
| 94 | + const remainingTime = graceMs - (elapsed - ttlMs); |
| 95 | + if (remainingTime > 0) { |
| 96 | + await sleep(`${remainingTime}ms`); // [!code highlight] |
| 97 | + } |
| 98 | + } |
| 99 | + |
| 100 | + return { aborted: true, reason: result.reason, expired: result.expired }; |
| 101 | +} |
| 102 | + |
| 103 | +/** |
| 104 | + * A distributed abort controller that works across process boundaries. |
| 105 | + * Uses a semantically meaningful ID (like a chat ID or task ID) to coordinate. |
| 106 | + */ |
| 107 | +export class DistributedAbortController { |
| 108 | + private id: string; |
| 109 | + readonly runId: string; |
| 110 | + |
| 111 | + private constructor(id: string, runId: string) { |
| 112 | + this.id = id; |
| 113 | + this.runId = runId; |
| 114 | + } |
| 115 | + |
| 116 | + /** |
| 117 | + * Creates or reconnects to a distributed abort controller. |
| 118 | + * If a controller with this ID already exists, reconnects to it. |
| 119 | + * Otherwise, starts a new workflow. |
| 120 | + * |
| 121 | + * @param id - A unique, semantically meaningful ID (e.g., "chat:123") |
| 122 | + * @param options.ttlMs - Time-to-live in ms (default: 24 hours) |
| 123 | + * @param options.graceMs - Grace period after abort (default: 1 hour) |
| 124 | + */ |
| 125 | + static async create( // [!code highlight] |
| 126 | + id: string, |
| 127 | + options: { ttlMs?: number; graceMs?: number } = {} |
| 128 | + ): Promise<DistributedAbortController> { |
| 129 | + const { ttlMs = DEFAULT_TTL_MS, graceMs = DEFAULT_GRACE_MS } = options; |
| 130 | + const token = getAbortToken(id); |
| 131 | + |
| 132 | + // Try to find an existing run with this hook token |
| 133 | + const existingHook = await getHookByToken(token).catch(() => null); // [!code highlight] |
| 134 | + |
| 135 | + if (existingHook) { |
| 136 | + // Reconnect to existing controller |
| 137 | + return new DistributedAbortController(id, existingHook.runId); |
| 138 | + } |
| 139 | + |
| 140 | + // Create a new workflow |
| 141 | + const run = await start(abortControllerWorkflow, [id, ttlMs, graceMs]); // [!code highlight] |
| 142 | + return new DistributedAbortController(id, run.runId); |
| 143 | + } |
| 144 | + |
| 145 | + /** |
| 146 | + * Triggers the abort signal. |
| 147 | + * Idempotent: safe to call multiple times or after the workflow has completed. |
| 148 | + */ |
| 149 | + async abort(reason?: string): Promise<void> { // [!code highlight] |
| 150 | + try { |
| 151 | + await abortHook.resume(getAbortToken(this.id), { reason }); |
| 152 | + } catch (error) { |
| 153 | + const msg = error instanceof Error ? error.message.toLowerCase() : ''; |
| 154 | + if (msg.includes('not found') || msg.includes('expired')) { |
| 155 | + return; |
| 156 | + } |
| 157 | + throw error; |
| 158 | + } |
| 159 | + } |
| 160 | + |
| 161 | + /** |
| 162 | + * Returns an AbortSignal that fires when abort() is called or TTL expires. |
| 163 | + * The signal fires with a reason indicating what triggered it. |
| 164 | + */ |
| 165 | + get signal(): AbortSignal { // [!code highlight] |
| 166 | + const run = getRun<{ aborted: boolean; reason?: string; expired?: boolean }>(this.runId); |
| 167 | + const controller = new AbortController(); |
| 168 | + const readable = run.getReadable<AbortMessage>(); |
| 169 | + |
| 170 | + (async () => { |
| 171 | + const reader = readable.getReader(); |
| 172 | + try { |
| 173 | + while (true) { |
| 174 | + const { done, value } = await reader.read(); |
| 175 | + if (done) break; |
| 176 | + if (value.type === "abort") { |
| 177 | + const reason = value.expired |
| 178 | + ? `${value.reason} (expired)` |
| 179 | + : value.reason; |
| 180 | + controller.abort(reason); |
| 181 | + break; |
| 182 | + } |
| 183 | + } |
| 184 | + } catch (error) { |
| 185 | + if (!controller.signal.aborted) { |
| 186 | + controller.abort( |
| 187 | + error instanceof Error ? error.message : "Stream read failed" |
| 188 | + ); |
| 189 | + } |
| 190 | + } finally { |
| 191 | + reader.releaseLock(); |
| 192 | + } |
| 193 | + })(); |
| 194 | + |
| 195 | + return controller.signal; |
| 196 | + } |
| 197 | +} |
| 198 | +``` |
| 199 | + |
| 200 | +### Usage: Single Process |
| 201 | + |
| 202 | +```typescript lineNumbers |
| 203 | +import { DistributedAbortController } from "./distributed-abort-controller"; |
| 204 | + |
| 205 | +// Create a controller with a meaningful ID |
| 206 | +const controller = await DistributedAbortController.create("chat:user-123"); |
| 207 | + |
| 208 | +// Get the signal and use it with fetch |
| 209 | +const signal = controller.signal; |
| 210 | +const response = await fetch("https://api.example.com/long-operation", { |
| 211 | + signal, |
| 212 | +}); |
| 213 | + |
| 214 | +// Later: abort the operation |
| 215 | +await controller.abort("User cancelled"); |
| 216 | +``` |
| 217 | + |
| 218 | +### Usage: Cross-Process Coordination |
| 219 | + |
| 220 | +```typescript lineNumbers |
| 221 | +import { DistributedAbortController } from "./distributed-abort-controller"; |
| 222 | + |
| 223 | +// Process A: Create the controller |
| 224 | +const controller = await DistributedAbortController.create("task:build-123"); |
| 225 | +// start long operation using controller.signal... |
| 226 | + |
| 227 | +// Process B: Reconnect and abort (no run ID sharing needed!) |
| 228 | +const sameController = await DistributedAbortController.create("task:build-123"); // [!code highlight] |
| 229 | +await sameController.abort("Cancelled by admin"); |
| 230 | + |
| 231 | +// Process C: Reconnect and listen |
| 232 | +const anotherRef = await DistributedAbortController.create("task:build-123"); |
| 233 | +anotherRef.signal.addEventListener("abort", (e) => { |
| 234 | + console.log("Task was cancelled:", (e.target as AbortSignal).reason); |
| 235 | +}); |
| 236 | +``` |
| 237 | + |
| 238 | +### Custom TTL |
| 239 | + |
| 240 | +```typescript lineNumbers |
| 241 | +import { DistributedAbortController } from "./distributed-abort-controller"; |
| 242 | + |
| 243 | +// Short-lived controller for a quick operation (5 minutes) |
| 244 | +const shortLived = await DistributedAbortController.create("quick-task", { |
| 245 | + ttlMs: 5 * 60 * 1000, |
| 246 | +}); |
| 247 | + |
| 248 | +// Long-lived controller for batch jobs (7 days) |
| 249 | +const longLived = await DistributedAbortController.create("batch-job", { |
| 250 | + ttlMs: 7 * 24 * 60 * 60 * 1000, |
| 251 | +}); |
| 252 | + |
| 253 | +// When TTL expires, the signal fires with expired reason |
| 254 | +shortLived.signal.addEventListener("abort", (e) => { |
| 255 | + const reason = (e.target as AbortSignal).reason; |
| 256 | + if (reason?.includes("expired")) { |
| 257 | + console.log("Controller expired, cleaning up..."); |
| 258 | + } |
| 259 | +}); |
| 260 | +``` |
| 261 | + |
| 262 | +### API Route for Remote Abort |
| 263 | + |
| 264 | +```typescript lineNumbers |
| 265 | +import { DistributedAbortController } from "@/lib/distributed-abort-controller"; |
| 266 | + |
| 267 | +export async function POST( |
| 268 | + request: Request, |
| 269 | + { params }: { params: Promise<{ id: string }> } |
| 270 | +) { |
| 271 | + const { id } = await params; |
| 272 | + const { reason } = await request.json(); |
| 273 | + |
| 274 | + const controller = await DistributedAbortController.create(id); |
| 275 | + await controller.abort(reason || "Cancelled via API"); |
| 276 | + |
| 277 | + return Response.json({ success: true }); |
| 278 | +} |
| 279 | +``` |
| 280 | + |
| 281 | +### Client Cancel Button |
| 282 | + |
| 283 | +```tsx lineNumbers |
| 284 | +"use client"; |
| 285 | + |
| 286 | +export function CancelButton({ taskId }: { taskId: string }) { |
| 287 | + const handleCancel = async () => { |
| 288 | + await fetch(`/api/abort/${taskId}`, { |
| 289 | + method: "POST", |
| 290 | + headers: { "Content-Type": "application/json" }, |
| 291 | + body: JSON.stringify({ reason: "User clicked cancel" }), |
| 292 | + }); |
| 293 | + }; |
| 294 | + |
| 295 | + return ( |
| 296 | + <button type="button" onClick={handleCancel}> |
| 297 | + Cancel Operation |
| 298 | + </button> |
| 299 | + ); |
| 300 | +} |
| 301 | +``` |
| 302 | + |
| 303 | +## Tips |
| 304 | + |
| 305 | +- **Use semantic IDs** — Use meaningful IDs like `chat:123` or `task:abc` instead of random UUIDs |
| 306 | +- **Create is idempotent** — Calling `create()` with the same ID reconnects to the existing controller |
| 307 | +- **TTL auto-cleanup** — Workflows self-terminate after TTL expires; no manual cleanup needed |
| 308 | +- **Signal is a getter** — Each access to `.signal` creates a new listener; cache it if needed |
| 309 | +- **One-shot** — Once aborted or expired, the workflow completes; create a new controller for new operations |
| 310 | + |
| 311 | +## Key APIs |
| 312 | + |
| 313 | +- [`defineHook()`](/docs/api-reference/workflow/define-hook) — type-safe hook for the abort trigger |
| 314 | +- [`getWritable()`](/docs/api-reference/workflow/get-writable) — write abort messages to the stream |
| 315 | +- [`sleep()`](/docs/api-reference/workflow/sleep) — TTL timer for auto-expiration |
| 316 | +- [`start()`](/docs/api-reference/workflow-api/start) — start the abort controller workflow |
| 317 | +- [`getHookByToken()`](/docs/api-reference/workflow-api/get-hook-by-token) — find existing run by hook token |
| 318 | +- [`getRun()`](/docs/api-reference/workflow-api/get-run) — reconnect to the workflow's readable stream |
0 commit comments