Server-Sent Events

Router.SSE registers a route whose response is a stream rather than a body. It is the other half of what a WebSocket is usually reached for, and the simpler half: the server sends, the client listens, and a browser reads it natively with EventSource, reconnecting on its own when a stream drops.

// An event stream route is declared like any other and answers 200 rather than
// upgrading: the guards run and the input binds before a byte of the stream is
// written, and the handler owns the stream until it returns.
r.SSE("/items/stream", handlers.StreamItems,
    muzak.Summary("Follow every change to the items"),
    muzak.WithSSE(muzak.SSEOptions{
        // A stream that says nothing for long enough is closed by proxies that
        // believe it to be idle, so a comment goes out instead.
        KeepAlive: 15 * time.Second,
        // A browser reconnects on its own when a stream ends, and this is how
        // soon.
        Retry: 2 * time.Second,
    }))

func StreamItems(ctx *muzak.Context, _ muzak.Empty, stream *muzak.SSEStream[schemas.ItemOut]) error {
    store := muzak.From[*core.ItemStore](ctx)

    updates := store.Watch(stream.Context())

    for {
        select {
        case <-stream.Context().Done():
            // The client went away or the server is shutting down. Either way
            // the conversation is over, and it is not a failure.
            return nil
        case change := <-updates:
            item := schemas.ItemOut{ID: change.Item.ID, Name: change.Item.Name}
            if err := stream.Send(item); err != nil {
                return err
            }
        }
    }
}

The type parameter is the contract: nothing but an ItemOut can be sent, and the generated document describes the stream with that type, in the same way a handler's return type describes an ordinary response.

curl -N 'http://localhost:8080/items/stream'

retry: 2000

event: item_update
id: 1
data: {"id":"plumbus","name":"Plumbus"}

event: item_update
id: 2
data: {"id":"fleeb","name":"Fleeb"}

The retry field is sent once, before anything else, because SSEOptions.Retry was set on the route above: a client that loses the stream on the very first event already knows how long to wait. The response carries Content-Type: text/event-stream; charset=utf-8, Cache-Control: no-cache, no-transform and X-Accel-Buffering: no. The type is set whatever a middleware set before it, because a stream served under any other type, such as the application/json a middleware may give every response in advance, is one EventSource refuses to read. The other two are left alone when something already set them.

An ordinary route

Middleware runs, guards run, dependencies resolve, and the input is bound and validated before a byte of the stream is written. A request that fails any of that is answered with the usual JSON error and never becomes a stream at all.

The response header is written before the handler is called, which is what lets a client see the stream open immediately. Anything that decides whether to serve a stream at all therefore belongs in a guard or a dependency, where there is still a response to say it in.

The stream

Nothing on SSEStream takes a context, unlike WSConn, because a stream belongs to one request.

stream.Context() is derived from the request's own and is cancelled when the client disconnects, when the request ends, or when the server begins shutting down. That makes it the one thing a handler has to watch, and every send after it reports muzak.ErrSSEStreamEnded, so a handler that only sends ends on its own.

for update := range updates {
    if err := stream.Send(update); err != nil {
        return err
    }
}

A handler that returns ErrSSEStreamEnded is treated as a stream that finished rather than one that failed, so a client going away is not logged as an error. Test for it with errors.Is; the reason the stream ended is wrapped inside.

Tracing draws the same line. With tracing on, the request's span ends when the stream does, and it is marked an error only when the handler fails, not when the client goes away. AppOptions.Observer is told about the stream once, after it has ended. See Observability.

A value the route's providers acquired with Acquire lives as long as the stream. Its release runs as soon as the handler returns and is handed what the handler returned, so a Transaction held for the stream commits only when the handler returns nil: a stream the client left ends with ErrSSEStreamEnded, and is rolled back. A release that fails is logged as the stream's failure. See Dependencies.

Writes are serialized, so any number of goroutines may write to one stream and each event goes out whole.

A stream is single use. The first failure ends it, and every later send reports that same error rather than writing into a response nobody is reading. Sends that report a mistake instead, such as an event name carrying a line break, leave the stream usable.

