# Broadcast coordination

### Reserve a connection slot

The new optional `admit({ ctx, request, signal })` hook runs once per physical
multiplexed connection, after context/authentication hooks and before the SSE
response. It may return a release function (synchronous or asynchronous), or
nothing. Throw a catalog error to reject the whole HTTP request. HTTP 429 and
5xx responses are retryable; `Retry-After` is honored by the browser. Admission
does not replace independent authorization for each channel.

This example uses the existing `LocksPort` to implement an **application-owned**
three-slot policy. Declare `locks: LocksPort` in `AppPorts` and wire your lock
provider in `server/providers.ts`. Memory locks limit one process; distributed
limits need a shared, atomic lease store. Redis is one option, not a requirement.
Add `BroadcastConnectionLimit` to `features/shared/errors.ts` with status `429`,
code `BROADCAST_CONNECTION_LIMIT`, and message `Too many broadcast connections`.
Place the policy in `server/broadcast-admission.ts`:

```ts
import type { LocksPort } from "@beignet/core/locks";
import { appError } from "@/features/shared/errors";

export async function reserveBroadcastConnection({
  locks, userId, signal,
}: { locks: LocksPort; userId: string; signal: AbortSignal }) {
  for (let slot = 0; slot < 3; slot++) {
    signal.throwIfAborted();
    const result = await locks.acquire(`broadcast:${userId}:${slot}`, {
      ttlMs: 300_000,
      waitMs: 0,
    });
    if (result.acquired) {
      // Return ownership even if cancelled while awaiting acquisition.
      return async () => { await result.lease.release(); };
    }
  }
  throw appError("BroadcastConnectionLimit", { headers: { "Retry-After": "5" } });
}
```

Register it on the endpoint in `app/api/broadcasts/route.ts`. This example
assumes the authenticated session is available in `ctx.auth`; `requireUserId(ctx)`
asserts it before acquisition:

```ts
import { createBroadcastRoute } from "@beignet/next";
import { auth } from "@/lib/route-auth";
import { getServer } from "@/server";
import { requireUserId } from "@beignet/core/ports";
import { reserveBroadcastConnection } from "@/server/broadcast-admission";
import { channels } from "@/server/broadcasts";

export const runtime = "nodejs";
export const maxDuration = 300;
export const { GET } = createBroadcastRoute({
  server: getServer,
  channels,
  hooks: [auth.required()],
  maxLifetimeMs: 240_000,
  admit: ({ ctx, signal }) => reserveBroadcastConnection({
    locks: ctx.ports.locks,
    userId: requireUserId(ctx),
    signal,
  }),
});
```

For Fetch hosts, use the same `admit` option on `createBroadcastRoute` from
`@beignet/web` in `server/broadcast-route.ts`, passing the assembled server and
its channel registry. The host still owns its request duration setting.

Beignet calls each acquired release function once on expiration, request/body
cancellation, provider closure, or failed stream setup. If only some channels
fail, their resources are released while accepted channels retain the connection.
If acquisition finishes after cancellation or the ten-second admission setup
timeout, its returned resource is released immediately. Pass `signal` to an
acquisition API when supported; the example checks it between attempts, then
returns a successful acquisition even if cancellation happened while awaiting
it. Throwing after acquiring without returning release would leak that resource.

Cleanup failures emit `broadcast.cleanup-error` without skipping other releases.
No application stream event listeners are needed. Each renewal runs fresh
admission and authorization. **Distributed leases still need expiry** because
abrupt process termination cannot run cleanup. Keep lease TTL above the stream
lifetime plus acquisition/setup/cleanup headroom; this example pairs a
five-minute lease with a four-minute stream. The hosting request limit must also
leave that headroom. Storage, slot count, expiry, and rejection policy remain
application-owned; the framework imposes no connection-limit store.