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 Lease ¶
type Lease interface {
// Fence is the monotonic token issued when the lease was granted; a later
// grant always carries a higher token.
Fence() int64
// Release drops the lease so another replica may run the conversation's next
// turn.
Release()
}
Lease is a held cross-replica conversation lock.
type Locker ¶
type Locker interface {
// Acquire blocks until this replica holds the conversation's lease or ctx is
// cancelled. The returned Lease is held until Release.
Acquire(ctx context.Context, conversationID string) (Lease, error)
}
Locker serializes a conversation's turns across server replicas. A nil Locker selects the single-instance path, where the in-process queue is the only serialization needed. See docs/design/server-coordination.md §7.
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 ¶
NewRegistry returns a turn registry. Pass a Locker to serialize a conversation's turns across replicas; nil keeps serialization in this process only, which is correct for a single-replica deployment.
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. It also cancels pending lease acquisitions, so a turn blocked waiting for another replica's conversation lease stops waiting rather than starting on a process that is going away.
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