Documentation
¶
Overview ¶
Package worker provides an asynchronous worker pool and utils for persisting conversation turns using the provided storage.Driver.
The pool decouples storage operations from the proxy's HTTP hot path so that the client-proxy-upstream interaction is fully transparent.
Embedding writes deliberately do NOT happen here: the derive worker family is the single writer of embeddings (pkg/spanembed), keyed by deterministic span identity, so the ingest hot path stays pure capture.
Index ¶
Constants ¶
This section is empty.
Variables ¶
This section is empty.
Functions ¶
This section is empty.
Types ¶
type Config ¶
type Config struct {
// Driver is the storage backend: the raw-turn layer plus the
// sessions surface (Postgres). Drivers without those capabilities
// (the in-memory test driver) make capture a no-op.
Driver storage.Driver
// NumWorkers is the number of background workers in the pool.
NumWorkers uint
// QueueSize is the capacity of the buffered job channel (defaults to 256).
QueueSize uint
// QueueByteBudget bounds the total retained bytes of queued-and-in-flight
// jobs. The count cap bounds slots; this bounds memory. Zero-value defaults.
QueueByteBudget int64
// Project is the git repository or project name to tag on stored nodes.
Project string
// Logger is the provided logger
Logger *slog.Logger
}
Config is the configuration options for the worker pool.
type Job ¶
type Job struct {
Provider string
AgentName string
// ThreadID is the harness sub-thread that fired this call ("" for
// the main thread), captured at the wire and stamped onto the
// turn's nodes as non-hashed metadata.
ThreadID string
// RequestID is Paper's canonical identifier for the captured HTTP attempt.
// UpstreamRequestID is the provider's independent identifier for its side
// of that attempt; it must never replace RequestID when the latter is empty.
RequestID string
UpstreamRequestID string
Req *llm.ChatRequest
Resp *llm.ChatResponse
// Weight is the estimated bytes this job retains live (its parsed body)
// while queued and processed; the pool holds it against the byte budget
// from admission until the job completes. Zero leaves the job unweighted,
// so only the slot-count cap applies.
Weight int
// RawRequest is the verbatim provider request body the proxy
// received, persisted unparsed into the immutable raw-turn layer so
// the deriver re-parses it (and fields unknown to this build
// survive). Empty for callers that don't capture into raw_turns
// (e.g. the in-memory test driver); the raw write is skipped then.
RawRequest json.RawMessage
// Session is the optional session-tracking envelope attached to
// the turn. When non-nil and the driver supports session-aware
// ingest (Postgres), the worker UPSERTs the turn's `sessions` row
// and folds its derived_status so the deriver can resolve the
// session and attach its spans. When nil OR when the driver does
// not implement that capability (e.g. inmemory), no sessions row is
// written — this keeps unit tests working without a Postgres
// backend. The local proxy always attaches an envelope so its
// captured turns surface in the deck.
Session *sessions.IngestEnvelope
}
Job is a unit of work for the worker pool to execute against.
type Pool ¶
type Pool struct {
// contains filtered or unexported fields
}
Pool processes storage jobs asynchronously via a worker pool.
func (*Pool) Admit ¶ added in v0.36.0
func (p *Pool) Admit(job Job) RejectReason
Admit weighs a job against the retained-byte budget and the slot-count cap and, when both admit it, hands it to the workers. The reserved weight is held until the job finishes processing (released in worker). RejectNone means the job was enqueued; the other reasons name the ceiling that declined it.
func (*Pool) Close ¶
func (p *Pool) Close()
Close signals workers to stop and waits for in-flight jobs to drain. Call this during graceful shutdown after the proxy HTTP server has stopped.
func (*Pool) Enqueue ¶
Enqueue submits a job for processing by the worker pool. Returns true if enqueued, false if rejected by either the byte budget or the slot-count cap, resulting in the job being dropped. Call Admit to learn which.
type RejectReason ¶ added in v0.36.0
type RejectReason int
RejectReason names the ceiling that declined a job so the ingest saturation signal can attribute a drop to the budget that bit. RejectNone means the job was enqueued.
const ( RejectNone RejectReason = iota RejectQueueFull RejectByteBudget )
Source Files