Documentation
¶
Overview ¶
Package batch is the batch subsystem: it accepts bulk inference submissions, runs each item as a background job (via the jobq module), and exposes submit/poll/cancel/results over the customer-facing /v1 surface.
Layering: batch is a CONSUMER of jobq. jobq owns durable execution (claim, retry, crash recovery) and the per-item payload bytes; this package owns the batch concept — the batch record, the batch→job mapping, the inference handler that turns one item into an upstream call (reusing the realtime pipeline), and the HTTP API. The relay hot path is untouched; admission only validates and enqueues, execution happens off to the side.
Index ¶
- Constants
- Variables
- type Batch
- type BatchView
- type Item
- type ItemResult
- type ItemView
- type Runner
- type Service
- func (s *Service) Cancel(ctx context.Context, id, relayKeyHash string) error
- func (s *Service) Handler() jobq.Handler
- func (s *Service) Results(ctx context.Context, id, relayKeyHash string) ([]ItemResult, error)
- func (s *Service) Routes() http.Handler
- func (s *Service) Status(ctx context.Context, id, relayKeyHash string) (*BatchView, error)
- func (s *Service) Submit(ctx context.Context, relayKeyHash, policyID, inbound string, items [][]byte) (string, error)
- type Status
- type Store
- func (s *Store) AddItem(ctx context.Context, batchID string, idx int, jobID string) error
- func (s *Store) Create(ctx context.Context, b *Batch) error
- func (s *Store) Get(ctx context.Context, id string) (*Batch, error)
- func (s *Store) Items(ctx context.Context, batchID string) ([]Item, error)
- func (s *Store) ListByRelayKey(ctx context.Context, relayKeyHash string) ([]*Batch, error)
- func (s *Store) SetCompleted(ctx context.Context, id string, st Status) error
- func (s *Store) SetStatus(ctx context.Context, id string, st Status) error
Constants ¶
const Queue = "inference"
Queue is the jobq queue name batch items run on. The Service registers its handler under this name; the worker pool drains it.
Variables ¶
var ErrCrossShape = errors.New("batch: cross-shape dispatch not yet supported")
ErrCrossShape is returned when a batch item's inbound shape differs from the resolved upstream adapter. v1 batch supports same-shape / byte-pass / native only; cross-shape translation is a follow-up. Surfaced explicitly rather than silently mis-dispatched (canonical rule 11).
var ErrForbidden = errors.New("batch: not owner")
ErrForbidden is returned when a caller asks about a batch they don't own.
var ErrNotFound = errors.New("batch: not found")
ErrNotFound is returned when a batch id is unknown.
Functions ¶
This section is empty.
Types ¶
type Batch ¶
type Batch struct {
ID string
RelayKeyHash string // owner — used for authz on read/cancel
PolicyID string
InboundShape string // the wire shape items are expressed in (adapter spec name)
Status Status
TotalItems int
CreatedAt time.Time
CompletedAt *time.Time
}
Batch is the durable record of one bulk submission.
type BatchView ¶
type BatchView struct {
*Batch
Counts map[jobq.State]int `json:"counts"`
Items []ItemView `json:"items"`
}
BatchView is a batch plus its items' live states and a per-state roll-up.
type ItemResult ¶
type ItemResult struct {
Idx int `json:"idx"`
State jobq.State `json:"state"`
Response json.RawMessage `json:"response,omitempty"`
Error string `json:"error,omitempty"`
}
ItemResult is one item's terminal outcome for the results endpoint.
type Runner ¶
type Runner struct {
Resolver *routing.Resolver
Pipeline *pipeline.Pipeline
Specs *adapter.Registry
Catalog *appcatalog.Catalog
}
Runner executes one batch item through the realtime inference pipeline, buffering the response. It deliberately reuses app/pipeline.Pipeline.Run so a batch item inherits keypool selection, circuit breakers, failover, and usage emission for free — the only difference from a live request is that the lifecycle source is "batch" and the response is buffered rather than streamed.
func (*Runner) Run ¶
func (rn *Runner) Run(ctx context.Context, requestID, relayKeyHash string, inbound adapters.Name, body []byte) (int, []byte, error)
Run executes one item. requestID ties the usage event to the item (the jobq job id); relayKeyHash + inbound select routing. It returns the upstream status and the buffered response body. Usage emits automatically with source="batch" when the pipeline body closes.
type Service ¶
type Service struct {
// contains filtered or unexported fields
}
Service is the batch subsystem's application layer: it accepts submissions, enqueues each item as a jobq job, and answers status/results/cancel. It is a pure consumer of jobq (execution) + Store (the batch record).
func NewService ¶
NewService wires the store, queue, and runner together.
func (*Service) Cancel ¶
Cancel cancels every not-yet-terminal item and marks the batch cancelled. relayKeyHash must match the batch owner.
func (*Service) Handler ¶
Handler is the jobq Handler that runs one batch item. Registered on the queue at boot. Pure input→output: it runs the item through the realtime pipeline (reusing keypool/breakers/usage) and returns the response bytes, which jobq persists as the item's result. A non-2xx upstream status is a failure so jobq records it (the body is not retained in v1 — follow-up).
func (*Service) Results ¶
Results returns each item's outcome: the response body for completed items, the error for failed ones. relayKeyHash must match the batch owner.
func (*Service) Routes ¶
Routes returns the /v1/batches HTTP surface. Mount it on the inference router behind the readiness → classify → relay-key-auth chain so every handler sees an authenticated relay key via inference.RelayKeyFromContext.
POST /v1/batches submit (JSON: {shape, requests:[...]})
GET /v1/batches/{id} status + per-item states
GET /v1/batches/{id}/results per-item outcomes
POST /v1/batches/{id}/cancel cancel pending items
type Status ¶
type Status string
Status is the coarse, cached lifecycle of a batch. The authoritative per-item state lives in jobq; this is a cheap roll-up plus the terminal cancellation marker.
const ( StatusQueued Status = "queued" // accepted, items enqueued StatusRunning Status = "running" // at least one item has started StatusCompleted Status = "completed" // all items finished successfully StatusFailed Status = "failed" // all items finished, some failed StatusCancelled Status = "cancelled" // caller cancelled )
type Store ¶
type Store struct {
// contains filtered or unexported fields
}
Store persists batch records and the batch→job mapping via sqlc-generated queries. Per-item execution state is NOT stored here — it is read from jobq by job id.
func (*Store) ListByRelayKey ¶
ListByRelayKey returns a caller's batches, newest first.
func (*Store) SetCompleted ¶
SetCompleted stamps the terminal status and completed_at.
Source Files