Documentation
¶
Overview ¶
Package websocket implements the chassis's `websocket` personality: the long-lived side of an HTTP Upgrade.
The web head owns the way in. An upgrade request runs through the stack's ordinary web ops like any other request — path, cookies, session lookup, Origin, whatever the stack decides — and a rule that wants the connection EXECs `txco://websocket/accept`. That op records the decision here, keyed by a session id the web head minted before the run. When the run ends the web head asks this controller whether an accept was recorded; if so it performs the RFC 6455 handshake and hands the socket over, and the request handler returns.
From then on the connection is a chassis-owned session:
- one reader goroutine turns frames into complete messages and queues them (bounded);
- one worker goroutine takes them in order and makes ONE bounded processor run per message into `<stack>/_websocket/0` — the stack never holds the socket, and an idle connection runs nothing;
- one pinger goroutine keeps the connection alive through proxies and closes it when it goes idle;
- `txco://websocket/{send,reply,close}` write back through the registry this controller keeps.
Deliberate v1 cuts, all labelled in docs/advanced/protocols/websocket.md: no outbound queue (a send is one bounded write; a peer that will not read is closed at --websocket-write-timeout), sends resolve on this node only (the registry seam is where a fleet directory plugs in), binary rides the envelope as base64 under the same size cap as text.
The personality is OFF by default: `websocket` must be in --personalities, and it only does anything alongside `web`.
Index ¶
- Constants
- Variables
- func IsUpgrade(r *http.Request) bool
- func RegisterRelay(name string, c RelayConstructor)
- func RequestedSubprotocols(r *http.Request) []string
- func ValidCloseCode(code int) bool
- type Accept
- type Controller
- func (c *Controller) Claim(sid string) (Accept, bool)
- func (c *Controller) CloseSession(ctx context.Context, tenant, id string, code int, reason string) error
- func (c *Controller) Count() int
- func (c *Controller) Enabled() bool
- func (c *Controller) LeaseTTL() time.Duration
- func (c *Controller) Local() LocalHandler
- func (c *Controller) NewSessionID() string
- func (c *Controller) RecordAccept(sid string, a Accept) error
- func (c *Controller) Send(ctx context.Context, tenant, id string, typ MessageType, data []byte) error
- func (c *Controller) SetRelay(r Relay, d Directory)
- func (c *Controller) Start()
- func (c *Controller) Stop()
- func (c *Controller) Upgrade(w http.ResponseWriter, r *http.Request, sid string, a Accept)
- type Directory
- type LocalHandler
- type MessageType
- type Registry
- type Relay
- type RelayConfig
- type RelayConstructor
Constants ¶
const ( // Source is the `_txc.src` every session run carries. Source = "websocket" // SubStack is the sub-stack a session's runs enter: <stack>/_websocket/0. SubStack = "_websocket" // StateMaxBytes caps the accept-time `state` object a stack attaches to a // session (stamped verbatim on every message envelope). StateMaxBytes = 4096 // Event names a stack may opt into at accept time. EventClose = "close" )
const DirectoryNamespace = "_txc.websocket"
DirectoryNamespace is the chassis-reserved KV namespace that records which node owns each session: key = session id, scoped per tenant like every KV key (<tenant>/_txc.websocket/<sid>), so a lookup with the wrong tenant simply misses — no cross-tenant existence oracle, structurally. Author- facing writers (txco://kv/*, KV/ seed packs) refuse `_txc.*` namespaces, the same reservation the blob index relies on.
Variables ¶
var ( ErrDisabled = errors.New("websocket personality is not enabled on this node") ErrSessionNotFound = errors.New("no such session") ErrSessionClosed = errors.New("session is closed") ErrWriteTimeout = errors.New("write timed out") ErrMessageTooLarge = errors.New("message exceeds the session's size limit") ErrPendingFull = errors.New("too many unclaimed accepts") // node path (directory or relay) could not be used. Tenant-independent, // so it reveals nothing about another tenant's sessions. ErrRelayUnavailable = errors.New("cross-node delivery is unavailable") // ErrSessionAttached: the session holds an attached transport (a PTY), // so its frames belong to the process, not to txco://websocket/send. The // pump writes through the private send path and is unaffected (D5). ErrSessionAttached = errors.New("session has an attached transport") )
Errors the Registry returns; the ops map them to `<into>.error.code`.
Functions ¶
func IsUpgrade ¶
IsUpgrade reports whether the request asks for a WebSocket upgrade: GET with `Connection: upgrade` and `Upgrade: websocket` (token lists, case-insensitive). The web head stamps `_txc.websocket.upgrade` on exactly these requests; everything else is ordinary HTTP.
func RegisterRelay ¶
func RegisterRelay(name string, c RelayConstructor)
RegisterRelay adds a named cross-node relay backend. Called from init() in the backend package; activated by a blank import. No backend is built in open core — the node-local registry is the default and only built-in.
func RequestedSubprotocols ¶
RequestedSubprotocols parses Sec-WebSocket-Protocol (a comma list that may span several header lines) into a clean list.
func ValidCloseCode ¶
ValidCloseCode reports whether a stack may close with this status code: the RFC 6455 codes a peer may send plus the private range. 1005/1006/1015 are reserved for the transport itself.
Types ¶
type Accept ¶
type Accept struct {
// Tenant is the pinned tenant slug of the upgrade run (unforgeable).
Tenant string
// Stack is the web stack that accepted; session runs enter
// <Stack>/_websocket/0.
Stack string
// Ingress and HostnameVerified are copied from the upgrade run's
// route facts onto every session run's route pre-stamp.
Ingress string
HostnameVerified bool
// State is a JSON object (≤ StateMaxBytes) stamped verbatim as
// `_txc.websocket.session.state` on every message; "" = none.
State string
// Subprotocols the stack is willing to speak, in preference order;
// SubprotocolRequired refuses the upgrade when none is offered.
Subprotocols []string
SubprotocolRequired bool
// Origins are host patterns allowed cross-origin; AnyOrigin disables
// the check. With neither, only same-host origins (or no Origin
// header — a non-browser client) are accepted.
Origins []string
AnyOrigin bool
// Events the stack wants as runs beyond messages (EventClose).
Events map[string]bool
// Per-session overrides, 0 = chassis default; the controller clamps
// them to the chassis ceilings.
IdleTimeout time.Duration
MaxMessageBytes int64
}
Accept is the decision a stack makes with txco://websocket/accept. The op fills it from its WITH clause and the pinned run identity; the web head consumes it at claim time.
type Controller ¶
type Controller struct {
// contains filtered or unexported fields
}
Controller is the session registry and lifecycle owner. It satisfies the server's Start()/Stop() controller contract and the Registry interface the txco://websocket/* ops depend on.
func NewController ¶
func NewController(ctx context.Context, pu *processor.Unit) *Controller
NewController constructs (but does not start) the websocket controller. Inert unless both `websocket` and `web` are in --personalities: the controller then never records an accept, so the web head never upgrades.
func (*Controller) Claim ¶
func (c *Controller) Claim(sid string) (Accept, bool)
Claim returns and removes the recorded accept for sid, if any.
func (*Controller) CloseSession ¶
func (c *Controller) CloseSession(ctx context.Context, tenant, id string, code int, reason string) error
CloseSession implements Registry: this node first; otherwise the node the directory names.
func (*Controller) Count ¶
func (c *Controller) Count() int
Count returns the number of open sessions on this node (tests, status).
func (*Controller) Enabled ¶
func (c *Controller) Enabled() bool
Enabled reports whether this node accepts websocket sessions.
func (*Controller) LeaseTTL ¶
func (c *Controller) LeaseTTL() time.Duration
LeaseTTL is the directory lease length (tests).
func (*Controller) Local ¶
func (c *Controller) Local() LocalHandler
Local returns the node-local delivery surface a Relay backend serves inbound requests through.
func (*Controller) NewSessionID ¶
func (c *Controller) NewSessionID() string
NewSessionID mints an opaque, unguessable, time-sortable session id: `ws_` + a ULID whose 80 entropy bits come from crypto/rand. Not hxid — its ULID entropy is math/rand-seeded, fine for a request id and wrong for an id that names a socket.
func (*Controller) RecordAccept ¶
func (c *Controller) RecordAccept(sid string, a Accept) error
RecordAccept implements Registry: the txco://websocket/accept op stores the decision for the session id the web head minted.
func (*Controller) Send ¶
func (c *Controller) Send(ctx context.Context, tenant, id string, typ MessageType, data []byte) error
Send implements Registry: this node first; otherwise, with a relay wired, the node the directory names.
func (*Controller) SetRelay ¶
func (c *Controller) SetRelay(r Relay, d Directory)
SetRelay wires cross-node delivery: a Relay to reach other nodes' sessions and a Directory that records which node owns each session. Call before Start. With neither, a session is reachable only on the node holding its socket (the open-core default).
func (*Controller) Start ¶
func (c *Controller) Start()
Start begins the pending-accept sweeper. There is no listener to bind: sessions arrive through the web head's Upgrade handoff.
func (*Controller) Stop ¶
func (c *Controller) Stop()
Stop closes every session with 1001 (going away) and waits, bounded by --websocket-drain-timeout. http.Server.Shutdown never waits for hijacked connections, so this is the only drain a websocket session gets.
func (*Controller) Upgrade ¶
func (c *Controller) Upgrade(w http.ResponseWriter, r *http.Request, sid string, a Accept)
Upgrade performs the handshake for a claimed accept and starts the session. It writes the HTTP response itself on every refusal, so the caller returns unconditionally afterwards. The request context dies with the handler; the session roots on the controller's context.
type Directory ¶
type Directory interface {
Register(ctx context.Context, tenant, sid, node, stack string) error
Refresh(ctx context.Context, tenant, sid, node, stack string) error
Unregister(ctx context.Context, tenant, sid string) error
// Lookup answers the owning node's address; found=false is a miss.
Lookup(ctx context.Context, tenant, sid string) (node string, found bool, err error)
}
Directory maps (tenant, session id) → the node whose inbox reaches the session's socket. Entries are leases: written on open, refreshed by the owning node while the session lives, deleted on close, and expired by the store if the node dies — so a stale entry can only ever point at a node that no longer answers, which the Relay reports as not found.
type LocalHandler ¶
type LocalHandler interface {
Send(ctx context.Context, tenant, sid string, typ MessageType, data []byte) error
Close(ctx context.Context, tenant, sid string, code int, reason string) error
}
LocalHandler is the node-local surface a Relay backend calls for the requests that arrive at this node's inbox. It resolves sessions on this node ONLY — never the cross-node Send — so a request can never bounce between nodes.
type MessageType ¶
type MessageType int
MessageType is the application-level frame kind. Our own type so the op layer never sees the WebSocket library's.
const ( MessageText MessageType = iota + 1 MessageBinary )
func (MessageType) String ¶
func (t MessageType) String() string
String renders the envelope spelling of the type.
type Registry ¶
type Registry interface {
Enabled() bool
// RecordAccept stores a stack's decision for the session id the web head
// minted on the upgrade request.
RecordAccept(sid string, a Accept) error
// Send writes one message to a live session of the tenant.
Send(ctx context.Context, tenant, id string, typ MessageType, data []byte) error
// CloseSession starts the close handshake on a live session of the tenant.
CloseSession(ctx context.Context, tenant, id string, code int, reason string) error
}
Registry is the surface the txco://websocket/* ops depend on. The Controller implements it for this node; a fleet session directory wraps it later without touching the ops.
type Relay ¶
type Relay interface {
// Node is the address other nodes use to reach THIS node's inbox — the
// backend's subject-safe rendering of RelayConfig.NodeID. The controller
// records it in the directory beside every session it owns.
Node() string
// Send writes one message to a session that lives on `node`. It returns
// the owning node's answer as the Registry errors (ErrSessionNotFound
// when nothing on that node owns the session, ErrSessionClosed,
// ErrWriteTimeout, ErrMessageTooLarge), ErrWriteTimeout when ctx expires
// first, or ErrRelayUnavailable when the bus itself is down.
Send(ctx context.Context, node, tenant, sid string, typ MessageType, data []byte) error
// Close starts the close handshake on a session that lives on `node`.
Close(ctx context.Context, node, tenant, sid string, code int, reason string) error
// Shutdown stops answering for this node and releases the bus
// connection, bounded by ctx.
Shutdown(ctx context.Context) error
}
Relay delivers a send or close to the node that owns a session when that node is not this one. Open core defines the seam and ships no backend; a backend registered by a build that has a message bus carries the request to the owning node's inbox and returns that node's answer.
The seam mirrors chassis/room's Relay: core defines the interface + the registry, a backend self-registers via init() + a blank import, and the backend reads its own connection details from its own env. Unlike the room relay, delivery here is addressed and answered — a `txco://websocket/send` reports whether the message reached the socket, so a relay call is a request with a reply, not a fire-and-forget publish.
type RelayConfig ¶
type RelayConfig struct {
// NodeID names this chassis (the operator's FQDN, per-Machine on a
// cloud fleet). The backend derives Node() from it.
NodeID string
Logger *zap.Logger
Handler LocalHandler
}
RelayConfig carries the chassis-owned facts a backend cannot obtain on its own (the bgservice.Config precedent). Connection details — URL, credentials, subject prefix — are the backend's own env.
type RelayConstructor ¶
type RelayConstructor func(cfg RelayConfig) (Relay, error)
RelayConstructor builds a Relay: connects, subscribes this node's inbox, and wires inbound requests to cfg.Handler.