feat: add AbortSignal support to all blocking operations

Author: NathanFlurry

CLAUDE.md

@@ -11,6 +11,10 @@ Small utilities for Rust/Tokio-like primitives in TypeScript. Zero overhead, no
 - Use single-line conventional commit messages. No co-authors.
 - Example: `git commit -m "feat(sync): add broadcast channel"`
 
+## Cancellation: AbortSignal vs Drop
+
+In Rust, cancellation is implicit: dropping a future cancels it. In JavaScript, promises cannot be "dropped"; they run to completion unless the underlying operation checks an `AbortSignal`. Any function in antiox that wraps or races user-provided async work must accept an optional `AbortSignal` so callers can cancel it, mirroring how Rust callers would simply drop the future.
+
 ## API Reference
 
 Before implementing or modifying any module, always look up the corresponding Tokio or Rust std docs first to verify correct type names, method signatures, and semantics:
@@ -58,7 +62,7 @@ pnpm test        # Run tests
 
 ## Documentation
 
-- Keep `README.md` up to date when adding or changing modules.
+- Keep `README.md` up to date when adding or changing modules. Code examples in README must reflect the current API signatures.
 - Keep `COMPATIBILITY.md` up to date when adding, removing, or modifying any API. Every exported function, class, and method must be tracked against its Rust equivalent. Mark new items as implemented, and document why anything is intentionally skipped.
 
 ## Code Style

README.md

@@ -458,9 +458,9 @@ handle.abort();
 
 // JoinSet for managing multiple tasks
 const set = new JoinSet<number>();
-set.spawn(async () => 1);
-set.spawn(async () => 2);
-set.spawn(async () => 3);
+set.spawn(async (signal) => 1);
+set.spawn(async (signal) => 2);
+set.spawn(async (signal) => 3);
 
 for await (const result of set) {
   console.log(result); // 1, 2, 3 (in completion order)
@@ -488,7 +488,9 @@ try {
   if (e instanceof TimeoutError) console.log("timed out");
 }
 
-for await (const tick of interval(1000)) {
+// All functions accept an optional AbortSignal for cancellation
+const controller = new AbortController();
+for await (const tick of interval(1000, controller.signal)) {
   console.log(`Tick ${tick}`);
   if (tick >= 4) break;
 }

src/sync/barrier.ts

@@ -14,6 +14,7 @@ export class BarrierWaitResult {
 
 interface Waiter {
 	resolve: (result: BarrierWaitResult) => void;
+	aborted?: boolean;
 }
 
 // Reusable barrier: n tasks synchronize, then the barrier resets for the next generation.
@@ -29,14 +30,16 @@ export class Barrier {
 		this.#n = n;
 	}
 
-	wait(): Promise<BarrierWaitResult> {
+	wait(signal?: AbortSignal): Promise<BarrierWaitResult> {
+		signal?.throwIfAborted();
 		this.#count++;
 
 		if (this.#count === this.#n) {
 			const result = new BarrierWaitResult(true);
 
 			while (!this.#waiters.isEmpty()) {
-				this.#waiters.shift()!.resolve(new BarrierWaitResult(false));
+				const w = this.#waiters.shift()!;
+				if (!w.aborted) w.resolve(new BarrierWaitResult(false));
 			}
 
 			this.#count = 0;
@@ -45,8 +48,20 @@ export class Barrier {
 			return Promise.resolve(result);
 		}
 
-		return new Promise<BarrierWaitResult>((resolve) => {
-			this.#waiters.push({ resolve });
+		return new Promise<BarrierWaitResult>((resolve, reject) => {
+			const waiter: Waiter = { resolve };
+			this.#waiters.push(waiter);
+			if (signal) {
+				const gen = this.#generation;
+				const onAbort = () => {
+					waiter.aborted = true;
+					// Only decrement if still in the same generation
+					if (this.#generation === gen) this.#count--;
+					reject(signal.reason);
+				};
+				signal.addEventListener("abort", onAbort, { once: true });
+				waiter.resolve = (v) => { signal.removeEventListener("abort", onAbort); resolve(v); };
+			}
 		});
 	}
 }

src/sync/broadcast.ts

@@ -126,7 +126,8 @@ export class BroadcastReceiver<T> {
 		this.#cursor = cursor;
 	}
 
-	recv(): Promise<T> {
+	recv(signal?: AbortSignal): Promise<T> {
+		signal?.throwIfAborted();
 		if (this.#cursor < this.#state.writePos - this.#state.capacity) {
 			const missed = this.#state.writePos - this.#state.capacity - this.#cursor;
 			this.#cursor = this.#state.writePos - this.#state.capacity;
@@ -145,6 +146,7 @@ export class BroadcastReceiver<T> {
 		}
 
 		return new Promise<T>((resolve, reject) => {
+			const cursor = this.#cursor;
 			const waiter: Waiter<T> = {
 				resolve: (value: T) => {
 					this.#cursor++;
@@ -153,12 +155,27 @@ export class BroadcastReceiver<T> {
 				reject,
 			};
 
-			let set = this.#state.waiters.get(this.#cursor);
+			let set = this.#state.waiters.get(cursor);
 			if (!set) {
 				set = new Set();
-				this.#state.waiters.set(this.#cursor, set);
+				this.#state.waiters.set(cursor, set);
 			}
 			set.add(waiter);
+
+			if (signal) {
+				const onAbort = () => {
+					const s = this.#state.waiters.get(cursor);
+					if (s) {
+						s.delete(waiter);
+						if (s.size === 0) this.#state.waiters.delete(cursor);
+					}
+					reject(signal.reason);
+				};
+				signal.addEventListener("abort", onAbort, { once: true });
+				const origResolve = waiter.resolve;
+				waiter.resolve = (v) => { signal.removeEventListener("abort", onAbort); origResolve(v); };
+				waiter.reject = (e) => { signal.removeEventListener("abort", onAbort); reject(e); };
+			}
 		});
 	}
 

src/sync/cancellation_token.ts

@@ -20,13 +20,26 @@ export class CancellationToken {
 		return this.#controller.signal.aborted;
 	}
 
-	cancelled(): Promise<void> {
+	cancelled(signal?: AbortSignal): Promise<void> {
+		signal?.throwIfAborted();
 		if (this.#controller.signal.aborted) {
 			return Promise.resolve();
 		}
 
-		return new Promise<void>((resolve) => {
-			this.#controller.signal.addEventListener("abort", () => resolve(), { once: true });
+		return new Promise<void>((resolve, reject) => {
+			const onCancel = () => {
+				if (signal) signal.removeEventListener("abort", onAbort!);
+				resolve();
+			};
+			let onAbort: (() => void) | undefined;
+			if (signal) {
+				onAbort = () => {
+					this.#controller.signal.removeEventListener("abort", onCancel);
+					reject(signal.reason);
+				};
+				signal.addEventListener("abort", onAbort, { once: true });
+			}
+			this.#controller.signal.addEventListener("abort", onCancel, { once: true });
 		});
 	}