feat(sockets): listener delivery strategies for throttling and batching

.claude/skills/fix-flaky-tests/SKILL.md

@@ -0,0 +1,70 @@
+---
+name: fix-flaky-tests
+description: Find, reproduce, root-cause and permanently fix flaky (intermittently failing) tests in the hyper-fetch monorepo. Use proactively whenever a test fails once and passes on re-run, when CI is red and then green without code changes, when a `waitFor` times out around 1000ms, when a test depends on sleeps or default mock delays, or when a user says "flaky", "intermittent", "timeout", "works locally", "fix flake". Also use after adding new async tests to prove they are deterministic before committing.
+---
+
+# Fix Flaky Tests
+
+A flaky test is a test with a hidden dependency on time, order or shared state. Never "fix" it by re-running, retrying,
+widening a timeout, or marking it skipped. Find the dependency and remove it.
+
+## Workflow
+
+1. **Reproduce, don't guess.** Run the single spec repeatedly and shuffled before reading any code:
+
+   ```bash
+   .claude/skills/fix-flaky-tests/scripts/repeat-test.sh packages/react __tests__/features/use-submit/use-submit.queue.spec.ts 10
+   ```
+
+   The script runs the file N times (default 10) with `pnpm exec vitest run` inside the package and prints the failure
+   rate. Add `--shuffle` to also randomize test order (`--sequence.shuffle`). A 0/10 result after a fix is the minimum
+   bar; prefer 20 runs for timing-related fixes.
+
+2. **Confirm it is pre-existing or new.** If the failing test is unrelated to the current change, check it on `main`
+   with `git stash push -u` → run → `git stash pop`. Pre-existing flakes still get fixed in the same branch (small,
+   separate commit) - leaving them costs everyone time.
+
+3. **Read the failure, not the test name.** The message tells you the class of flake:
+   - `waitFor` / `testLoading` timing out around `1000ms` → the awaited state was already gone or never reached (race).
+   - "Calling the test function inside another test function" → structural (misplaced braces), not timing.
+   - `act` warnings / "state update on unmounted component" → missing cleanup or delivery after unsubscribe.
+   - Different failures each run → shared mutable state between tests.
+   - Fails only in the full suite, passes alone → order dependence or leaked globals / timers / servers.
+
+4. **Classify the root cause** (see table) and apply the matching fix pattern. Fix the cause, keep the assertion.
+
+5. **Verify**: repeat-run the spec (10-20x), run the package suite once, run `pnpm typecheck` in the package.
+
+6. **Commit with the cause in the message**, e.g. `test(react): make useSubmit queue spec deterministic - mock resolved
+   before second hook mounted`.
+
+## Root causes and fixes in this repo
+
+| Symptom | Root cause | Fix |
+| --- | --- | --- |
+| Second hook/instance expected to see an in-flight request but sees finished | `mockRequest(request)` default delay is **20ms** (`packages/testing/src/http/http.mock.ts`) - request finishes before the next render/mount | Pass an explicit `delay` (e.g. `{ delay: 500 }`) so the window is wide enough; never rely on the default |
+| Assertion right after an async emit | Treating async delivery as sync | Wrap in `await waitFor(() => ...)`; for sockets use `waitForConnection(socket)` before emitting |
+| Test uses `await sleep(n)` and asserts a count | Real-timer race | Use `vi.useFakeTimers({ toFake: ["setTimeout", "clearTimeout"] })` + `vi.advanceTimersByTime`, or `waitFor`. Keep `Date`/promises real so mock servers still connect |
+| Fake timers break connection setup | mock-socket / msw use timers to connect | Connect with real timers first (`await waitForConnection`), then switch to fake timers inside the test or `beforeEach` |
+| Passes alone, fails in suite | Leaked state: `client` cache/dispatchers, socket listeners, servers, spies | Reset in `beforeEach` (`client.clear()`, `vi.resetAllMocks()`), stop servers in `afterEach/afterAll`, `vi.useRealTimers()` in `afterEach` |
+| Random test fails each run | Shared module-level fixture mutated by tests | Create fixtures inside `beforeEach`; use unique topics/keys (`getUniqueRequestId`) |
+| Only fails on CI | Slower machine amplifies a race | Same fixes as above; never just raise `testTimeout` |
+| Timer-based delivery (throttle/batch/debounce) asserts after unsubscribe | Pending timer fires after cleanup | Assert nothing is delivered after dispose; ensure the implementation clears timers on unsubscribe |
+
+## Rules
+
+- A flake found while working on something else is still fixed, in its own small commit. Do not leave `it.skip`.
+- The fix must make the test deterministic, not merely rarer. If you cannot explain *why* it failed, you have not fixed it.
+- Prefer controlling time (fake timers, explicit mock delays) over waiting longer.
+- Prefer `waitFor` over `sleep`. Prefer asserting on events over asserting on elapsed time.
+- When you touch a flaky spec, also scan its sibling specs for the same pattern (same helper, same default delay) and
+  fix them in the same commit.
+- After fixing, run the whole package suite once - some flakes hide others.
+
+## Monorepo notes
+
+- Tests are Vitest; packages expose `pnpm test` and `pnpm typecheck`. Run a single file with
+  `pnpm exec vitest run <path>` from the package directory.
+- `@hyper-fetch/testing` helpers: `createHttpMockingServer` (`mockRequest(request, { delay, data, status })`),
+  `createWebsocketMockingServer` (`emitListenerEvent`), `createSseMockingServer`, `waitForConnection`, `sleep`.
+- Dependents resolve sibling packages through `dist/` - if types look stale, `pnpm build` the dependency first.