A stream never writes into a response it has handed back. A send still encoding when the handler returned held the write lock, and the stream used to give up after WriteTimeout and return the response anyway, which crashed the process or could write one client's event into the next request on the connection.

Events with more than data

type SSEEvent[Out any] struct {
    Name      string        // the type a browser dispatches it under
    ID        string        // what a reconnect resumes from
    Retry     time.Duration // how long to wait before reconnecting
    Comment   string        // text no client acts on
    Data      *Out          // the value, encoded as JSON
    Text      string        // a payload written as it stands
    EmptyData bool          // a data field even with no payload
}

// sendChange writes one change as an event a browser can listen for by name
// and resume from by identifier.
func sendChange(stream *muzak.SSEStream[schemas.ItemOut], change core.Change) error {
    item := schemas.ItemOut{ID: change.Item.ID, Name: change.Item.Name}
    return stream.SendEvent(muzak.SSEEvent[schemas.ItemOut]{
        Name: "item_update",
        ID:   strconv.Itoa(change.Seq),
        Data: &item,
    })
}

const source = new EventSource("/items/stream")
source.addEventListener("item_update", (event) => {
    const item = JSON.parse(event.data)
    // ...
})

Data is a pointer so that an event carrying nothing can be told from one carrying a zero value, and it cannot be combined with Text. An event setting both is refused.

A client dispatches an event only when it has a data field, which it has when it carries Data, a Text that is not empty, or EmptyData. An event without one is still written, and a browser and SSEReader alike apply its ID and Retry, but neither hands it to the application, Name included. The zero value is the extreme case: a blank line that no client acts on. EmptyData writes a data field with nothing in it, which is dispatched as an event whose data is the empty string. That is how a stream of text sends an empty message, as in SSEEvent{Text: line, EmptyData: true} for a log line that may be blank, and how an event that is only a name, such as a tick, is dispatched at all.

Retry is whole milliseconds, and a positive delay shorter than a millisecond is written as 1, where it used to be truncated to 0 and told a client to reconnect at once.

Name and ID may not contain a line break. An event stream is a sequence of lines, so a break in one of those fields would end it and let whatever followed be read as fields of its own, which on a stream carrying one client's input to another is event forgery. Such a send is refused rather than repaired, and the stream stays usable.

Payloads that are not JSON

Text writes a payload as it stands, for a log line or for the sentinel some completion APIs end with. A stream whose events are all text declares muzak.Empty as its model, and the generated document then says there is no schema rather than describing one that does not exist.

// StreamChat answers a prompt one token at a time, which is the shape every
// chat completion API streams in.
func StreamChat(ctx *muzak.Context, in schemas.ChatIn, stream *muzak.SSEStream[muzak.Empty]) error {
    for word := range strings.SplitSeq(in.Text, " ") {
        select {
        case <-stream.Context().Done():
            // The client closed the tab, or the server is shutting down. There
            // is no one left to answer.
            return nil
        case <-time.After(tokenDelay):
        }
        if err := stream.SendEvent(muzak.SSEEvent[muzak.Empty]{Name: "token", Text: word}); err != nil {
            return err
        }
    }
    // The sentinel some clients expect at the end of a completion. It is text
    // rather than a value, which is what Text is for.
    return stream.SendEvent(muzak.SSEEvent[muzak.Empty]{Name: "done", Text: "[DONE]"})
}

A payload spanning several lines is written as several data lines and arrives whole, which is both what the format asks for and what stops a value from ending its own field.

A stream reached by POST

A stream is not tied to GET. Router.SSEHandle registers one for any method, which is what a protocol that streams its answer to a posted document needs, and there the input binds a request body like any other route.

r.SSEHandle(http.MethodPost, "/chat/stream", handlers.StreamChat,
    muzak.Summary("Answer a prompt one token at a time"),
    muzak.WithSSE(muzak.SSEOptions{
        // A completion is never quiet for long, so the keepalive is only there
        // for the pause before the first token.
        KeepAlive: 10 * time.Second,
    }))

