200 lines
9.5 KiB
Markdown
200 lines
9.5 KiB
Markdown
---
|
|
name: cyclone-tail
|
|
description: "Cyclone live-tail streaming wire format and the useTailStream / useMergedTail hook triplet. Use when: adding a new streaming list page, changing the wire format, debugging stalled/reconnecting state, or modifying the StatusPill behavior."
|
|
---
|
|
|
|
# cyclone-tail
|
|
|
|
Cyclone keeps the Claims, Remittances, and Activity pages live without
|
|
polling: every store write publishes an internal EventBus event, the
|
|
page opens a `GET /api/<resource>/stream` HTTP/1.1 chunked-NDJSON
|
|
connection, and new rows land in the table the moment they hit the
|
|
database. This skill codifies the wire format, the hook triplet, and
|
|
the backoff/stall machinery so additions stay consistent with the
|
|
three streaming pages already shipped (`Claims`, `Remittances`,
|
|
`ActivityLog`).
|
|
|
|
## When to use
|
|
|
|
- **Adding a new streaming page.** Mounting `useTailStream(resource)`
|
|
on a page — you need the hook triplet shape (initial fetch +
|
|
`useTailStream` + `useMergedTail`), the `<TailStatusPill>` wiring,
|
|
and the dedup rules in `useMergedTail`.
|
|
- **Changing the wire format.** Adding a new event type — update
|
|
`TailEvent` in `src/lib/tail-stream.ts:22-44`, the dispatch switch
|
|
in `src/hooks/useTailStream.ts:173-195`, the emitter in
|
|
`backend/src/cyclone/api_helpers.py:tail_events()`, and
|
|
`references/wire-format.md`.
|
|
- **Debugging stalled/reconnecting state.** Confirm whether the
|
|
backend is heartbeating (`CYCLONE_TAIL_HEARTBEAT_S`, default `15s`)
|
|
or the stall timer fired (`STALL_TIMEOUT_MS = 30_000` at
|
|
`src/hooks/useTailStream.ts:53`).
|
|
- **Tuning heartbeat/stall timing.** Changing the 30s stall threshold
|
|
or the 15s heartbeat interval — README's "Status pill" + "Knobs"
|
|
tables need to stay in sync (`README.md:109-129`).
|
|
|
|
## Conventions
|
|
|
|
1. **Wire format.** Newline-delimited JSON. Every line is
|
|
`{"type": ..., "data": ...}`. Known `type` values: `item`
|
|
(per-row envelope), `snapshot_end` (`{"count": N}` marker after the
|
|
snapshot), `heartbeat` (`{"ts": "<iso-8601>"}` keep-alive),
|
|
`item_dropped` (`{"id": "..."}` queue-overflow notice), `error`
|
|
(`{"message": "..."}` promoted to a thrown error by the hook).
|
|
Defined at `src/lib/tail-stream.ts:22-44`; emitted by
|
|
`backend/src/cyclone/api.py:1357-2006`. See
|
|
`references/wire-format.md`.
|
|
2. **Hook triplet.** Streaming pages compose three pieces:
|
|
`useTailStream(resource)` (`src/hooks/useTailStream.ts:80` — opens
|
|
the stream, drives the backoff/stall state machine, dispatches
|
|
`item` events into `useTailStore`),
|
|
`useMergedTail(resource, baseItems, filterFn?)`
|
|
(`src/hooks/useMergedTail.ts:25` — returns
|
|
`baseItems + tailSlice` dedup'd against `baseItems`), and a
|
|
per-resource initial-fetch hook (`useClaims`, `useRemittances`,
|
|
`useActivity`). The page wires all three — see
|
|
`src/pages/Claims.tsx:87-89`.
|
|
3. **Stall threshold.** 30 seconds of total silence — heartbeats
|
|
included — flips status to `stalled` and surfaces the
|
|
`↻ Reconnect` button on `<TailStatusPill>`. Constant:
|
|
`STALL_TIMEOUT_MS = 30_000` at `src/hooks/useTailStream.ts:53`.
|
|
Re-armed on every event including `heartbeat` and `item_dropped`
|
|
(`useTailStream.ts:124-140`). Don't change without updating
|
|
`README.md:109-123`.
|
|
4. **Snapshot first.** Every stream emits the snapshot before any
|
|
live `item` events, then closes with exactly one `snapshot_end`
|
|
carrying `{"count": N}`. The hook uses `snapshot_end` to flip
|
|
`<TailStatusPill>` from `connecting` to `live` and to reset the
|
|
reconnect backoff counter (`useTailStream.ts:174-180`).
|
|
5. **Content-Type.** Stream endpoints respond with
|
|
`media_type="application/x-ndjson"` — see
|
|
`backend/src/cyclone/api.py:1401,1894,2005`. Frontend sets
|
|
`Accept: application/x-ndjson` at `src/lib/tail-stream.ts:67-70`.
|
|
Never `application/json` for a stream endpoint.
|
|
6. **Backoff.** Transient errors retry with `1s → 2s → 4s → 8s → 16s
|
|
→ 30s` capped — `BACKOFF_STEPS_MS` at
|
|
`src/hooks/useTailStream.ts:48-50`. Counter resets on every
|
|
`snapshot_end`.
|
|
7. **Heartbeat knob.** Idle heartbeat interval is configurable via
|
|
`CYCLONE_TAIL_HEARTBEAT_S` (default `15`, parsed at call time in
|
|
`backend/src/cyclone/api_helpers.py:185-198`). Tests override to
|
|
a small value to keep runtime bounded.
|
|
|
|
## Patterns
|
|
|
|
### Page-hook skeleton — `Claims.tsx`
|
|
|
|
Pattern from `src/pages/Claims.tsx:85-90`. Three hooks in order:
|
|
`useClaims(params)` (initial TanStack Query fetch),
|
|
`useTailStream("claims")` (opens the stream, owns status state), and
|
|
`useMergedTail("claims", data?.items ?? [], tailFilterFn)` (combines
|
|
initial snapshot with live tail, dedup'd by id, filtered by the page's
|
|
predicate applied AFTER dedup at `useMergedTail.ts:72-74`).
|
|
|
|
```ts
|
|
import { useClaims } from "@/hooks/useClaims";
|
|
import { useTailStream } from "@/hooks/useTailStream";
|
|
import { useMergedTail } from "@/hooks/useMergedTail";
|
|
|
|
export function ClaimsPage() {
|
|
const { data } = useClaims({ status: "submitted" });
|
|
const { status, lastEventAt, forceReconnect } = useTailStream("claims");
|
|
const tailFilterFn = (c: Claim) => c.status === "submitted";
|
|
const items = useMergedTail("claims", data?.items ?? [], tailFilterFn);
|
|
// <TailStatusPill status={status} lastEventAt={lastEventAt}
|
|
// onReconnect={forceReconnect} /> + <Table items={items} />
|
|
}
|
|
```
|
|
|
|
### Backend `/api/foo/stream` endpoint
|
|
|
|
Pattern from `backend/src/cyclone/api.py:1357-1401`. Register BEFORE
|
|
`/api/foo/{foo_id}` so the literal `stream` segment doesn't match as
|
|
an id. Two phases — eager snapshot, then live subscription — wrapped
|
|
in `StreamingResponse` with `media_type="application/x-ndjson"`.
|
|
|
|
```python
|
|
@app.get("/api/foo/stream")
|
|
async def foo_stream(
|
|
request: Request,
|
|
status: str | None = Query(None),
|
|
limit: int = Query(100, ge=1, le=1000),
|
|
) -> StreamingResponse:
|
|
bus: EventBus = request.app.state.event_bus
|
|
|
|
async def gen() -> AsyncIterator[bytes]:
|
|
# 1. Snapshot.
|
|
rows = store.iter_foos(status=status, limit=limit)
|
|
for row in rows:
|
|
yield _ndjson_line({"type": "item", "data": row})
|
|
yield _ndjson_line({"type": "snapshot_end", "data": {"count": len(rows)}})
|
|
# 2. Live subscription + heartbeats.
|
|
async for chunk in _tail_events(request, bus, ["foo_written"]):
|
|
yield chunk
|
|
|
|
return StreamingResponse(gen(), media_type="application/x-ndjson")
|
|
```
|
|
|
|
`_ndjson_line` and `_tail_events` live in
|
|
`backend/src/cyclone/api_helpers.py`; the latter forwards EventBus
|
|
events as `item` lines and emits `heartbeat` lines on the cadence
|
|
from `heartbeat_seconds()`.
|
|
|
|
### `<TailStatusPill>` wiring
|
|
|
|
Pattern from `src/components/TailStatusPill.tsx:22-83`. The pill
|
|
takes `status` + `lastEventAt` from `useTailStream` plus
|
|
`forceReconnect` so the `↻ Reconnect` button wires to the hook's
|
|
`reconnectNonce` bump (aborts and reopens the stream —
|
|
`useTailStream.ts:99-101`). The button only renders when
|
|
`status === "stalled" || status === "error"` (`TailStatusPill.tsx:52`).
|
|
|
|
## Anti-patterns
|
|
|
|
- **Don't hand-roll `fetch` + `ReadableStream` parsing in a page.**
|
|
All stream consumers go through `streamTail(resource, opts?)` at
|
|
`src/lib/tail-stream.ts:58`. The parser handles `TextDecoderStream`,
|
|
newline splitting, malformed-line tolerance (`console.warn` + skip),
|
|
the `KNOWN_TYPES` allowlist, and abort-on-signal semantics
|
|
(`tail-stream.ts:75,98,107`). Duplicating it loses all of those.
|
|
- **Don't change the wire format on one endpoint without updating
|
|
the others.** The parser (`src/lib/tail-stream.ts`) and the hook
|
|
dispatch switch (`useTailStream.ts:173-195`) are shared across all
|
|
three live streams. Adding a new event type means updating
|
|
`TailEvent`, `KNOWN_TYPES`, the dispatch `case`, the `armStall`
|
|
re-arm list, and the backend emitter — in that order.
|
|
- **Don't emit `item` events before `snapshot_end`.** The hook uses
|
|
`snapshot_end` as the marker to flip `<TailStatusPill>` from
|
|
`connecting` to `live` and to reset the reconnect backoff counter.
|
|
Emitting `item`s first lands rows in `useTailStore` but the UI
|
|
still reads "Connecting" — operators see a flash of stale state.
|
|
Emit snapshot, then `snapshot_end`, then live (`api.py:1386-1401`).
|
|
- **Don't change the 30s stall threshold without updating the README
|
|
"Status pill" table.** The constant
|
|
(`STALL_TIMEOUT_MS = 30_000` at `useTailStream.ts:53`) is the
|
|
contract the pill is documented against — a bump needs the matching
|
|
edit at `README.md:118`.
|
|
- **Don't add `useTailStream` calls from `useFoo.ts` hooks.**
|
|
Streaming connections belong on the page (`src/pages/Claims.tsx`,
|
|
`Remittances.tsx`, `ActivityLog.tsx`). Hoisting into `useFoo`
|
|
couples the lifecycle to whoever mounts it and breaks
|
|
one-resource-one-page ownership.
|
|
|
|
## Related skills
|
|
|
|
- **`cyclone-frontend-page`** — page components live in `src/pages/`;
|
|
load when adding or refactoring a streaming page to confirm the
|
|
route + `PageHeader` + table conventions.
|
|
- **`cyclone-api-router`** — endpoint conventions (`api_routers/` for
|
|
resource-group routers, `api.py` for the live-tail endpoints); load
|
|
when adding or changing an HTTP endpoint that surfaces streamed or
|
|
paginated data.
|
|
- **`cyclone-store`** — write-path conventions in `store.py` and the
|
|
pubsub event contract (`claim_written`, `remittance_written`,
|
|
`activity_recorded`); load when adding a new entity whose writes
|
|
should fan out to a stream endpoint.
|
|
- **`cyclone-tests`** — frontend `*.test.tsx` siblings cover
|
|
`useTailStream`, `useMergedTail`, `TailStatusPill`; backend
|
|
`test_api_stream_live.py` covers the three live-tail endpoints;
|
|
load when the increment changes wire-format behavior or adds a
|
|
streaming hook. |