Documentation
¶
Overview ¶
Package workerquery is the Worker protocol's second message kind: a query, answered by a Worker Instance with its Worker's credential and never written anywhere (ADR-0473).
A job is durable work. Creating it is an event, leasing it is an event, completing it is an event, and recovery replays all three — which is right for work a process depends on. A person searching a directory from a task form is not that: there is nothing to recover, and a search term in the log is personal data kept for no reason. So a query lives in this package's memory for as long as the HTTP request waiting on it, and no longer.
What it does share with a job is the boundary that matters. It is leased by the same Worker Instances, under the same fail-closed worker scope (ADR-0194/0297), routed by the same job type and to an Instance that reports holding the Worker it names (ADR-0168), and its answer is fenced by a lease token. The engine still holds no credential, and the processor, the WAL, the state store and recovery never see a query at all.
The broker is neither engine nor design-time state, so it is not behind the run loop (I3 binds process state, and nothing here is). It holds its own mutex, as the server's rate limiters do, and every structure under it is bounded: the queue per job type, the batch a poll takes, and every query's life by its deadline.
Index ¶
Constants ¶
const MaxPollWait = 50 * time.Second
MaxPollWait bounds one long poll, as maxJobPollWait bounds a job poll. It stays below Limits.PresentFor, so an Instance inside a poll is always present.
Variables ¶
var ( // once rather than after the deadline, because the person is watching a spinner. ErrUnavailable = errors.New("workerquery: no worker instance serves this worker") // ErrBusy: the queue for that job type is full. ErrBusy = errors.New("workerquery: too many queries are waiting for this worker") // ErrTimeout: nobody answered before the deadline. ErrTimeout = errors.New("workerquery: no answer before the deadline") )
The answers a waiting asker can get instead of a page.
var ( // ErrGone: the query is no longer waiting — answered, withdrawn, timed out, or // lost with a restart. The answer is dropped. ErrGone = errors.New("workerquery: the query is no longer waiting") // ErrFenced: the answer does not carry the lease this query was handed out under. ErrFenced = errors.New("workerquery: the answer does not hold this query's lease") )
The answers a Worker Instance can get for its answer.
Functions ¶
This section is empty.
Types ¶
type Broker ¶
type Broker struct {
// contains filtered or unexported fields
}
Broker holds the queries waiting for a Worker Instance and the ones leased to one. Build it with NewBroker; the zero value is not usable.
func (*Broker) Ask ¶
func (b *Broker) Ask(ctx context.Context, jobType, connector string, q lookup.Query, timeout time.Duration) (lookup.Page, error)
Ask queues one query for an Instance holding connector and waits for its answer, the timeout, or ctx — whichever comes first. A query nobody can answer is refused at once, and so is one that would overfill its queue.
A Worker's error comes back as the *lookup.Error it reported.
func (*Broker) Lease ¶
func (b *Broker) Lease(ctx context.Context, jobType, worker string, connectors []string, max int, wait time.Duration) []Leased
Lease hands an Instance up to max queries of jobType addressed to one of the Workers it holds, waiting up to wait for one to arrive. Every poll, productive or not, is what keeps the Instance present.
type Leased ¶
type Leased struct {
ID string `json:"id"`
LeaseToken string `json:"leaseToken"`
Type string `json:"type"`
// Connector names the Worker — the tenant — the query is for. The Instance holds
// its credential; the query only names it, as a job does.
Connector string `json:"connector"`
// Deadline is when the asker stops waiting, in Unix milliseconds. An Instance has
// no reason to spend a provider call past it.
Deadline int64 `json:"deadline"`
Query json.RawMessage `json:"query"`
}
Leased is one query as a Worker Instance receives it.
type Limits ¶
type Limits struct {
// QueuePerType is how many queries may wait, unleased, for one job type.
QueuePerType int
// PresentFor is how long a poll counts as an Instance being there. It must exceed
// the longest long poll, or an Instance waiting inside one would read as absent.
PresentFor time.Duration
// MaxBatch is the most queries one poll takes.
MaxBatch int
}
Limits bound the broker.
type Service ¶
type Service struct {
// contains filtered or unexported fields
}
Service serves the worker half of the protocol: the poll and the answer. The asker's half is Broker.Ask, called by whichever area of the server has a query to put.
func NewService ¶
NewService serves broker over HTTP.
func (*Service) HandleActivate ¶
func (s *Service) HandleActivate(w http.ResponseWriter, r *http.Request)
HandleActivate is POST /api/v1/worker-queries/activate: a Worker Instance's poll for queries, the query twin of POST /api/v1/jobs/activate.
Body: {"type": "io.atlas.entra", "worker": "w1", "connectors": ["ucc_entra"], "maxQueries": 4, "waitMs": 30000}. The connectors are the Workers the Instance holds credentials for; only queries naming one of them are handed out, and reporting them is also what makes the Instance present for those Workers.
func (*Service) HandleAnswer ¶
func (s *Service) HandleAnswer(w http.ResponseWriter, r *http.Request)
HandleAnswer is POST /api/v1/worker-queries/{id}/answer: the page a Worker Instance read, or the code it failed with, fenced by the lease it was handed.
Body: {"worker": "w1", "leaseToken": "…", "page": {"items": [...], "next": "…"}} or {"worker": "w1", "leaseToken": "…", "error": {"code": "provider_throttled", "retryAfter": 10}}.
Source Files