Documentation
¶
Overview ¶
Package server is the high-level quicrtc HTTP/3 + WebTransport listener. It binds the wire/feed/session/cert pieces together and exposes a simple Publisher handle for the consumer's encoder loop.
One server = one publisher, many subscribers. For multi-upstream fan-out (one server aggregating from N publishers, or relay topologies), see the relay package.
Persisted session state for cross-disconnect resumption.
When a subscriber disconnects, we don't immediately tear down its per-session state — broadcaster receivers, track set, last-seen sequence numbers. Instead we move them into the sessionStore keyed by SessionID. The receivers' channels buffer AUs that arrive during the gap (subject to per-receiver buffer capacity).
On reconnect with HELLO.SessionID, the server finds the persisted state, hands it to the new Session, and the new pump goroutines drain the buffered AUs onto fresh uni streams. Subscribers see a continuous sequence with no gap.
Persisted entries expire after Config.Session.ResumeWindow if no reconnect happens (default 60s). Eviction is opportunistic on add/lookup.
Index ¶
- func DefaultSupportedFeatures() []string
- func NewTransport(cfg Config) (transport.Transport, error)
- type Config
- type Publisher
- type Server
- func (s *Server) AddTrack(name string) *Publisher
- func (s *Server) AddTrackSpec(spec TrackSpec) *Publisher
- func (s *Server) AddTrackWithPriority(name string, priority uint8) *Publisher
- func (s *Server) AddTrackWithTrackID(name string, priority uint8, trackID uint8) *Publisher
- func (s *Server) Addr() string
- func (s *Server) AnySession() *session.Session
- func (s *Server) CertHashB64() string
- func (s *Server) IsDraining() bool
- func (s *Server) ListenAndServe(ctx context.Context) error
- func (s *Server) Metrics() metrics.Metrics
- func (s *Server) Publisher() *Publisher
- func (s *Server) RemoveTrack(name string)
- func (s *Server) ShareLink() string
- func (s *Server) Shutdown(drainCtx context.Context) error
- func (s *Server) Slug() string
- func (s *Server) SubscriberCount() int
- type SessionHandle
- func (h SessionHandle) Context() context.Context
- func (h SessionHandle) InboundRecv(ctx context.Context, name string) (pubsub.AccessUnit, error)
- func (h SessionHandle) ReceiveDatagram(ctx context.Context) ([]byte, error)
- func (h SessionHandle) SendDatagram(payload []byte) error
- func (h SessionHandle) SessionID() string
- type SessionStore
- type TrackSpec
Constants ¶
This section is empty.
Variables ¶
This section is empty.
Functions ¶
func DefaultSupportedFeatures ¶
func DefaultSupportedFeatures() []string
DefaultSupportedFeatures lists every optional protocol feature the current build implements. Use as a starting point for Config.SupportedFeatures; trim to disable specific features.
Types ¶
type Config ¶
type Config struct {
// Addr is the UDP address to listen on for QUIC + HTTP/3, e.g.
// ":4433" or "0.0.0.0:4433".
Addr string
// CertBundle, if non-nil, is the TLS bundle to serve. Otherwise
// an ephemeral ECDSA P-256 cert is generated covering loopback +
// CertExtraIPs.
CertBundle *cert.Bundle
// CertExtraIPs are extra SAN IPs for the auto-generated cert
// (e.g. LAN, public IPv6, Tailscale). Ignored if CertBundle is
// supplied.
CertExtraIPs []net.IP
// Slug is the per-session shared secret subscribers must echo in
// HELLO. Empty -> a fresh 128-bit base64url slug is generated.
Slug string
// ExternalHost is the host (no scheme/port) that subscribers will
// reach the server on. Used in ShareLink. Defaults to the first
// CertExtraIPs entry, or "127.0.0.1" if none.
ExternalHost string
// AllowedOrigins are the HTTP Origins permitted by the
// WebTransport upgrade. Empty disables origin checking. Slug
// auth is the security boundary; this is defense in depth.
AllowedOrigins []string
// MaxSessions caps concurrent subscribers. <=0 picks 16.
MaxSessions int
// SDP is the codec/dimensions advertisement sent on HELLO. Set
// before publishing.
SDP wire.SDP
// Session is forwarded to session.New for handshake/idle/data
// tunables; zero values pick session defaults.
Session session.Config
// PerSubscriberBuffer is the broadcaster's per-subscriber channel
// depth. <=0 picks 30 (~1s @ 30fps).
PerSubscriberBuffer int
// OnDataChannel, if non-nil, is invoked once per session with the
// session's DataChannel. The server passes this through to
// session.Config.
OnDataChannel func(*datachannel.Channel)
// Metrics, if non-nil, receives observability events. nil =
// metrics.NoopMetrics{} (zero cost). Plug a Prometheus-backed
// implementation here for production.
Metrics metrics.Metrics
// Logger, if non-nil, is the structured logger used for server
// events. nil = slog.Default(). Set to a discard handler to
// silence completely.
Logger *slog.Logger
// SessionStore, if non-nil, is the backend that holds parked
// session state across socket disconnects. Default is
// NewMemorySessionStore — sufficient for single-instance
// deployments. Multi-instance setups should pair sticky LB
// routing (keyed on the SessionID issued in HELLO) with the
// memory store, OR plug a shared backend.
SessionStore SessionStore
// AuthValidator, if non-nil, replaces shared-slug auth with a
// caller-supplied validator. Receives the raw HELLO slug field
// (callers can stuff a JWT or bearer token there). Returns the
// tenant scope (used to namespace parked sessions for resume so
// tenants cannot cross-resume each other's sessions) and nil if
// the credential is valid; any error rejects.
//
// For shared-slug (single-tenant) auth, leave this nil and rely
// on the auto-generated Slug. For multi-tenant SaaS, plug a JWT
// validator that resolves to a tenant — return the tenant ID as
// the first return value.
AuthValidator func(credential string) (tenant string, err error)
// OnBackpressure, if non-nil, is invoked when a subscriber
// signals receive-side congestion. trackName identifies the
// track (empty = session-level). level is 0..100 (drained..full).
// Implementations should be cheap (called from the session's
// control reader); offload heavy work to a goroutine. Use the
// signal to lower bitrate, drop a modality, or pause non-critical
// publishing.
OnBackpressure func(sessionID, trackName string, level uint8)
// OnKeyframeRequest, if non-nil, is invoked when a subscriber
// asks for a fresh keyframe on a specific track via Client.
// RequestKeyframe. The application should signal its encoder to
// emit a keyframe NOW — the typical use is computer-use / live-
// screen workloads recovering from a P-frame loss event without
// waiting for the next natural keyframe.
OnKeyframeRequest func(sessionID, trackName string)
// OnSession, if non-nil, is invoked once per accepted session
// after the HELLO handshake completes — i.e., the peer has
// authenticated and SessionID is populated. The SessionHandle
// exposes per-session capabilities (datagram send/receive, inbound
// PublishBack track read) so applications can implement kind-aware
// behaviors without changes to the core pump dispatch.
//
// The callback runs on the session goroutine; long-running work
// should spawn its own goroutine. Note: prior to v0.1.1 this
// callback fired pre-handshake against an unauthenticated peer;
// the contract was tightened to fire post-authentication only.
OnSession func(h SessionHandle)
// InboundRateLimit caps inbound (PublishBack) traffic per session.
// Production deployments should set this to prevent a single
// misbehaving subscriber from saturating the session. Zero =
// unlimited (default; safe for closed deployments).
InboundRateLimit session.RateLimit
// SupportedFeatures is the optional protocol features this server
// offers. Intersected with the client's HELLO.Features and echoed
// in SDP.Features. nil = use the defaults (all features the
// current build implements).
SupportedFeatures []string
// CertGetter, if non-nil, replaces the static cert from CertBundle
// with a per-handshake lookup. Plug a *cert.Reloader here to pick
// up filesystem-rotated certs without restarting the server.
// When nil, CertBundle's cert is used unchanged.
//
// reloader, _ := cert.NewReloader(certPath, keyPath, cert.ReloaderOptions{...})
// cfg.CertGetter = reloader.GetCertificate
CertGetter func(*tls.ClientHelloInfo) (*tls.Certificate, error)
}
Config holds tunables for a quicrtc server.
type Publisher ¶
type Publisher struct {
// contains filtered or unexported fields
}
Publisher is what the consumer uses to feed access units.
type Server ¶
type Server struct {
// contains filtered or unexported fields
}
Server is one running quicrtc instance.
func New ¶
New constructs a Server. ListenAndServe must be called separately to actually bind and serve.
func (*Server) AddTrack ¶
AddTrack registers a new publishable track at default priority (KindVideo, priority 4). Equivalent to AddTrackSpec(TrackSpec{ Name: name }). For non-video tracks, use AddTrackSpec so the per- kind DeliveryClass dispatch actually runs.
func (*Server) AddTrackSpec ¶
AddTrackSpec registers a publishable track described by spec. The Kind field selects the DeliveryClass — KindTokens routes through the persistent low-latency uni stream, KindTelemetry through datagrams, KindToolCalls through one bidi stream per call, etc. Zero-Kind (or KindVideo) preserves the legacy stream-per-GOP path.
Priority handling: spec.Priority's zero value (0) is treated as "use the Kind's default" via track.DefaultPriority. This lets callers register audio tracks at PriorityCritical (also 0) by leaving Priority unset, rather than getting silently demoted to PriorityNormal.
Idempotent on Name; second call returns the existing Publisher and ignores subsequent spec values (use RemoveTrack + AddTrackSpec to change).
func (*Server) AddTrackWithPriority ¶
AddTrackWithPriority is a legacy entry point preserved for backward compatibility. New code should use AddTrackSpec.
Idempotent on name; second call returns the existing Publisher and ignores the new priority (use RemoveTrack + AddTrack to change it).
func (*Server) AddTrackWithTrackID ¶
AddTrackWithTrackID is a legacy entry point preserved for backward compatibility. New code should use AddTrackSpec, which carries Kind — without a Kind, the track always uses the stream-per-GOP path.
Idempotent on name; second call returns the existing Publisher.
func (*Server) Addr ¶
Addr returns the local listen address (with port resolved if the configured port was 0). Available after Listen has begun.
func (*Server) AnySession ¶
AnySession returns the most recently-attached live Session, or nil. Used by the publisher-side Transport adapter to read inbound PublishBack AUs. In multi-subscriber deployments, each subscriber has its own Session and "most recent" is a pragmatic default; full per-subscriber routing arrives with capability negotiation in Phase 3.
func (*Server) CertHashB64 ¶
CertHashB64 returns the SHA-256(DER) cert hash in the form browsers expect for serverCertificateHashes.
func (*Server) IsDraining ¶
IsDraining reports whether a Shutdown is in progress. Used by the upgrade handler to reject new sessions during drain.
func (*Server) ListenAndServe ¶
ListenAndServe binds the listener and runs until ctx is cancelled or a fatal error occurs. Returns nil on graceful shutdown.
func (*Server) Metrics ¶
Metrics returns the metrics sink. Useful for tests and for integrators that want to read snapshot counters from the in-memory sink without going through Prometheus.
func (*Server) Publisher ¶
Publisher returns the default ("primary") Publisher handle. Kept for single-track callers that pre-date AddTrack.
func (*Server) RemoveTrack ¶
RemoveTrack tears down a previously-added track. The per-track broadcaster is closed (no more AUs accepted), all session pumps for this track are stopped, and a TypeUnannounce frame is sent to each connected subscriber so they can close their per-track recv queues. No-op if the track doesn't exist.
func (*Server) ShareLink ¶
ShareLink returns a URL the browser can use to connect. Includes the slug and cert hash in the fragment so the page (served on the same origin) can read them with location.hash.
func (*Server) Shutdown ¶
Shutdown drains the server gracefully:
- Stops accepting new sessions (sets draining flag).
- Sends TypeClose on every live session's control stream so subscribers learn to stop sending.
- Waits up to drainCtx.Deadline for in-flight sessions to exit cleanly.
- Closes the underlying webtransport listener and UDP socket.
Returns drainCtx.Err() if drain didn't complete in budget, nil otherwise. Safe to call multiple times.
func (*Server) SubscriberCount ¶
SubscriberCount returns the current number of connected subscribers. Used by the transport adapter for Stats.
type SessionHandle ¶
type SessionHandle struct {
// contains filtered or unexported fields
}
SessionHandle exposes per-session capabilities to the application via Config.OnSession. Applications use it to implement kind-aware behaviors that the core pump dispatch doesn't yet integrate — principally, fire-and-forget telemetry over QUIC datagrams and receiving subscriber-published tracks per-session.
SendDatagram and ReceiveDatagram pass directly through to the underlying webtransport.Session; the same MTU and reliability caveats apply (DatagramTooLargeError on oversize, no retransmit).
func (SessionHandle) Context ¶
func (h SessionHandle) Context() context.Context
Context returns the underlying WebTransport session's context. It is cancelled when the session closes — callers can `select` on it to know when SendDatagram / ReceiveDatagram will no longer succeed, without having to call them and inspect the error.
func (SessionHandle) InboundRecv ¶
func (h SessionHandle) InboundRecv(ctx context.Context, name string) (pubsub.AccessUnit, error)
InboundRecv blocks until the next AccessUnit on a subscriber- published track arrives on THIS session. Unlike Server.AnySession, AUs from other sessions are not mixed in. Blocks until the subscriber announces the track if it hasn't yet.
func (SessionHandle) ReceiveDatagram ¶
func (h SessionHandle) ReceiveDatagram(ctx context.Context) ([]byte, error)
ReceiveDatagram blocks until one datagram from the subscriber arrives or ctx is done.
func (SessionHandle) SendDatagram ¶
func (h SessionHandle) SendDatagram(payload []byte) error
SendDatagram sends one unreliable QUIC datagram to the subscriber attached to this session. Returns DatagramTooLargeError if the payload exceeds the path MTU.
func (SessionHandle) SessionID ¶
func (h SessionHandle) SessionID() string
SessionID returns this session's unique identifier. Empty when called from OnSession — the ID is allocated during HELLO handshake.
type SessionStore ¶
type SessionStore interface {
// AllocateID returns a fresh, unique session ID.
AllocateID() string
// Park stores the receivers under (tenant, sessionID) with TTL
// semantics. onEvict, if non-nil, is invoked synchronously when
// the entry expires WITHOUT being resumed — used to unsubscribe
// receivers from their broadcasters so the broadcasters stop
// fanning to channels nobody will ever read again. Resume does
// NOT invoke onEvict (the caller is responsible for the receivers
// once they've been resumed).
Park(tenant, sessionID string, receivers map[string]*pubsub.Receiver, onEvict func())
// Resume looks up + removes the parked entry scoped to tenant.
// Returns nil if the (tenant, sessionID) pair doesn't exist or
// has expired.
Resume(tenant, sessionID string) map[string]*pubsub.Receiver
}
SessionStore is the pluggable backend that holds parked session state across socket disconnects. The default in-memory impl is sufficient for single-instance deployments; production multi- instance setups can plug a shared backend (Redis, etcd, postgres) for cross-instance resume.
Sessions are namespaced by (tenant, sessionID): a resume request from tenant A cannot match a parked session belonging to tenant B even if both happen to share a SessionID. For single-tenant deployments, tenant is simply the empty string and the namespace is global.
IMPORTANT scope note: a parked session holds in-process receivers (channels into broadcasters). These cannot be serialized across processes. So a "shared" backend can hold session metadata (track set, last-seen sequences, expiresAt) but the actual replay path requires either:
- Sticky routing — LB sends reconnect to the same instance that holds the receivers (recommended P0).
- Active session migration between instances (future P1) — complex; involves shipping replay buffers + reattaching to the broadcaster on the new instance.
For P0, sticky routing is the supported path. The HELLO/SDP SessionID can be used as the LB affinity key.
func NewMemorySessionStore ¶
func NewMemorySessionStore(ttl time.Duration) SessionStore
NewMemorySessionStore returns the default in-memory SessionStore. Equivalent to passing nil to server.Config.SessionStore.
type TrackSpec ¶
type TrackSpec struct {
// Name uniquely identifies the track within this server. Required.
Name string
// Kind selects the DeliveryClass via track.DefaultDeliveryClass.
// Zero value ("") is treated as KindVideo for back-compat with the
// legacy AddTrack* methods that don't carry a Kind.
Kind track.Kind
// Priority is the RFC 9218 priority hint (lower = more urgent).
// Zero value picks PriorityNormal (4).
Priority uint8
// TrackID is the 1-byte demux key used by datagram envelopes when
// Kind is KindTelemetry (or any DatagramOrStream track). Pick
// distinct values per session; collisions silently merge tracks
// on the receiver side. Unused for stream-based deliveries.
TrackID uint8
}
TrackSpec is the modern entry point for AddTrack. All fields except Name are optional; zero values pick safe defaults (KindVideo at priority 4 with no trackID). The Kind field is what unlocks the per-kind delivery dispatch in feed.Pump — without it, the track falls through to the legacy stream-per-GOP path regardless of payload shape.
Source Files