Documentation
¶
Overview ¶
Package jobbroker is the stack's async job broker: it accepts paid requests the x402-verifier has already verified + settled, replays them against the offer's real upstream with no client-facing deadline, and serves free status pages plus gated results. One broker per stack, multi-tenant across offers; jobs are runtime data in SQLite on a PVC — deliberately NOT CRDs (etcd write amplification, 1.5MiB caps on result bodies, TTL churn) and NOT k8s Jobs (the work is an HTTP replay against a running Service, not a batch pod).
Index ¶
- Constants
- func NewID() string
- type Job
- type ListSummary
- type Server
- type Store
- func (s *Store) Close() error
- func (s *Store) Create(j *Job) error
- func (s *Store) Finish(id string, status int, contentType string, body []byte, errMsg string, ...) error
- func (s *Store) Get(id string) (*Job, error)
- func (s *Store) ListForWallet(offer, wallet string, limit int) ([]ListSummary, error)
- func (s *Store) MarkRunning(id string, at time.Time) error
- func (s *Store) SweepExpired(now time.Time) (int64, error)
Constants ¶
const ( HeaderUpstreamURL = "X-Obol-Upstream-Url" HeaderOffer = "X-Obol-Offer" HeaderPayTo = "X-Obol-Pay-To" HeaderJobTTL = "X-Obol-Job-Ttl" HeaderVisibility = "X-Obol-Result-Visibility" HeaderPublicPrefix = "X-Obol-Public-Prefix" HeaderUpstreamAuth = "X-Obol-Upstream-Auth" // Set by the verifier from verified credentials. HeaderPaymentPayer = "X-Payment-Payer" HeaderVerifiedWallet = "X-Verified-Wallet" )
Contract headers between the x402-verifier and the broker. The verifier sets all of them (client-supplied copies never survive the gate), so the broker can stay stateless about offers: everything it needs rides the request.
const ( StatePending = "pending" StateRunning = "running" StateComplete = "complete" StateFailed = "failed" )
Job states.
const HeaderBrokerSig = "X-Obol-Broker-Sig"
HeaderBrokerSig carries the verifier's HMAC over the contract headers, so the broker can reject forged submits even from a pod the NetworkPolicy allows (defense in depth with the F1 NetworkPolicy). Mirrored in internal/x402 (see authgate.go) — the two packages don't share code.
Variables ¶
This section is empty.
Functions ¶
Types ¶
type Job ¶
type Job struct {
ID string
Token string // capability credential returned in the 202 body
Offer string // "<namespace>/<name>"
Payer string // wallet that paid (lowercase 0x…), may be empty
PayTo string // offer's payTo wallet (sellers list their jobs with it)
State string
CreatedAt time.Time
StartedAt time.Time
DoneAt time.Time
ExpiresAt time.Time
Method string
Path string // path relative to the offer root, e.g. /submit
ContentType string
Body []byte
UpstreamURL string // in-cluster base URL to replay against
CallbackURL string // optional completion webhook
Visibility string // payer | public
ResultStatus int
ResultContentType string
Result []byte
Error string
}
Job is one accepted async request and (eventually) its stored result.
type ListSummary ¶
type ListSummary struct {
ID string `json:"jobId"`
State string `json:"state"`
Path string `json:"path"`
CreatedAt time.Time `json:"createdAt"`
}
ListSummary is the /jobs listing row (no bodies).
type Server ¶
type Server struct {
// contains filtered or unexported fields
}
Server is the broker's HTTP surface.
func (*Server) Handler ¶
Handler returns the broker mux. The verifier strips the offer prefix before proxying, so paths arrive offer-relative:
/jobs → list (wallet-scoped) /jobs/<id> → status (free; JSON or HTML by Accept) /jobs/<id>/result → stored result (visibility-gated) anything else → submit (a paid request the verifier settled)
func (*Server) RunSweeper ¶
RunSweeper deletes expired jobs every interval until stop is closed.
type Store ¶
type Store struct {
// contains filtered or unexported fields
}
Store persists jobs. All methods are safe for concurrent use (database/sql pools; SQLite serializes writers).
func OpenStore ¶
OpenStore opens (creating if needed) the job database at path. Use ":memory:" for tests.
func (*Store) Finish ¶
func (s *Store) Finish(id string, status int, contentType string, body []byte, errMsg string, at time.Time) error
Finish records the upstream outcome. status<400 → complete, else failed (money already settled at acceptance — the status page says so plainly).
func (*Store) ListForWallet ¶
func (s *Store) ListForWallet(offer, wallet string, limit int) ([]ListSummary, error)
ListForWallet returns jobs the wallet may see under one offer: all of the offer's jobs when the wallet is the offer's payTo (the seller), else only jobs the wallet paid for.
Source Files