Socket sessions#

A Request in, a send and a close callback out. That is the whole seam, which is why ws, socket.io, uWS, Bun, Deno and a Durable Object all drive the same core.

import { createActorSocketSession } from '@sigx/actors/server';

const session = await createActorSocketSession({
    host,
    request,                                   // the UPGRADE request
    send: (message) => socket.send(message),
    close: (code, reason) => socket.close(code, reason),
});

socket.on('message', (data) => session.handle(String(data)));
socket.on('close', () => session.close());

session.handle(message) never throws — a protocol breach closes the socket through your close callback instead. session.close() aborts every in-flight call and is idempotent.

Most people never call this directly: attachActorSocket wraps it for Node, and workerSocket() for Workers.

What pins at the upgrade, and what does not#

Origin is checked once, at the upgrade. This is the one check that cannot move to per-message, because the cookies ride the upgrade: a browser will open a cross-origin WebSocket with credentials attached and no preflight to stop it. The policy is posture.origin ?? 'same-origin' — the same contract as ServerFnRequestOptions.origin.

Identity is pinned once, from the upgrade request's cookies.

The full prelude re-runs on every message. Middleware may be a rate limiter, so it has to see each call; authentication itself stays memoized on the connection.

A refused upgrade — origin, or authentication — answers before serving a byte.

Calls#

Unary calls, streams and cancellation all ride the one connection, with a per-call deadline from posture.timeoutMs. Middleware sees per-call locals, but responseHeaders and status() are inert: there is no response to put a header on.

Guards reach the pipeline through the same single door every transport uses — enterActorRequest(), actorPosture() and actorPrincipal().

The caps HTTP gave you for free#

An HTTP mount gets body limits, connection limits and timeouts from the server around it. A socket has none of that, so the session states them:

An unparseable message closes 1003.

maxSubscriptions is validated at construction, not at first use — a security bound must not be disabled by a typo. The connection is the right unit for it, because the connection is what actually costs activations: over one socket a page can hold thousands of in-flight calls, each able to force an activation.

Subscriptions#

A subscription is established through the full pipeline — authorization, dispatch, the first value — under the app posture's timeoutMs. One whose first value cannot be produced in time receives a per-subscription 504 error frame, and the session releases the watch and the keep-alive it held, instead of hanging silently. Other subscriptions on the connection are unaffected, and a subscription that has delivered its first value is never timed out; pushes arrive at the watch loop's own cadence. The $live mount applies the same rule — see establishment has a deadline.

throttlePolicy decides how much say a client gets over its own delivery rate. A subscription may carry a requested window (w, throttleMs on the client), which the policy floors and rounds up to a fixed ladder:

createActorSocketSession({
    host, request, send, close,
    throttlePolicy: { min: 50, buckets: [50, 250, 1000, 5000] },   // DEFAULT_THROTTLE_POLICY
});

min is the fastest window a client may ask for; buckets is the ascending ladder a request is rounded up to. The default floor is the runtime's own 50 ms watch throttle, so a client can only ever ask to be served more slowly. { min: 0, buckets: [0, 16, 50] } opts a deployment into sub-50 ms delivery deliberately; { min: 50, buckets: [50] } refuses the feature. Validated at construction — an empty, unsorted, or entirely-below-floor ladder closes the socket with 1011 rather than surfacing per subscription. DEFAULT_THROTTLE_POLICY and the LiveThrottlePolicy type are exported from @sigx/actors/server.

Reporting the send buffer#

send(message: string) is the whole contract, so the session cannot see how much the transport has queued. An adapter that can supplies bufferedBytes():

createActorSocketSession({
    host, request,
    send: (message) => socket.send(message),
    close: (code, reason) => socket.close(code, reason),
    bufferedBytes: () => socket.bufferedAmount ?? null,
});

It surfaces as session.stats().bufferedBytes and, summed across open sessions, as socketStats().bufferedBytes — the host's own backpressure gauge. Return null, never 0, when the transport cannot say; omitting the option does the same. A genuine 0 — the transport asked, nothing is queued — is a real measurement and is reported as one. @sigx/actors-ws/node supplies it automatically.

Shedding a slow consumer#

send() is fire-and-forget, so a client that stops reading costs nothing on the wire and everything on the host: its frames queue in the host's own send buffer, and nothing on the client side reports it. maxBufferedBytes bounds that queue per connection:

createActorSocketSession({
    host, request, send, close,
    bufferedBytes: () => socket.bufferedAmount ?? null,
    maxBufferedBytes: 4 * 1024 * 1024,   // default 0 = off
});

After each subscription frame the session reads bufferedBytes(). Over the cap, it closes one subscription — the one that put the most bytes into the buffer since it was last empty — with the ordinary per-subscription terminal frame:

{"i":3,"e":{"status":503,"data":{"kind":"shed"}}}

The connection stays open; calls and the other subscriptions continue. While the buffer stays over the cap, each further delivery sheds the next heaviest subscription, so a connection that never drains ends up holding no subscriptions and at most the cap plus one frame. A slow consumer loses live values rather than being funded from the host's heap — the same posture as the 16-value drop-oldest buffer in front of each subscriber.

On the client, the live channel's onError receives an error with status === 503 and data.kind === 'shed', which is how "this subscriber was too slow" reads differently from "the read failed". The data shape is exported as SocketShedReason from @sigx/actors/socket-wire.

Default 0 is off, and the value is validated at construction like maxConnectionMs: a non-integer or negative value throws and closes the accepted socket 1011. The cap needs bufferedBytes() — an adapter that answers null never sheds, so on the Cloudflare adapters, which supply no buffer depth, it is inert. @sigx/actors-ws/node forwards the option and supplies the seam. Sheds are counted in socketStats().subscriptionsShed, and show as a shed row in sigx actors stats and top once there are any.

Session lifetime#

Both default to 0 (off), and both are validated at construction.

revalidateMs re-runs authentication against the pinned upgrade request on a fresh context every interval, and closes 1008 when it no longer stands: the authenticate hook throws, a previously-authenticated connection comes back anonymous, or the identity changes — a swap mid-connection is a reconnect, not a mutation.

maxConnectionMs caps one connection's lifetime and closes 1008 at the cap. It is also the credential-refresh mechanism, precisely because the reconnect is a fresh upgrade carrying the browser's current cookies. Subscriptions re-establish over it; in-flight calls fail un-retried, as on any drop.

Set one of them in any app with sign-out or short-lived sessions. With both off, a connection authenticated an hour ago is still trusted.

Next steps#

• @sigx/actors-ws — the client transport and the Node adapter.
• Observability — socketStats().
• Wire protocol — the messages themselves.