// ChatIn is the prompt a chat stream answers.
//
// It is an ordinary JSON body: an event stream reached by POST binds and
// validates one exactly as any other route does.
type ChatIn struct {
    Text string `json:"text" doc:"The prompt to answer, one token at a time"`
}

// Validate bounds the prompt, and is applied before a byte of the stream is
// written.
func (in *ChatIn) Validate(v *muzak.Validation) {
    v.String(&in.Text).Trim().Required().MinLen(1).MaxLen(280)
}

curl -N -X POST 'http://localhost:8080/chat/stream' \
     -H 'Content-Type: application/json' -d '{"text":"what is a plumbus"}'

An input that binds nothing from the body leaves it to the handler, which reads it from ctx.Request().Body even though the stream's header has already gone out, over HTTP/1.1 as over HTTP/2. It is read under ServerOptions.ReadTimeout, as any route's body is, so a client that declares a body and sends half of it cannot hold the handler for longer than that. Read it before sending the first event, because a client may send the whole body before it reads any of the response. Until the body has been read to its end, net/http cannot watch the connection for the client going away, so a disconnect is noticed when a write fails rather than at once, which with the keepalive on takes a couple of its intervals.

Resuming a dropped stream

A browser remembers the last identifier it saw and sends it back in Last-Event-ID when it reconnects. stream.LastEventID() reads it, and that is what turns a dropped connection into a stream that picks up where it left off rather than one that starts again.

func StreamItems(ctx *muzak.Context, _ muzak.Empty, stream *muzak.SSEStream[schemas.ItemOut]) error {
    store := muzak.From[*core.ItemStore](ctx)

    // A browser sends back the identifier of the last event it saw when its
    // EventSource reconnects. The value is the client's, so one that is not a
    // number is treated as no value at all rather than as an error.
    seen := 0
    if last := stream.LastEventID(); last != "" {
        if parsed, err := strconv.Atoi(last); err == nil {
            seen = parsed
        }
    }

    // Subscribing before the backlog is read is what stops a change made in
    // between from falling through the gap between the two.
    updates := store.Watch(stream.Context())
    for _, change := range store.Since(seen) {
        if err := sendChange(stream, change); err != nil {
            return err
        }
        seen = change.Seq
    }

    for {
        select {
        case <-stream.Context().Done():
            return nil
        case change := <-updates:
            if change.Seq <= seen {
                // Already sent from the backlog above.
                continue
            }
            if err := sendChange(stream, change); err != nil {
                return err
            }
            seen = change.Seq
        }
    }
}

LastEventID is a value the client controls, so treat it as input rather than as a cursor to be trusted.

What a stream bounds

A stream costs a connection and a goroutine for as long as a client cares to hold it.

None of the defaults bounds how long a stream lasts, and on a route anyone can open that leaves the whole budget to whoever takes it first. A client that reads nothing is never ended by WriteTimeout, and with MaxLifetime unset nothing else ends it. MaxStreams divided by MaxStreamsPerIP is how many addresses fill the process: 1024 / 64 = 16 with the defaults, which is sixteen IPv4 addresses, or sixteen IPv6 /56 prefixes, a single /52. Each opens 64 streams and reads none of them, and every other stream is refused with 503 until they let go.

A route that clients you do not control can open is worth more than the defaults:

• Put it behind a guard that authenticates the client where it can be. That is the only thing that ties a stream to someone rather than to an address.
• Raise MaxStreams and lower MaxStreamsPerIP, so that the one divided by the other is larger than the number of addresses an abuser can be expected to have. MaxStreams can go as far as the file descriptor limit allows. How low MaxStreamsPerIP can go depends on who shares an address: one browser needs a stream per open tab, while an office or a carrier's NAT puts many users behind one.
• Count IPv6 clients by their /48 with ClientIPOptions.ConnectionIPv6Prefix: 48, which makes each of those addresses cost a whole site's allocation.
• Set MaxLifetime, to minutes rather than hours, so that a stream whose client has stopped listening is reclaimed and every slot turns over. A browser reconnects and resumes on its own, so a legitimate client loses nothing. MaxLifetime alone does not stop a client that reconnects as soon as its stream ends, which is what the settings above are for.