.claude/skills/fix-flaky-tests/scripts/repeat-test.sh

@@ -0,0 +1,27 @@
+#!/usr/bin/env bash
+# Run one Vitest spec file N times inside a package and report the failure rate.
+# Usage: repeat-test.sh <package-dir> <spec-path-relative-to-package> [runs=10] [--shuffle]
+set -u
+PKG_DIR="${1:?package dir required, e.g. packages/react}"
+SPEC="${2:?spec path required}"
+RUNS="${3:-10}"
+SHUFFLE=""
+if [[ "${3:-}" == "--shuffle" || "${4:-}" == "--shuffle" ]]; then
+  SHUFFLE="--sequence.shuffle"
+  [[ "${3:-}" == "--shuffle" ]] && RUNS=10
+fi
+
+cd "$PKG_DIR" || exit 1
+fails=0
+for i in $(seq 1 "$RUNS"); do
+  if out=$(pnpm exec vitest run "$SPEC" $SHUFFLE 2>&1); then
+    printf "run %2d: pass\n" "$i"
+  else
+    fails=$((fails + 1))
+    printf "run %2d: FAIL\n" "$i"
+    echo "$out" | grep -E "×|FAIL|Error|expected|received|Timed out" | head -15 | sed 's/^/    /'
+  fi
+done
+echo "----"
+echo "failures: $fails / $RUNS ($SPEC)"
+[[ "$fails" -eq 0 ]]

documentation/docs/guides/react/02-sockets/listening.mdx

