sse

README

sse

Server-Sent Events that resume from an exact cursor: a Go broadcast hub and a TypeScript client speaking one wire.

github.com/cplieger/sse is a broadcast hub for SSE endpoints whose clients resume from an exact position: every frame carries an <epoch>:<offset> cursor, every connection opens with a hello that says whether the presented cursor was honoured, and a hub refuses at construction any retention it could not keep. @cplieger/sse, published from web/, is the browser half of the same wire: it owns the connection over fetch, presents the cursor it holds, measures liveness on bytes, closes a hidden tab's stream and reopens it on return, and holds incoming frames while the application asks the server what changed. Both halves ship from one repository and one tag, and every timing constant they share is written once in timing.json and pinned by a test in each language.

The Go module has one dependency, webhttp, for the JSON error envelope its refusals answer with; webhttp itself is standard-library only. The TypeScript package has no runtime dependencies.

Install

• Go: go get github.com/cplieger/sse@latest
• TS: npx jsr add @cplieger/sse or npm i @cplieger/sse

Usage

hub, err := sse.New(
	sse.WithReplay(1024),
	sse.WithReplayTTL(10*time.Minute),
	sse.WithReplyMaxEvents(256),
	sse.WithPresence(func(ev sse.PresenceEvent) { presence.Record(ev) }),
)
if err != nil {
	return err // wraps sse.ErrConfig and names the offending option
}

mux.HandleFunc("GET /api/events", func(w http.ResponseWriter, r *http.Request) {
	hub.Serve(w, r,
		sse.WithTopic(r.URL.Query().Get("chat_id")),
		sse.OnConnect(func(w *sse.Writer, h sse.Hello) error {
			return w.Event("connected", fmt.Appendf(nil, `{"resumed":%t}`, h.Resumed))
		}),
		sse.WithClientTag(r.Header.Get("SSE-Client")),
	)
})
mux.Handle("POST /api/sync", webhttp.RouteTimeout(hub.DigestHandler(resolve), 10*time.Second, "digest timed out"))

offset, err := hub.Publish(sse.Event{Name: "notify", Topic: chatID, Data: payload})
if errors.Is(err, sse.ErrFrameTooLarge) || errors.Is(err, sse.ErrInvalidUTF8) {
	// Refused before any state changed: no offset was consumed. Publish a
	// fetch instruction instead of the payload.
}

// Inside webhttp.Run's WithPreDrain hook, so streams release before the HTTP drain:
if err := hub.Shutdown(ctx); err != nil {
	slog.Warn("sse: streams still draining", "error", err)
}

ExampleNew, ExampleHub_Serve and ExampleHub_DigestHandler in example_test.go run this shape end to end. The browser side of the same endpoint is the createStream example in web/README.md.

API

Constructing a hub

