Documentation
¶
Overview ¶
Package turnqueue serializes the turns of one conversation.
Two paths reach it -- a WebSocket message and an HTTP post -- and a conversation may only be running one turn at a time whichever arrives. The registry is server-scoped rather than connection-scoped for that reason: a second browser tab is a second connection but the same conversation.
Index ¶
Constants ¶
const MaxQueued = agent.DefaultMaxQueuedMessages
MaxQueued caps how many turns may wait behind the one running in a conversation. Past that a submission is refused rather than silently accepted: a user who cannot see the backlog should not be able to grow it without limit.
Variables ¶
var ErrDraining = errors.New("this server is shutting down; retry the turn")
ErrDraining is returned once the server has begun shutting down. A turn is a model call that writes message history, so starting one this process cannot finish is worse than refusing it: the caller retries against an instance that will still be here.
var ErrQueueFull = errors.New("too many turns are already queued for this conversation")
ErrQueueFull is returned by the registry when a conversation is at its cap.
Functions ¶
This section is empty.
Types ¶
type Job ¶
type Job struct {
// OnDequeue, when set, is called just before run for a job that had to wait.
// A job that started immediately never sees it, which is what lets a surface
// announce "this queued message is starting now" without a race.
OnDequeue func()
Done chan struct{}
Dropped atomic.Bool
// contains filtered or unexported fields
}
Job is one conversation turn waiting for its conversation to be free.
type Registry ¶
type Registry struct {
// contains filtered or unexported fields
}
Registry owns the per-conversation turn queues for the whole server.
It lives on the Handler rather than on a WebSocket connection because a conversation outlives any one connection: the same conversation is reachable from a reconnected socket, a second browser tab, the HTTP API, and a system turn reporting a finished task. Serialization anchored to a connection would let two of those interleave their reads and writes of the same message history.
func NewRegistry ¶
func NewRegistry() *Registry
func (*Registry) Drain ¶
func (r *Registry) Drain()
Drain refuses new turns. Turns already running are left alone — Wait is what gives them their moment to finish.
func (*Registry) RunSync ¶
RunSync submits a turn and waits for it to finish. It is what a request that has to stream its own turn back to the caller uses, where a fire-and-forget submit would return before there was anything to send.
A caller that goes away before its turn starts marks the job dropped and returns the context error; the queue moves on to the next turn.
func (*Registry) Submit ¶
Submit hands a turn to the conversation's queue. It returns 0 when the turn started immediately, or the job's 1-based position when it had to wait. Submit never blocks; the turn runs on a goroutine the registry owns.
Source Files