stream

type Limits ¶

type Limits struct {
	PerCredential int
	PerTenant     int
	PerProcess    int
}

Limits is how many streams may be open at once, at each of the three levels a long-lived connection has to be bounded at.

Three rather than one, because they answer three different failures. Per credential stops one client reconnecting in a loop and holding a hundred sockets it never reads. Per tenant stops one busy workspace taking a whole pod's capacity from everybody else on it. Per process is the load shedding of observability-reliability.md §6 applied to the resource a stream actually consumes: above the threshold, new connections are refused with `503` and a `Retry-After` before latency tips over for everyone - the connections already open are never dropped to make room, because shedding is about not accepting more work, not about abandoning work in hand.

type Refusal ¶

type Refusal string

Refusal says which limit was reached, for the problem document and the metric. A closed set, because it is a label (observability-reliability.md §3.2).

const (
	// RefusedNone means the slot was granted.
	RefusedNone Refusal = ""
	// RefusedCredential is one client holding too many.
	RefusedCredential Refusal = "credential"
	// RefusedTenant is one workspace holding too many.
	RefusedTenant Refusal = "tenant"
	// RefusedProcess is this pod holding too many, whoever they belong to.
	RefusedProcess Refusal = "process"
	// RefusedDraining is the process shutting down.
	RefusedDraining Refusal = "draining"
)

func (r Refusal) String() string

type Registry ¶

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

Registry counts what is open and hands out the right to open one more.

Process-local, and deliberately so. The `api` role is stateless and horizontally scaled, so there is no shared count to keep - a limit per process is a limit per pod, which is exactly the resource being protected. A cluster-wide cap would need coordination on the request path for a number nobody can act on anyway.

func NewRegistry(limits Limits) *Registry

func (r *Registry) Admit(credential, tenant string) (Slot, Refusal)

func (r *Registry) CloseAll()

func (r *Registry) Draining() bool

func (r *Registry) Open() int

type Slot ¶

type Slot struct {
	// Closing is closed when the process wants the stream to end. The handler selects on it beside
	// the client's own context, so a shutdown ends the connection the same way a client leaving
	// does - by returning from the handler, which is what lets the server drain (§9).
	Closing <-chan struct{}
	// contains filtered or unexported fields
}

Slot is the right to hold one stream open.

func (s Slot) Release()

type Writer ¶

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

Writer frames server-sent events onto a response, and flushes each one.

The flush is the whole reason this type exists rather than a few Fprintf calls. net/http buffers a response, so an event that was written and not flushed is an event the client gets when the next four kilobytes arrive - which for a quiet workspace is minutes later, and which makes a stream indistinguishable from a broken one.

Here rather than in either adapter because both write the same wire format: `GET /stream` carries change records to a browser and `GET /mcp` carries JSON-RPC notifications to an agent, and the framing - the field names, the blank line, the flush, the newline stripping - is the same protocol in both. What differs is what goes in the `data`, which is each adapter's own.

func NewWriter(w http.ResponseWriter) *Writer

func (s *Writer) Comment(text string) error

func (s *Writer) Event(id, name, data string) error

func (s *Writer) Retry(milliseconds int) error

func (s *Writer) Write(frame string) error