New(opts ...Option) (*Hub, error) returns an error wrapping ErrConfig for an option set the hub could not honour; MustNew panics with the same message, for package-level construction in main. Each option refuses its own value on the spot. The rules that compare two options (a ring without a TTL, a TTL below the client's watchdog window, a reply cap above the ring, a write timeout at or below the keepalive) and the derived defaults run once after every option has been applied, so option order never changes the verdict.

Option	Default	Refused
WithReplay(n)	0 (no replay)	negative; non-zero without WithReplayTTL
WithReplayTTL(d)	none	at or below zero; below max(3 × keepalive, 15s) + 30s (75s at the default keepalive)
WithReplayMaxBytes(n)	64 × MaxFrameBytes	below MaxFrameBytes
WithReplyMaxEvents(n)	min(ring, 256)	negative; above the ring
WithClientBuffer(n)	max(ring, 256)	at or below zero
WithMaxClients(n)	0 (unlimited)	never; zero or negative means unlimited
WithKeepalive(d)	15s	below 1ms
WithKeepaliveEvent(name)	sse:keepalive	CR or LF in the name; "" selects the : keepalive comment form
WithReconnectDelay(d)	1500ms	below 1ms
WithWriteTimeout(d)	2 × keepalive	at or below the keepalive
WithLogger(l)	slog.Default()	never; nil keeps the default
WithPresence(fn)	none	never; nil removes the hook

The ring is bounded three ways at once (count, age, bytes) and the oldest entries are evicted first; WithReplyMaxEvents caps how much of it one resume may be sent, so a long replay cannot hold a reconnecting client.

Serving and publishing

• (*Hub).Serve(w, r, opts ...ServeOption): subscribes the request and streams until the peer leaves, the request context ends, the client is reset as slow, a write fails, or the hub shuts down. It owns the proxy-defensive headers (Cache-Control: no-cache, no-transform, X-Accel-Buffering: no), the retry: field, the hello, the Last-Event-ID replay, keepalives and the sse:reset frame. Every write runs under WithWriteTimeout and the deadline is cleared after it, so a peer that stopped reading ends its own connection instead of the stream goroutine. A http.Flusher reachable through an Unwrap() chain works; the 500 streaming_unsupported refusal fires only when no flusher exists at any depth, and 503 sse_unavailable answers a connection over the client cap or after Shutdown.
• Serve options: WithTopic(t) (receive broadcasts plus events scoped to t; the empty topic receives everything), OnConnect(fn func(w *Writer, h Hello) error) (runs after the retry: field, the hello and the replay have been flushed, so the client's connect deadline never measures the hook; it receives the Hello the client received and writes initial-state frames through Writer.Event(name, data), each its own bounded write and flush, none carrying an id; an error ends the connection), WithClientTag(tag) (the application's presence key, carried on PresenceEvent.Tag; the empty string means no tag, so r.Header.Get("SSE-Client") can be passed through unconditionally, and any other value outside [A-Za-z0-9_-]{1,64} is treated as absent with one Warn).
• (*Hub).Publish(Event) (uint64, error): assigns the next offset, appends to the ring and fans out to every matching client without blocking. Checks run in a fixed order before the frame is accepted: a name starting with sse:, holding CR or LF, or equal to the configured keepalive name panics (a fixed property of the call site); Data that is not UTF-8 returns ErrInvalidUTF8; an encoded frame above MaxFrameBytes (1 MiB, terminating blank line included) returns ErrFrameTooLarge. A refused frame consumes no offset and leaves the ring untouched, before and after Shutdown alike; a valid frame after Shutdown is dropped with (0, nil), as is any frame on a nil hub. Data is UTF-8 text split on CRLF, CR and LF into data: lines, and the hub owns the slice from the call on: marshal fresh per publish. A client whose channel is full is reset as slow and dropped.
• (*Hub).Position() Position: the epoch, the oldest offset the ring can replay and the newest offset published, after evicting expired entries, so the floor a gauge reports is the floor a subscription at that instant would be judged against.
• (*Hub).Snapshot() []ReplayEvent: a copy of the ring, oldest first, each entry's Data cloned; a diagnostic surface. (*Hub).ClientCount() counts subscribed clients and (*Hub).QueuedFrames() sums the frames waiting in their channels, both gauge sources.
• (*Hub).SetMaxClients(n): replaces the cap at runtime for hot-reloaded configuration; lowering it evicts nobody.
• (*Hub).Shutdown(ctx) error: refuses new subscriptions, signals every client to write its sse:reset {"reason":"shutdown"} and blocks until the stream goroutines return, bounded by ctx. Per client the wait is the remainder of a running OnConnect hook plus max(write timeout, 250ms). Call it from webhttp.Run's WithPreDrain hook so streams release before the HTTP drain.
• Wire constants: Wire (the contract revision every hello declares), MaxOffset (2^53 - 1, so a JavaScript peer never rounds an offset), MaxFrameBytes. ParseCursor(s) (Cursor, error) parses a Last-Event-ID value (16 lowercase hex characters, a colon, a decimal offset; the empty string is the zero Cursor with no error; anything else malformed wraps ErrCursor) and Cursor.String() renders it back.

The hello and its verdicts

The first frame of every connection is event: sse:hello carrying Hello{Wire, Epoch, Floor, Head, Resumed, Verdict, KeepaliveMS, KeepaliveEvent} (floor and head are decimal strings on the wire). A client branches on Resumed only; Verdict feeds logs and counters:

• fresh: no cursor was presented.
• resumed: the ring covers every missed frame within the reply cap, and they follow the hello in order.
• gap_floor: the cursor is below the ring's floor, so frames were lost to retention.
• gap_budget: the ring covers the gap but the reply cap would truncate it.
• gap_ahead: the cursor is beyond the head.
• epoch_changed: the cursor belongs to another process lifetime.
• cursor_invalid: the cursor did not parse.

Every verdict other than resumed yields Resumed: false and no replay; the client reconciles from authoritative state instead.

The digest

(*Hub).DigestHandler(resolve Resolver, opts ...DigestOption) http.Handler answers a client returning from sleep: given the versions it holds, pinned to an epoch, which subjects changed or were removed. Resolver is func(ctx, held []Held) ([]State, error), answering exactly one State{Subject, Version, Status} per requested (Kind, Ref) and none the request did not carry (StatusCurrent, StatusGone, StatusForbidden); it runs on the request goroutine, so the application bounds it in time and in flight (webhttp.RouteTimeout in the example above). Mount the handler inside the application's authentication and cross-origin middleware; it performs neither check. Options: WithDigestMaxSubjects(n) (default 256; a larger request is 400) and WithDigestMaxBody(n) (default 512 KiB; a larger body is 413).

Direction	JSON
Request (POST, application/json)	{"epoch": "<hex16>", "subjects": [{"kind": "chat", "ref": "c1", "version": "7"}]}
Response (200 for every well-formed request)	{"epoch": "<hex16>", "floor": "0", "head": "42", "checked": 1, "must_refetch": false, "changed": [{"kind": "chat", "ref": "c1", "version": "9"}], "removed": [{"kind": "chat", "ref": "c2", "reason": "gone"}]}; reason is gone or forbidden

must_refetch is true when the request epoch is absent or not this hub's, when the resolver fails, or when its output does not match the request by key; changed then holds nothing and the client refetches everything. Validation refusals are 400 digest_invalid naming the first violated rule, 405 with Allow: POST, 413 digest_too_large, and 415 digest_content_type.

Presence

WithPresence(fn func(PresenceEvent)) installs one hook that sees each arrival after its hello was flushed and exactly one departure per arrival, on the stream's own goroutine and never under the hub lock; a panic in it is recovered and logged. PresenceEvent{At, Kind, Topic, Verdict, Cause, Write, Epoch, Tag, ClientID} carries Kind PresenceConnected or PresenceDisconnected (the PresenceKind strings connected and disconnected); a departure's Cause is closed (the request context ended), dead (a write failed with the socket still open; Write names keepalive, frame or hook), evicted (reset as slow), shutdown, or hook_failed (OnConnect returned an error). ClientID is minted per connection, so two tabs are two clients; Tag is the WithClientTag value the application folds them by. Dead is observability of socket death, not a liveness bound: the kernel may take minutes to notice.

Testing against the hub

github.com/cplieger/sse/ssetest is the test seam consumers reuse: Serve(t, hub, opts...) starts an httptest server for a hub and returns its URL, ReadFrames(r, n) parses dispatched frames off the wire in one call, FrameReader reads successive batches off one live stream through a single buffered reader, Recorder is a ResponseRecorder that flushes but answers http.ErrNotSupported to the deadline setters, and Fixture is the controllable server (publish, stall, restart, delay, mutate) the TypeScript suites drive through the ssetest/cmd binary.

The TypeScript client

@cplieger/sse is published from web/ to npm and JSR at the repository's release tag. createStream owns the connection over fetch and ReadableStream: it presents the held cursor, validates the hello, measures liveness on bytes against max(3 × keepalive_ms, 15s), closes a hidden tab's stream and reopens it on visibility, backs off with full jitter, and holds incoming frames while the application's revalidate runs; createVersionMap and createDigestClient drive the digest, and createWorkerHost with attachToWorker share one connection across a profile's tabs through a SharedWorker with a per-tab fallback.

Contributing

Issues and PRs are welcome. See CONTRIBUTING.md for the conventions and how to run the checks locally.

Disclaimer

This project is built with care and follows security best practices, but it is intended for personal / self-hosted use. No guarantees of fitness for production environments. Use at your own risk.

This project was built with AI-assisted tooling using Claude, GPT, and Kiro. The human maintainer defines architecture, supervises implementation, and makes all final decisions.

License

Apache-2.0. See LICENSE.

Documentation

Overview

Package sse is a broadcast hub for Server-Sent Events whose clients resume from an exact position: every frame carries an "<epoch>:<offset>" id, every connection opens with a hello that says whether the presented cursor was honoured, and a hub refuses at construction any retention it could not keep.

A Hub owns the ring and the subscriber set; Serve adapts one HTTP request into a subscriber. The client presents its last id as Last-Event-ID and the hello answers with the epoch, the ring's floor and head, the keepalive and a Verdict: Resumed is true only when the ring covers every missed frame inside the reply cap, and a fresh connection, a cursor below the floor, another epoch's cursor or a malformed one all yield false, so the client reconciles.

Every timing constant both halves share is written once in timing.json and pinned by a test in each language; the TypeScript client is published as @cplieger/sse from web/ and documented in web/README.md.

Index

• Constants
• Variables
• type Cursor
  • func ParseCursor(s string) (Cursor, error)
  • func (c Cursor) String() string
• type DigestOption
  • func WithDigestMaxBody(n int64) DigestOption
  • func WithDigestMaxSubjects(n int) DigestOption
• type Event
• type Held
• type Hello
• type Hub
  • func MustNew(opts ...Option) *Hub
  • func New(opts ...Option) (*Hub, error)
  • func (h *Hub) ClientCount() int
  • func (h *Hub) DigestHandler(resolve Resolver, opts ...DigestOption) http.Handler
  • func (h *Hub) Position() Position
  • func (h *Hub) Publish(ev Event) (uint64, error)
  • func (h *Hub) QueuedFrames() int
  • func (h *Hub) Serve(w http.ResponseWriter, r *http.Request, opts ...ServeOption)
  • func (h *Hub) SetMaxClients(n int)
  • func (h *Hub) Shutdown(ctx context.Context) error
  • func (h *Hub) Snapshot() []ReplayEvent
• type Option
  • func WithClientBuffer(n int) Option
  • func WithKeepalive(d time.Duration) Option
  • func WithKeepaliveEvent(name string) Option
  • func WithLogger(l *slog.Logger) Option
  • func WithMaxClients(n int) Option
  • func WithPresence(fn func(PresenceEvent)) Option
  • func WithReconnectDelay(d time.Duration) Option
  • func WithReplay(n int) Option
  • func WithReplayMaxBytes(n int) Option
  • func WithReplayTTL(d time.Duration) Option
  • func WithReplyMaxEvents(n int) Option
  • func WithWriteTimeout(d time.Duration) Option
• type Position
• type PresenceCause
• type PresenceEvent
• type PresenceKind
• type ReplayEvent
• type Resolver
• type ServeOption
  • func OnConnect(fn func(w *Writer, h Hello) error) ServeOption
  • func WithClientTag(tag string) ServeOption
  • func WithTopic(topic string) ServeOption
• type State
• type Status
• type Subject
• type Verdict
• type Writer
  • func (sw *Writer) Event(name string, data []byte) error

Examples

• Hub.DigestHandler
• Hub.Serve
• New

Constants

const MaxFrameBytes = 1 << 20

MaxFrameBytes is the largest encoded frame, terminating blank line included. Publish and Writer.Event return ErrFrameTooLarge above it.

const MaxOffset uint64 = 1<<53 - 1

MaxOffset is the largest offset ever emitted or accepted: the JavaScript exact-integer bound, so a peer holding offsets as Number never rounds one.

const Wire = 1

Wire is the revision of the stream contract every hello declares.

Variables

var (
	// ErrFrameTooLarge is returned by Publish and Writer.Event when the encoded
	// frame would exceed MaxFrameBytes; it is wrapped with the size and the cap.
	ErrFrameTooLarge = errors.New("sse: frame exceeds MaxFrameBytes")
	// ErrInvalidUTF8 is returned by Publish and Writer.Event when Data is not
	// valid UTF-8; it is wrapped with the byte offset of the first invalid byte.
	ErrInvalidUTF8 = errors.New("sse: data is not valid UTF-8")
)

var ErrConfig = errors.New("sse: incoherent configuration")

ErrConfig is wrapped by every construction-time refusal New returns; the message names the offending option or pair.

var ErrCursor = errors.New("sse: invalid cursor")

ErrCursor is returned by ParseCursor, wrapped with the reason, for any non-empty string that is not a well-formed "<epoch>:<offset>".

Types

type Cursor

type Cursor struct {
	Epoch  string
	Offset uint64
}

Cursor is a stream position: the hub epoch that issued Offset, and Offset itself. The zero Cursor means no position.

func ParseCursor

func ParseCursor(s string) (Cursor, error)

ParseCursor parses a Last-Event-ID value. The empty string is no cursor and yields the zero Cursor with a nil error. Anything else must be exactly 16 lowercase hex characters, a colon and a decimal offset with no leading zero at most MaxOffset; a bare integer is invalid.

func (Cursor) String

func (c Cursor) String() string

String renders the cursor as "<epoch>:<offset>", the id: field's value.

type DigestOption

type DigestOption func(*digestConfig)

DigestOption configures one DigestHandler.

func WithDigestMaxBody

func WithDigestMaxBody(n int64) DigestOption

WithDigestMaxBody caps the request body in bytes (default 512 KiB, sized so a full default batch of worst-case escaped refs and versions cannot 413). Non-positive keeps the default.

func WithDigestMaxSubjects

func WithDigestMaxSubjects(n int) DigestOption

WithDigestMaxSubjects caps the subjects one request may carry (default 256); a larger request is 400. Non-positive keeps the default.

type Event

type Event struct {
	// Topic scopes delivery: "" reaches every client, any other value only the
	// clients subscribed to that topic.
	Topic string
	// Name is the event: field. It may be empty (no event: line; a browser
	// dispatches the frame as "message"), must not start with "sse:", and must
	// not contain CR or LF.
	Name string
	// Data is the data: payload, UTF-8 text split on CRLF, CR and LF into one
	// data: line each. Publish takes ownership of the slice: the ring retains
	// it until eviction and replay reads it, so a caller that writes to it
	// afterwards changes replay bytes after the UTF-8 and size checks passed.
	// Marshal fresh per publish.
	Data []byte
}

Event is one server-sent event to publish.

type Held

type Held struct {
	Subject
	Version string
}

Held is a Subject with the version the client presents for it.

type Hello

type Hello struct {
	Epoch          string  `json:"epoch"`
	Verdict        Verdict `json:"verdict"`
	KeepaliveEvent string  `json:"keepalive_event"`
	Floor          uint64  `json:"floor,string"`
	Head           uint64  `json:"head,string"`
	Wire           int     `json:"wire"`
	KeepaliveMS    int     `json:"keepalive_ms"`
	Resumed        bool    `json:"resumed"`
}

Hello is the handshake frame's payload, written once per connection right after the retry: field and handed to the OnConnect hook unchanged. Floor and Head are decimal strings on the wire because the peer is a JavaScript client.

type Hub

type Hub struct {
	// contains filtered or unexported fields
}

Hub is a broadcast fan-out for Server-Sent Events with a replay ring. The zero value is not usable; construct with New. Safe for concurrent use by any number of publishers and subscribers.

func MustNew

func MustNew(opts ...Option) *Hub

MustNew is New for package-level construction in main; it panics with the wrapped ErrConfig message on an incoherent option set.

func New

func New(opts ...Option) (*Hub, error)

New returns a Hub configured by opts, or an error wrapping ErrConfig for an incoherent option set. A nil option is skipped.

Example

package main

import (
	"errors"
	"fmt"
	"time"

	"github.com/cplieger/sse"
)

func main() {
	hub, err := sse.New(
		sse.WithReplay(1024),
		sse.WithReplayTTL(10*time.Minute),
		sse.WithReplyMaxEvents(256),
	)
	if err != nil {
		panic(err)
	}

	offset, err := hub.Publish(sse.Event{Name: "notify", Data: []byte(`{"n":1}`)})
	fmt.Println(offset, err)

	pos := hub.Position()
	fmt.Println(pos.Floor, pos.Head)

	// A ring without an age bound is refused at construction.
	_, err = sse.New(sse.WithReplay(1024))
	fmt.Println(errors.Is(err, sse.ErrConfig))
}

Output:
1 <nil>
1 1
true

func (*Hub) ClientCount

func (h *Hub) ClientCount() int

ClientCount returns the number of subscribed clients: registered and not yet reset as slow. A connection inside its OnConnect hook is counted; one that Publish evicted is not, even while its goroutine is still writing the reset. After Shutdown it counts the goroutines still draining.

func (*Hub) DigestHandler

func (h *Hub) DigestHandler(resolve Resolver, opts ...DigestOption) http.Handler

DigestHandler answers POST application/json digest requests: which of the client's held subjects changed or were removed, pinned to the hub's epoch and head. Mount it inside the application's authentication and cross-origin middleware; it performs neither check itself. The response is must_refetch when the request epoch is absent or not this hub's, when resolve fails, or when its output does not match the request by key; every well-formed request is answered 200.

Example

package main

import (
	"context"
	"encoding/json"
	"fmt"
	"net/http/httptest"
	"strings"
	"time"

	"github.com/cplieger/sse"
	"github.com/cplieger/webhttp/v3"
)

func main() {
	hub := sse.MustNew()
	current := map[sse.Subject]string{{Kind: "chat", Ref: "c1"}: "7"}
	resolve := func(_ context.Context, held []sse.Held) ([]sse.State, error) {
		states := make([]sse.State, 0, len(held))
		for _, h := range held {
			version, ok := current[h.Subject]
			if !ok {
				states = append(states, sse.State{Subject: h.Subject, Status: sse.StatusGone})
				continue
			}
			states = append(states, sse.State{Subject: h.Subject, Version: version})
		}
		return states, nil
	}
	srv := httptest.NewServer(webhttp.RouteTimeout(hub.DigestHandler(resolve), 10*time.Second, "digest timed out"))
	defer srv.Close()

	body := fmt.Sprintf(`{"epoch":%q,"subjects":[{"kind":"chat","ref":"c1","version":"5"},{"kind":"chat","ref":"c2","version":"1"}]}`, hub.Position().Epoch)
	resp, err := srv.Client().Post(srv.URL, "application/json", strings.NewReader(body))
	if err != nil {
		panic(err)
	}
	defer resp.Body.Close()

	var out struct {
		Changed []sse.Held `json:"changed"`
		Removed []struct {
			sse.Subject
			Reason sse.Status `json:"reason"`
		} `json:"removed"`
		MustRefetch bool `json:"must_refetch"`
	}
	if err := json.NewDecoder(resp.Body).Decode(&out); err != nil {
		panic(err)
	}
	fmt.Println(resp.StatusCode, out.MustRefetch)
	fmt.Println(out.Changed)
	fmt.Println(out.Removed)
}

Output:
200 false
[{{chat c1} 7}]
[{{chat c2} gone}]

func (*Hub) Position

func (h *Hub) Position() Position

Position returns the replay window after evicting expired entries, so its Floor is the one a subscription at this instant would be judged against.

func (*Hub) Publish

func (h *Hub) Publish(ev Event) (uint64, error)

Publish assigns the event the next offset, appends it to the ring and fans it out to every matching client without blocking; a client whose buffer is full is reset as slow and dropped. It returns the offset, or (0, nil) on a nil *Hub and for a valid frame after Shutdown (dropped). A Name outside Event's grammar panics; Data that is not UTF-8 returns ErrInvalidUTF8 and an encoded frame above MaxFrameBytes returns ErrFrameTooLarge, and neither refusal consumes an offset or touches the ring, before or after Shutdown.

func (*Hub) QueuedFrames

func (h *Hub) QueuedFrames() int

QueuedFrames returns the sum of frames queued across client channels.

func (*Hub) Serve

func (h *Hub) Serve(w http.ResponseWriter, r *http.Request, opts ...ServeOption)

Serve subscribes the request to the hub and streams until the peer leaves, the request context ends, the client is reset as slow, a write fails, or the hub shuts down. It owns the response headers, the retry: field, the hello, the Last-Event-ID replay, keepalives and the sse:reset frame.

It answers 503 sse_unavailable over the client cap or after Shutdown, and 500 streaming_unsupported when no http.Flusher is reachable through the ResponseWriter or its Unwrap chain.

Example

package main

import (
	"context"
	"encoding/json"
	"fmt"
	"net/http"
	"net/http/httptest"
	"time"

	"github.com/cplieger/sse"
	"github.com/cplieger/sse/ssetest"
)

func main() {
	hub := sse.MustNew(sse.WithReplay(64), sse.WithReplayTTL(10*time.Minute))

	mux := http.NewServeMux()
	mux.HandleFunc("GET /api/events", func(w http.ResponseWriter, r *http.Request) {
		hub.Serve(w, r,
			sse.OnConnect(func(w *sse.Writer, h sse.Hello) error {
				return w.Event("connected", fmt.Appendf(nil, `{"resumed":%t}`, h.Resumed))
			}),
			sse.WithClientTag(r.Header.Get("SSE-Client")),
		)
	})
	srv := httptest.NewServer(mux)
	defer srv.Close()

	// A first connection presents no cursor: the hello is fresh.
	first, err := open(srv, "")
	if err != nil {
		panic(err)
	}
	frames, err := ssetest.ReadFrames(first.Body, 3)
	if err != nil {
		panic(err)
	}
	fmt.Println(frames[0].Event, frames[0].Data)
	fmt.Println(frames[1].Event, describe(frames[1].Data))
	fmt.Println(frames[2].Event, frames[2].Data)

	if _, err := hub.Publish(sse.Event{Name: "notify", Data: []byte(`{"n":1}`)}); err != nil {
		panic(err)
	}
	live, err := ssetest.ReadFrames(first.Body, 1)
	if err != nil {
		panic(err)
	}
	cursor, err := sse.ParseCursor(live[0].ID)
	if err != nil {
		panic(err)
	}
	fmt.Println(live[0].Event, cursor.Offset, live[0].Data)
	first.Body.Close()

	// A frame published while the client is away is replayed on resume.
	if _, err := hub.Publish(sse.Event{Name: "notify", Data: []byte(`{"n":2}`)}); err != nil {
		panic(err)
	}
	second, err := open(srv, cursor.String())
	if err != nil {
		panic(err)
	}
	defer second.Body.Close()
	frames, err = ssetest.ReadFrames(second.Body, 4)
	if err != nil {
		panic(err)
	}
	fmt.Println(frames[1].Event, describe(frames[1].Data))
	fmt.Println(frames[2].Event, frames[2].ID == sse.Cursor{Epoch: cursor.Epoch, Offset: 2}.String(), frames[2].Data)
	fmt.Println(frames[3].Event, frames[3].Data)

	// Shutdown ends every stream with a reset frame naming the reason.
	ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
	defer cancel()
	fmt.Println(hub.Shutdown(ctx))
	reset, err := ssetest.ReadFrames(second.Body, 1)
	if err != nil {
		panic(err)
	}
	fmt.Println(reset[0].Event, reset[0].Data)
}

func open(srv *httptest.Server, cursor string) (*http.Response, error) {
	req, err := http.NewRequestWithContext(context.Background(), http.MethodGet, srv.URL+"/api/events", nil)
	if err != nil {
		return nil, err
	}
	req.Header.Set("Accept", "text/event-stream")
	req.Header.Set("SSE-Client", "tab-1")
	if cursor != "" {
		req.Header.Set("Last-Event-ID", cursor)
	}
	return srv.Client().Do(req)
}

func describe(hello string) string {
	var h sse.Hello
	if err := json.Unmarshal([]byte(hello), &h); err != nil {
		panic(err)
	}
	return fmt.Sprintf("%s resumed=%t head=%d", h.Verdict, h.Resumed, h.Head)
}

Output:
retry 1500
sse:hello fresh resumed=false head=0
connected {"resumed":false}
notify 1 {"n":1}
sse:hello resumed resumed=true head=2
notify true {"n":2}
connected {"resumed":true}
<nil>
sse:reset {"reason":"shutdown"}

func (*Hub) SetMaxClients

func (h *Hub) SetMaxClients(n int)

SetMaxClients replaces the client cap at runtime (0 or negative means unlimited). Lowering it evicts nobody; admission enforces it.

func (*Hub) Shutdown

func (h *Hub) Shutdown(ctx context.Context) error

Shutdown refuses new subscriptions, signals every client to write its sse:reset and waits for the stream goroutines to return, bounded by ctx. Per client the wait is the remainder of a running OnConnect hook plus max(write timeout, 250ms). Returns ctx.Err() when ctx expires first.

func (*Hub) Snapshot

func (h *Hub) Snapshot() []ReplayEvent

Snapshot returns a copy of the ring, oldest first, with each entry's Data cloned so the caller cannot change replay bytes. A diagnostic surface.

type Option

type Option func(*config) error

Option configures a Hub at construction. An option refuses its own value on the spot; rules that compare two options run after every option has been applied, so option order never changes the verdict.

func WithClientBuffer

func WithClientBuffer(n int) Option

WithClientBuffer sets the per-client channel capacity (default max(ring capacity, 256)). A client that falls this many frames behind is reset as slow. Refused at or below zero.

func WithKeepalive

func WithKeepalive(d time.Duration) Option

WithKeepalive sets the keepalive interval (default 15s), announced to the client as hello.keepalive_ms. Refused below one millisecond.

func WithKeepaliveEvent

func WithKeepaliveEvent(name string) Option

WithKeepaliveEvent names the keepalive frame's event (default "sse:keepalive"). The empty string selects the comment form ": keepalive", invisible to EventSource consumers. A name holding CR or LF is refused.

func WithLogger

func WithLogger(l *slog.Logger) Option

WithLogger sets the logger for connection and eviction diagnostics (default slog.Default()). A nil logger keeps the default.

func WithMaxClients

func WithMaxClients(n int) Option

WithMaxClients caps concurrent clients; Serve answers 503 beyond it. Zero or negative means unlimited.

func WithPresence

func WithPresence(fn func(PresenceEvent)) Option

WithPresence installs the one hook that receives every PresenceEvent (nil removes it). It runs on the arriving or departing stream's own goroutine, after the hello flush or after the client is unregistered, never under the hub lock, so a slow hook delays only that stream. A panic in it is recovered and logged at Error.

func WithReconnectDelay

func WithReconnectDelay(d time.Duration) Option

WithReconnectDelay sets the stream's retry: field (default 1500ms), the wait a client observes before reconnecting. Refused below one millisecond, since 0 on the wire means reconnect at once.

func WithReplay

func WithReplay(n int) Option

WithReplay sets the ring capacity in events (default 0, no replay). A non-zero capacity requires WithReplayTTL. Negative is refused.

func WithReplayMaxBytes

func WithReplayMaxBytes(n int) Option

WithReplayMaxBytes caps the sum of encoded frame bytes the ring holds (default 64 × MaxFrameBytes); the oldest entries are evicted when an append would exceed it. Refused below MaxFrameBytes.

func WithReplayTTL

func WithReplayTTL(d time.Duration) Option

WithReplayTTL sets the maximum age of a ring entry (no default; required when WithReplay is non-zero). Refused at or below zero, and refused below max(3 × keepalive, 15s) + 30s, the window a client needs to reconnect after a half-open stream.

func WithReplyMaxEvents

func WithReplyMaxEvents(n int) Option

WithReplyMaxEvents caps the replay one connection may receive (default min(ring capacity, 256)); a resume needing more is gap_budget. Refused when negative or above the ring capacity.

func WithWriteTimeout

func WithWriteTimeout(d time.Duration) Option

WithWriteTimeout bounds every write to a client (default 2 × keepalive). Refused at or below the keepalive interval.

type Position

type Position struct {
	Epoch string
	Floor uint64
	Head  uint64
}

Position is the hub's replay window at one instant: the epoch, the oldest offset the ring can replay (0 when empty) and the newest offset published.

type PresenceCause

type PresenceCause string

PresenceCause says why a client left, on a disconnected PresenceEvent.

const (
	PresenceClosed     PresenceCause = "closed"
	PresenceDead       PresenceCause = "dead"
	PresenceEvicted    PresenceCause = "evicted"
	PresenceShutdown   PresenceCause = "shutdown"
	PresenceHookFailed PresenceCause = "hook_failed"
)

The causes a disconnected event carries. Closed and dead are the socket's account; evicted and shutdown are the two sse:reset reasons; hook_failed is the application ending its own connection from OnConnect.

type PresenceEvent

type PresenceEvent struct {
	At    time.Time
	Kind  PresenceKind
	Topic string
	// Verdict is the hello's verdict on connected; empty on disconnected.
	Verdict Verdict
	// Cause is set on disconnected; empty on connected.
	Cause PresenceCause
	// Write names the failed write: "keepalive", "frame" or "hook" on a dead
	// event, "reset" on an evicted or shutdown event whose reset write failed.
	Write string
	Epoch string
	// Tag is the WithClientTag value, or "" when none was passed or it failed
	// the grammar. A presence table folds clients by it.
	Tag string
	// ClientID is process-unique and minted at subscribe, so two connects from
	// one browser are two clients.
	ClientID uint64
}

PresenceEvent is one arrival or departure as WithPresence reports it. A connected event is delivered once per Serve that flushed its hello, and exactly one disconnected event follows it; a Serve that failed before the hello produces neither. Dead means a write or flush failed with the socket still open, which the kernel may take minutes to notice: it is observability of socket death, not a liveness bound.

type PresenceKind

type PresenceKind string

PresenceKind says whether a PresenceEvent is an arrival or a departure.

const (
	PresenceConnected    PresenceKind = "connected"
	PresenceDisconnected PresenceKind = "disconnected"
)

The two kinds a PresenceEvent carries.

type ReplayEvent

type ReplayEvent struct {
	At     time.Time
	Event  Event
	Offset uint64
}

ReplayEvent is one ring entry as Snapshot returns it.

type Resolver

type Resolver func(ctx context.Context, held []Held) ([]State, error)

Resolver answers the current State of every held subject, exactly one per requested (Kind, Ref) and none the request did not carry. It runs on the request goroutine with the request's context and no library-level concurrency bound; the application bounds it in time and in flight. The principal comes from ctx however the application's middleware stored it.

type ServeOption

type ServeOption func(*serveConfig)

ServeOption configures one Serve call.

func OnConnect

func OnConnect(fn func(w *Writer, h Hello) error) ServeOption

OnConnect installs a hook that runs after the retry: field, the hello and the replay have been flushed, before live delivery. It receives the Hello the client received and writes initial-state frames through the Writer; an error ends the connection. The hook runs on the stream goroutine, holds no hub lock, and may call Publish and Position. Keep it short: a hook that runs past the client's hello timeout loses the tab to a reconnect.

func WithClientTag

func WithClientTag(tag string) ServeOption

WithClientTag attaches an application-chosen presence key to the connection. The empty string means no tag, so a handler may pass the SSE-Client header through unconditionally; any other tag outside [A-Za-z0-9_-]{1,64} is treated as absent with one Warn naming its length and the class of failure, never the bytes.

func WithTopic

func WithTopic(topic string) ServeOption

WithTopic subscribes the connection to broadcasts plus events scoped to topic. The empty topic (the default) receives everything.

type State

type State struct {
	Subject
	Version string
	Status  Status
}

State is a resolver's answer for one requested subject.

type Status

type Status string

Status is a resolver's answer for a subject it no longer serves.

const (
	StatusCurrent   Status = ""
	StatusGone      Status = "gone"
	StatusForbidden Status = "forbidden"
)

StatusCurrent means the subject exists and Version is its current version; StatusGone means it no longer exists; StatusForbidden means it exists and the principal may not read it. In a multi-principal deployment the resolver decides whether the principal may learn the distinction and answers gone for both when it may not.

type Subject

type Subject struct {
	Kind string
	Ref  string
}

Subject is one application-defined unit of state the digest compares: a Kind the application chooses and an opaque Ref within it.

type Verdict

type Verdict string

Verdict is the hello's diagnostic account of how a presented cursor was judged. Clients branch on Hello.Resumed only; the verdict feeds logs and counters.

const (
	VerdictFresh         Verdict = "fresh"
	VerdictResumed       Verdict = "resumed"
	VerdictGapFloor      Verdict = "gap_floor"
	VerdictGapBudget     Verdict = "gap_budget"
	VerdictGapAhead      Verdict = "gap_ahead"
	VerdictEpochChanged  Verdict = "epoch_changed"
	VerdictCursorInvalid Verdict = "cursor_invalid"
)

The seven verdicts. The three gap variants are kept distinct so a counter can tell a retention problem from a reply-cap problem from a client defect.

type Writer

type Writer struct {
	// contains filtered or unexported fields
}

Writer is handed to OnConnect only; frames it writes carry no id. Serve always supplies it, so there is no constructor and no useful zero value.

func (*Writer) Event

func (sw *Writer) Event(name string, data []byte) error

Event writes one frame as its own bounded write and flush; name may be empty. It returns ErrFrameTooLarge or ErrInvalidUTF8 exactly as Publish does (the size check has no id line here), panics on the same names, and returns the deadline error when the peer stopped reading, which the hook propagates.