@@ -82,6 +82,9 @@ function LiveChat() {
 In the example above, `onChatMessage` is a pre-defined listener request. You can learn more about creating them in the
 [sockets guide](/docs/guides/sockets/listening).
 
+If the stream is fast and your component re-renders too often, configure a `delivery` strategy on the listener - see
+[Throttling & Batching in React](/docs/guides/react/sockets/throttling).
+
 ---
 
 :::success Congratulations!

documentation/docs/guides/react/02-sockets/throttling.mdx

@@ -0,0 +1,177 @@
+---
+sidebar_position: 3
+title: React - Socket Throttling & Batching
+sidebar_label: Throttling & Batching
+---
+
+# Throttling & Batching in React
+
+With `useListener`, every delivered message is a state update, and every state update is a render. On a stream that
+pushes hundreds of messages per second, that is hundreds of renders per second for a screen that refreshes sixty times.
+
+The fix lives on the listener, not in the component. Configure a `delivery` strategy once, and `useListener` simply
+receives fewer, better-shaped updates. No hook options, no manual throttling in effects.
+
+---
+
+:::secondary What you'll learn
+
+1.  How a **`latest`** listener turns a price firehose into one render per frame.
+2.  How a **`batch`** listener delivers arrays so you can append many rows in one state update.
+3.  How to let the user **choose the refresh rate** with `setDelivery` and `useMemo`.
+4.  The one **gotcha** with custom strategies and React re-renders.
+
+:::
+
+> Start with the framework-agnostic [Throttling & Batching guide](/docs/guides/sockets/throttling) if you have not
+> read it - this page focuses on the React side.
+
+---
+
+## Sampling a price ticker
+
+Define the listener with a `latest` strategy. `useListener` needs no changes: `data` is simply the newest price in the
+last window.
+
+```ts title="api/socket.ts"
+export const onPrice = socket.createListener<Price>()({
+  topic: "prices",
+  // highlight-next-line
+  delivery: { strategy: "latest", interval: 16 },
+});
+```
+
+```tsx title="PriceTicker.tsx"
+import { useListener } from "@hyper-fetch/react";
+import { onPrice } from "./api/socket";
+
+export function PriceTicker() {
+  // highlight-start
+  // Renders at most ~60 times per second regardless of how fast the server sends.
+  // highlight-end
+  const { data, connected } = useListener(onPrice);
+
+  if (!connected) return <span>Connecting…</span>;
+  return <span>{data ? data.value.toFixed(2) : "—"}</span>;
+}
+```
+
+---
+
+## Batching a trade log
+
+With `batch`, `data` and the `onEvent` payload are **arrays**, typed automatically as `Trade[]`. Append the whole batch
+in a single state update.
+
+```ts title="api/socket.ts"
+export const onTrade = socket.createListener<Trade>()({
+  topic: "trades",
+  // highlight-next-line
+  delivery: { strategy: "batch", interval: 100, maxSize: 500 },
+});
+```
+
+```tsx title="TradeLog.tsx"
+import { useState } from "react";
+import { useListener } from "@hyper-fetch/react";
+import { onTrade } from "./api/socket";
+
+export function TradeLog() {
+  const [trades, setTrades] = useState<Trade[]>([]);
+  const { onEvent } = useListener(onTrade);
+
+  onEvent(({ data }) => {
+    // highlight-start
+    // `data` is Trade[] - one setState per window instead of one per trade.
+    // highlight-end
+    setTrades((previous) => [...previous, ...data].slice(-1000));
+  });
+
+  return (
+    <ul>
+      {trades.map((trade) => (
+        <li key={trade.id}>
+          {trade.symbol} {trade.price}
+        </li>
+      ))}
+    </ul>
+  );
+}
+```
+
+---
+
+## Letting the user pick the refresh rate
+
+`setDelivery` returns a new listener. Memoize it on the user's choice and pass it to `useListener`; the hook
+re-subscribes when the delivery changes and disposes the previous window.
+
+```tsx title="PriceTicker.tsx"
+import { useMemo, useState } from "react";
+import { useListener } from "@hyper-fetch/react";
+import { onPrice } from "./api/socket";
+
+export function PriceTicker() {
+  const [rate, setRate] = useState(16);
+  // highlight-start
+  // A new listener only when `rate` changes - not on every render.
+  // highlight-end
+  const live = useMemo(() => onPrice.setDelivery({ strategy: "latest", interval: rate }), [rate]);
+  const { data } = useListener(live);
+
+  return (
+    <>
+      <select value={rate} onChange={(event) => setRate(Number(event.target.value))}>
+        <option value={16}>Realtime</option>
+        <option value={250}>Relaxed</option>
+        <option value={1000}>Slow</option>
+      </select>
+      <span>{data?.value ?? "—"}</span>
+    </>
+  );
+}
+```
+
+---
+
+## The gotcha: custom strategies and identity
+
+`useListener` decides whether to re-subscribe by comparing the listener's `topic`, `options` and `delivery`:
+
+- **Preset objects** (`{ strategy: "latest", interval: 16 }`) are compared **by value**. Writing them inline is fine.
+- **Custom strategy functions** are compared **by identity**. Define them at module level, next to your listeners. A
+  strategy created inside a component is a new function on every render and would re-subscribe each time, dropping
+  whatever was buffered.
+
+```ts title="api/socket.ts"
+// ✅ module level - stable identity
+export const summarize = createDeliveryStrategy<Trade, Summary>(/* ... */);
+export const onTradeSummary = socket.createListener<Trade>()({ topic: "trades", delivery: summarize });
+```
+
+---
+
+:::success Congratulations!
+
+You can now keep React UIs smooth on the busiest streams.
+
+- A **`latest`** listener caps renders at the rate you choose.
+- A **`batch`** listener gives you arrays and one state update per window.
+- **`setDelivery` + `useMemo`** lets the user control the rate.
+- Custom strategies live at **module level** to keep a stable identity.
+
+:::
+
+<LinkCard
+  type="docs"
+  title="useListener"
+  description="Full reference for the useListener hook."
+  to="/docs/react/sockets/use-listener"
+/>
+
+<LinkCard
+  type="docs"
+  title="Throttling & Batching"
+  description="The framework-agnostic guide with timelines, semantics and custom strategies."
+  to="/docs/guides/sockets/throttling"
+/>

documentation/docs/guides/sockets/emitting.mdx

@@ -1,5 +1,5 @@
 ---
-sidebar_position: 3
+sidebar_position: 4
 title: Guide - Socket Emitting
 sidebar_label: Emitting
 ---
@@ -36,7 +36,7 @@ type ChatMessageType = {
 //highlight-end
 
 const sendChatMessage = socketInstance.createEmitter<ChatMessageType>()({
-  endpoint: "chat-message", // topic of the event
+  topic: "chat-message", // topic of the event
 });
 ```
 
@@ -86,7 +86,7 @@ Let's create an emitter to send a private message to a specific user.
 
 ```typescript
 const sendPrivateMessage = socketInstance.createEmitter<{ message: string }>()({
-  endpoint: "private-message/:userId",
+  topic: "private-message/:userId",
 });
 ```
 
@@ -115,7 +115,7 @@ data, or encrypting messages:
 ```typescript
 const sendMessage = socketInstance
   .createEmitter<{ message: string }>()({
-    endpoint: "chat-message",
+    topic: "chat-message",
   })
   .setPayloadMapper((payload) => ({
     ...payload,

documentation/docs/guides/sockets/listening.mdx

@@ -25,7 +25,7 @@ that can be easily managed and integrated into your application, including React
 ## Creating a Listener
 
 To start listening for events, you first need to create a listener. This is done using the `socket.createListener`
-method, which requires an `endpoint` that corresponds to the event name you want to listen to. It's also a good practice
+method, which requires a `topic` that corresponds to the event name you want to listen to. It's also a good practice
 to provide a type for the expected message payload for type safety.
 
 ```ts
@@ -34,7 +34,7 @@ to provide a type for the expected message payload for type safety.
 // This helps to avoid re-creating listeners and attaching multiple handlers for the same event.
 // highlight-end
 const onChatMessage = socketInstance.createListener<{ message: string }>()({
-  endpoint: "chat-message",
+  topic: "chat-message",
 });
 ```
 
@@ -44,7 +44,7 @@ received from this event will be correctly typed.
 ## Starting and Stopping
 
 Once you have a listener, you can start listening for events by calling the `.listen()` method with a callback function.
-This function will be executed every time a new message arrives on the specified endpoint.
+This function will be executed every time a new message arrives on the specified topic.
 
 The `.listen()` method returns an `unlisten` function, which you can call to stop listening for events. It's crucial to
 call this cleanup function when the listener is no longer needed to prevent memory leaks.
@@ -127,6 +127,22 @@ unlistenTwo();
 unlistenOne();
 ```
 
+## Dealing with High-Frequency Streams
+
+If a topic emits hundreds of messages per second, you rarely want every one of them to reach your callbacks. The
+`delivery` option lets a listener sample (`latest`) or batch (`batch`) messages before they are delivered, while
+`listen()` and `useListener` keep working unchanged:
+
+```ts
+const onPrice = socketInstance.createListener<Price>()({
+  topic: "prices",
+  delivery: { strategy: "latest", interval: 16 },
+});
+```
+
+Read the [Throttling & Batching guide](/docs/guides/sockets/throttling) for timelines, trade-offs and custom
+strategies.
+
 ---
 
 :::success Congratulations!

documentation/docs/guides/sockets/setup.mdx

@@ -46,7 +46,7 @@ type ChatMessageType = {
 };
 
 export const onChatMessage = socket.createListener<ChatMessageType>()({
-  endpoint: "chat-message", // endpoint of the event
+  topic: "chat-message", // topic of the event
 });
 ```
 
@@ -70,7 +70,7 @@ type AcknowledgementResponseType = {
 };
 
 export const sendChatMessage = socket.createEmitter<ChatMessageType, AcknowledgementResponseType>()({
-  endpoint: "chat-message", // endpoint of the event
+  topic: "chat-message", // topic of the event
 });
 ```