app := muzak.New(muzak.AppOptions{
    ClientIP: muzak.ClientIPOptions{ConnectionIPv6Prefix: 48},
    SSE: muzak.SSEOptions{
        MaxStreams:      8192,
        MaxStreamsPerIP: 8,
        MaxLifetime:     15 * time.Minute,
    },
})

Clearing the listener's read and write deadlines matters more than it sounds. A stream is a response that does not end, so it would otherwise die at ServerOptions.WriteTimeout however healthy it was, and the read deadline would cancel the request, and with it the stream, at ServerOptions.ReadTimeout and blame the client. The read deadline is cleared once nothing is left of the request body to read. That is as the stream opens, unless the body was left to the handler: then it is cleared when the handler has read the body to its end, so that body is read under ReadTimeout as any route's is.

The stream's own deadline is armed around each write and taken off afterwards. It used to be armed before every event and never removed. On HTTP/2 that deadline is a timer that resets the stream when it fires, so with the default 10 second WriteTimeout and 15 second keepalive every idle stream died before its first keepalive, and browsers reconnected in a loop. Now a graceful shutdown or a handler's return ends an idle stream cleanly rather than aborting it, since an interrupt is applied only to a write that is in flight.

Ending a stream on purpose

SSEOptions.MaxLifetime is unset by default, which leaves a stream open for as long as its handler runs and its client keeps the connection. It is worth setting on any route a client you do not control can open. When it has passed, the stream writes a closing comment and ends the way a finished response does. A browser's EventSource reconnects on its own, sending the identifier of the last event it saw in Last-Event-ID, so a handler that resumes from stream.LastEventID() loses nothing.

muzak.WithSSE(muzak.SSEOptions{MaxLifetime: 30 * time.Minute})

The handler is told the way it is told of a disconnect, through stream.Context() and the ErrSSEStreamEnded every later send reports, and the stream's slot is released when the handler returns. A stream is not cut mid-event: the ceiling is applied between events. Nothing is written after the closing comment, where a keepalive already waiting for the write lock could go out behind it. A negative value removes a bound a wider scope set. A client that does not reconnect on its own, a script using SSEDial for instance, must be written to.

Options

Set them application-wide through AppOptions.SSE and narrow them for a router or a route with WithSSE. Layering works field by field, so a route that only lengthens the keepalive keeps the application's write timeout.

MaxStreams and MaxStreamsPerIP may only be set on the application: the resource they protect is the process, not a route, so a router or a route that sets either is refused when the application is built.

The keepalive exists because a proxy that sees an idle connection for long enough closes it, and because a silent stream is indistinguishable from a dead one. A keepalive arrives every interval, measured from the last write. It used to wait on fixed ticks and skip a tick whose last write was younger than the interval; its own write lands a hair after its tick, so the tick after it always skipped, and a keepalive asked for every 15 seconds arrived at 15, 45 and 60, twice the silence a proxy idle timeout was meant to be kept off with. This was seen behind Cloud Run. The keepalive clock is monotonic, so a wall-clock step cannot starve it.

A stream whose response cannot carry write deadlines is still served, but its WriteTimeout is not enforced. Middleware that wraps the response writer, forwards Flush and has no Unwrap() http.ResponseWriter method hides the connection's deadlines, and the first stream on each route logs a warning naming which writer stopped the chain. Give the wrapper an Unwrap.

What does not apply

Compression leaves an event stream alone. Holding events in a compressor's window until something forces them out is the one thing a stream cannot survive.

The origin check a WebSocket needs has no counterpart here. An EventSource is subject to the same-origin policy and to CORS like any other request, so AppOptions.CORS already governs it. See CORS.

Nothing a handler fails with is disclosed. The response header went out before the handler ran, so a failure ends the stream and the reason goes to the log.

Shutdown

