websocket

package
v0.2.28 Latest Latest
Warning

This package is not in the latest version of its module.

Go to latest
Published: Sep 13, 2026 License: MPL-2.0 Imports: 26 Imported by: 0

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

View Source
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"
)
View Source
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

View Source
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")
	// ErrRelayUnavailable: the session is not on this node and the cross-
	// 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

func IsUpgrade(r *http.Request) bool

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

func RequestedSubprotocols(r *http.Request) []string

RequestedSubprotocols parses Sec-WebSocket-Protocol (a comma list that may span several header lines) into a clean list.

func ValidCloseCode

func ValidCloseCode(code int) bool

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.

func NewKVDirectory

func NewKVDirectory(kv *kvstore.KV, ttl time.Duration) Directory

NewKVDirectory builds a Directory over a KV handle. Hand it a handle built with no TTL clamp (kvstore.New(store, 0, 0)): --kv-max-ttl guards authors' keys and must not shorten a session lease.

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.

func OpenRelay

func OpenRelay(name string, cfg RelayConfig) (Relay, error)

OpenRelay constructs the named relay backend. Unknown name is a hard error listing what is available.

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.

Jump to

Keyboard shortcuts

? : This menu
/ : Search site
f or F : Jump to
y or Y : Canonical URL