Documentation
¶
Overview ¶
Package dispatch receives newline-delimited envelope JSON over HTTP, runs the full receive ladder per line, and answers with newline-delimited ack JSON. Send posts messages and collects the replies. See docs/plans/dispatch.md for the contract.
The ladder runs, per line, in fixed order and fails fast: Decode, VerifySignature, Room.Accepts, resolve, handle, then NewAck, Confirm, and Encode build the reply. agent.EmitMessageDelivered and agent.EmitMessageAcked are best-effort diagnostics outside this ladder, called after their point in the sequence with their error return ignored; they never fail a line.
MessageDeliveredEvent fires after VerifySignature and before Room.Accepts. It means "signature verified," not "room-admitted": a delivered event can still precede an admission rejection.
Index ¶
Constants ¶
const DefaultMaxBodyBytes int64 = 1 << 20
DefaultMaxBodyBytes caps one NDJSON request body when Options.MaxBodyBytes is zero. It bounds the memory one request can commit before any line runs the receive ladder.
const DefaultReplayCapacity = 100000
DefaultReplayCapacity caps the entry count of the ledger New builds internally when Options.Ledger is nil. A caller-supplied Ledger is not bounded by this constant; the caller owns its Store's capacity. The cap makes replay protection a bounded window: an evicted key is processed again if it arrives later. It bounds the records that hold no live claim, not the records that do.
const DefaultReplayLease = 30 * time.Second
DefaultReplayLease bounds one line's replay claim when Options.ReplayLease is zero. This is not a crash-detection timeout: size it above Handler.Handle's expected p99 latency, or a slow handler's own replay can re-run work before its first claim completes. See the Scope section warning in docs/plans/dispatch.md.
Variables ¶
var ( ErrNoID = errors.New("dispatch: endpoint id is required") ErrNoRoom = errors.New("dispatch: room is required") ErrNoResolve = errors.New("dispatch: resolve func is required") ErrBadMethod = errors.New("dispatch: POST required") ErrBadRequest = errors.New("dispatch: request body read failed") ErrBadMaxBody = errors.New("dispatch: max body bytes must not be negative") // ErrBadReplayLease reports a negative Options.ReplayLease, a // ReplayLease under one second, or a negative // Options.ReplayCapacity. Options.Validate returns this sentinel // for any of the three. ErrBadReplayLease = errors.New("dispatch: replay lease and capacity must not be negative") // ErrReplay reports a message the ledger already admitted: a // completed, failed, or blocked key, or a key still claimed by an // in-flight duplicate. Endpoint.Handler answers this with a // "replay:" error line instead of running resolve or handle again. ErrReplay = errors.New("dispatch: message already processed") )
Sentinel errors for New and Endpoint.Handler; test with errors.Is.
Functions ¶
This section is empty.
Types ¶
type Endpoint ¶
type Endpoint struct {
// contains filtered or unexported fields
}
Endpoint receives NDJSON envelope messages and answers with NDJSON acks. Build one with New.
func New ¶
New validates opts, builds a Bus when opts.Bus is nil, and returns the wired Endpoint. Emit accepts an event name with no subscriber, so the endpoint needs no fixture subscriptions on the bus.
func (*Endpoint) Handler ¶
Handler serves POST requests with NDJSON bodies. A non-POST method answers 405 with ErrBadMethod. An unreadable body answers 400 with ErrBadRequest, as does a body past Options.MaxBodyBytes. Each request line runs the receive ladder and contributes one reply line, an ack or a JSON error object; the stream stays open across a per-line failure.
type Handler ¶
Handler resolves one received message into a restatement. New's caller supplies a Resolve func that looks one up per message. Implementations receive an already-verified, already-admitted message.
type Options ¶
type Options struct {
// ID is this endpoint's identity; it becomes each ack's From.
ID string
// Room gates admission: only a signed message from a member,
// addressed only to members, and naming this room, is admitted.
Room *room.Room
// Resolve looks up the Handler that owns an admitted message.
Resolve func(ctx context.Context, m envelope.Message) (Handler, error)
// Bus receives MessageDeliveredEvent and MessageAckedEvent. Built
// when nil; no handler is subscribed. Callers add handlers through
// Bus().Subscribe.
Bus *events.Bus
// MaxBodyBytes caps one request body. Zero resolves to
// DefaultMaxBodyBytes; a negative value fails Validate. A body
// past the cap answers 400 with ErrBadRequest.
MaxBodyBytes int64
// Ledger provides replay protection over the receive ladder. Built
// as a bounded in-memory ledger, sharing Bus for its events, when
// nil.
Ledger *ledger.Ledger
// ReplayLease bounds one line's replay claim. Zero resolves to
// DefaultReplayLease. A negative value, or a value under one
// second, fails Validate. Size this above Handler.Handle's
// expected p99 latency; see the warning in the Scope section of
// docs/plans/dispatch.md.
ReplayLease time.Duration
// ReplayCapacity caps the entry count of the ledger New builds
// internally when Ledger is nil. Zero resolves to
// DefaultReplayCapacity. A negative value fails Validate. Ignored
// when Ledger is set; the caller-supplied Ledger owns its own
// Store's capacity.
//
// Replay protection is a bounded window, not a permanent
// guarantee: a key evicted under this cap is processed again if it
// arrives later, and the endpoint answers a fresh ack. The cap
// bounds the records that hold no live claim. A record claimed by
// an in-flight Handle call is never evicted, so a hard memory
// bound needs a request-rate limit in front of the endpoint.
ReplayCapacity int
}
Options configures New. ID, Room, and Resolve are required.
func (Options) Validate ¶
Validate checks ID, Room, Resolve, MaxBodyBytes, ReplayLease, and ReplayCapacity, in that order, and returns the first sentinel that fails. A nonzero ReplayLease under one second fails Validate: this is a sanity floor against a unit-confusion bug, not a floor tied to any handler's real latency.
type SendResult ¶
SendResult is one reply line's outcome, in request order.
Source Files