Open streams are tracked, so a graceful shutdown ends every one of them and waits for the handlers, instead of waiting out its whole deadline once per stream. An idle stream ends cleanly. Streams are ended once readiness has turned unavailable and ServerOptions.DrainDelay has passed, as the listeners close and ordinary requests drain, and the background tasks and lifecycle components are dealt with only after the handlers have returned. See Lifecycle. A shutdown can also arrive while a stream is still opening: it is decided under the lock, and the header is written as an armed write a shutdown can stop, so an interrupt landing in that window is not lost.

Documentation

An SSE route is documented with a 200 response carrying text/event-stream, whose schema describes the data field of a single event. OpenAPI 3.1 has no way to say "many of these, one after another", and one event's data is the only part a client has to be able to decode. A stream whose model is muzak.Empty is documented with no schema at all. The 503 a draining or full server answers with is documented too, and a status a route declared with WithResponseDoc keeps its own description.

Testing

func TestStreamItems(t *testing.T) {
    client := testclient.New(t, buildApp())

    stream := client.SSE("/items/stream")

    client.Post("/items/", testclient.JSON(map[string]string{"id": "plumbus", "name": "Plumbus"})).
        AssertStatus(http.StatusCreated)

    item := stream.Decode[schemas.ItemOut]()
    if item.ID != "plumbus" {
        t.Errorf("id = %q, want plumbus", item.ID)
    }
}

muzak.SSEDial is the reading half of the same engine, so a route is tested over a real connection rather than against a second implementation:

reader, _, err := muzak.SSEDial(ctx, "http://"+app.Addr()+"/items/stream", muzak.SSEDialOptions{})
if err != nil {
    return err
}
defer reader.Close()

for {
    message, err := reader.Next(ctx)
    if errors.Is(err, muzak.ErrSSEStreamEnded) {
        return nil
    }
    if err != nil {
        return err
    }
    item, err := message.Decode[schemas.ItemOut]()
    if err != nil {
        return err
    }
    _ = item
}

A response that is not 200 with a text/event-stream body is refused rather than parsed, because a stream reader that quietly accepts an HTML error page reports "no events" for what is actually a failure. SSEDialOptions carries the method, body, headers, a LastEventID to resume from, a read limit, a read timeout, a handshake timeout and KeepComments for seeing the keepalives.

reader.Close may be called from another goroutine, and releases a Next that is waiting. Once a reader has been closed, or a read cancelled or timed out, Next reports ErrSSEStreamEnded even for an event that had already arrived and was waiting in the buffer.

A few things SSEDial does not do, on purpose:

• It does not follow redirects. A 3xx is refused and the response returned with the error. Dial the Location yourself if it is trusted. It used to send custom headers such as X-Api-Key to another host, and Authorization over an https-to-http downgrade.
• Opening a stream is bounded. It removes the supplied client's Timeout so that it does not cut the stream short, and used to put nothing in its place, so a server that accepted the request and never answered held the call for as long as a context without a deadline lasts. SSEDialOptions.HandshakeTimeout now bounds it, defaulting to the Timeout of the supplied HTTPClient when it has one and otherwise to muzak.DefaultSSEHandshakeTimeout (30 seconds); a negative value removes the bound. It ends when the stream is open and says nothing about how long the stream lasts or idles. A server that takes longer than 30 seconds to send response headers needs it set higher, or negative.
• SSEReader.Retry is capped at an hour and accepts digits only. A retry: field was parsed with ParseInt and multiplied into a duration, so a server sending 9223372036855 made Retry() negative, and a client that sleeps for it reconnected in a tight loop. A value with a sign, a space or an exponent is now ignored as the format requires, and a value beyond an hour is reported as an hour.
• SSEReader.LastEventID moves only when an event completes. An id is held in a buffer, as the HTML specification keeps it, and becomes the last event ID when the blank line that ends its event arrives, whether or not the event carried data. An event the stream cut off part way, by dropping, by running past the read timeout or by exceeding the read limit, leaves it where it was, so a reconnect that passes it to SSEDialOptions.LastEventID asks for that event again rather than for the one after it.

Where to go next

WebSockets covers the two-way half, and Testing covers asserting on a stream.