job

type CountOpts struct {
	// Queue filters by queue name. Empty means all queues.
	Queue string
	// State filters by job state. Empty means all states.
	State State
}

type Definition[T any] struct {
	// Name is the unique identifier for this job type.
	Name string

	// Handler is the function that processes the job payload.
	Handler func(ctx context.Context, payload T) error

	// Opts configures retries, queue, priority, and timeout.
	Opts Options
}

type DequeueOpts struct {
	// Queues restricts the claim to these queue names.
	//
	// An EMPTY list is not a portable request and callers should not
	// send one. The backends genuinely disagree, and the disagreement
	// predates this option:
	//
	//	memory    claims from every queue
	//	postgres  claims nothing (queue = ANY(NULL) matches no row)
	//	sqlite    claims nothing, by an explicit early return before the
	//	          query is built
	//	redis     claims nothing — the index is one sorted set per queue
	//	          name and there has never been a cross-queue index
	//	mongo     claims nothing, by an explicit early return
	//
	// Mongo's guard is the one that had to be added: {queue: {$in: nil}}
	// marshals to BSON null and the server rejects the whole query with
	// "$in needs an array", so an empty list was a hard error rather
	// than any behaviour at all. It returns empty to match the three
	// backends a real deployment would be running, rather than inventing
	// an all-queues scan that no other persistent backend performs.
	//
	// The conformance suite does not exercise an empty list and no
	// caller in this repository sends one. Unifying the five would be a
	// behaviour change to store/memory and is deliberately not made
	// here; documenting the split is what stops a caller assuming it.
	Queues []string

	// Limit is the maximum number of jobs to claim. It counts eligible
	// jobs only: a job excluded by Budget or CustomKeys must not consume
	// a slot, or one oversized job at the head of the queue would starve
	// a worker that had capacity for everything behind it.
	//
	// Limit <= 0 claims nothing and returns (nil, nil); this is unified
	// across all five backends, checked before Validate.
	Limit int

	// Budget is the free capacity the caller is offering, in canonical
	// units (cpu millicores, memory and disk bytes, gpu milli-devices).
	// A job is eligible on a dimension when its requirement is <= the
	// budget for that dimension.
	//
	// An ABSENT key is unconstrained, not zero. This inverts
	// resource.Set.Fits, which treats absent capacity as zero, and the
	// inversion is deliberate: a worker that declares only memory must
	// still claim GPU-requiring jobs, because otherwise adding a
	// dimension to one worker's config would silently strand work on
	// every worker that had not been updated yet.
	//
	// A key present with the value zero is a real constraint — a worker
	// with no free memory — and excludes any job requiring more than
	// zero of it. That is why IsUnbounded tests key presence rather than
	// resource.Set.IsZero.
	//
	// Custom keys in Budget are ignored by the predicate. Quantity
	// matching on a custom dimension would need a document comparison or
	// a join table in five backends to serve a rare case; offer custom
	// keys through CustomKeys instead.
	Budget resource.Set

	// CustomKeys are the custom resource keys the caller offers,
	// typically free.CustomKeys(). A job is eligible only if every custom
	// key it requires appears here.
	//
	// An empty list is read against IsUnbounded rather than on its own,
	// because "no custom keys" means two different things to two
	// different callers. If the whole of o is unbounded, the caller does
	// not use the resource model at all and claims everything, custom
	// keys included — that is the backward-compatibility guarantee. If o
	// is bounded in any other way, an empty list means this worker
	// genuinely has no custom resources, and a job requiring an fpga must
	// not be handed to it.
	//
	// Reading an empty list as unconstrained in the bounded case would
	// let a resource-aware worker claim work it cannot possibly run;
	// reading it as "offers nothing" in the unbounded case would strand
	// every custom-key job in the fleet the day this option shipped.
	//
	// Only key containment is tested at dequeue; the quantity is enforced
	// locally after the claim, by the admission path that already owns
	// the accounting. Matching a quantity here would need a document
	// comparison or a join table in five backends to serve a rare case.
	//
	// Backends match against the delimited string
	// resource.EncodeCustomKeys produced at enqueue, whose leading and
	// trailing separators are what stop ",fpga," matching a worker that
	// only offers ",fpga-large,". The test is subset, not substring: a
	// job needing {fpga, tpu} is eligible for a caller offering
	// {fpga, nvme, tpu}, even though the offered list interleaves a key
	// the job does not want.
	CustomKeys []string

	// PreferHashes are PrimaryInputHash values the caller already has
	// staged locally. A job whose PrimaryInputHash appears here sorts
	// ahead of jobs at the same priority, saving a re-download.
	//
	// Backends must bind PreferredHashes rather than this field: an
	// empty string here is not a locality signal and must not become one.
	//
	// This is advisory and must NEVER filter, and must never outrank
	// priority: locality that could reorder across priority bands would
	// let a steady stream of locally cached work starve the high-priority
	// job the pool exists to run first. The full ordering is priority
	// descending, then preferred before unpreferred, then RunAt
	// ascending.
	//
	// It is deliberately NOT a term of IsUnbounded. If it were, opts
	// carrying only PreferHashes would count as bounded, and the empty
	// CustomKeys rule would then reject every custom-resource job — this
	// field would filter transitively, contradicting the paragraph above.
	PreferHashes []string

	// ReservedFor restricts the claim to a single job. When set, no other
	// job may be returned, and that job is still subject to every other
	// constraint here — a targeted claim that could bypass Budget would
	// reintroduce exactly the overcommit this predicate prevents.
	ReservedFor *id.JobID

	// WorkerID is the worker taking the lease. It is required when
	// LeaseUntil is set and ignored otherwise.
	WorkerID id.WorkerID

	// LeaseUntil, when non-zero, makes the claim grant a lease: the
	// claimed rows get WorkerID, this expiry, and an incremented
	// lease_epoch, in the same statement that claims them.
	//
	// A zero value grants no lease and leaves every lease column
	// untouched, which is exactly how DequeueJobs behaved before leases
	// existed. That is the backward-compatibility guarantee: a caller
	// that does not opt in cannot be affected by this.
	//
	// The grant must be part of the claim, not a second write, and the
	// reason is the opposite of the obvious one. Reclamation cannot
	// rescue a half-granted job on the timing that leases exist to
	// provide: Lease.IsExpired reports false for a zero ExpiresAt
	// precisely so the reclaim loop never steals a job that was never
	// leased. A crash between a claim and a separate grant would leave a
	// row running with no expiry, and the fine-grained lease machinery
	// cannot see it at all.
	//
	// Such a row is recovered eventually, because reclamation adopts an
	// unleased running job once it has been silent for
	// UnleasedReclaimGrace, but that clause is a coarse compatibility
	// fallback measured in minutes, not the per-job TTL a lease buys. A
	// half-granted job would wait it out, which is precisely the coarse
	// stale-job behaviour leases exist to replace.
	//
	// One write means a claimed job always carries a lease something can
	// act on promptly.
	LeaseUntil time.Time
}

type HandlerFunc func(ctx context.Context, payload []byte) error

type Job struct {
	dispatch.Entity

	ID          id.JobID      `json:"id"`
	Name        string        `json:"name"`
	Queue       string        `json:"queue"`
	Payload     []byte        `json:"payload"`
	State       State         `json:"state"`
	Priority    int           `json:"priority"`
	MaxRetries  int           `json:"max_retries"`
	RetryCount  int           `json:"retry_count"`
	LastError   string        `json:"last_error,omitempty"`
	ScopeAppID  string        `json:"scope_app_id,omitempty"`
	ScopeOrgID  string        `json:"scope_org_id,omitempty"`
	WorkerID    id.WorkerID   `json:"worker_id,omitempty"`
	RunAt       time.Time     `json:"run_at"`
	StartedAt   *time.Time    `json:"started_at,omitempty"`
	CompletedAt *time.Time    `json:"completed_at,omitempty"`
	HeartbeatAt *time.Time    `json:"heartbeat_at,omitempty"`
	Timeout     time.Duration `json:"timeout,omitempty"`

	// ArtifactBindings carries the encoded map of declared input names to
	// artifact refs. It travels with the job because Payload is opaque to
	// the engine: bindings placed inside it would be invisible to the
	// scheduler and to the staging middleware.
	ArtifactBindings []byte `json:"artifact_bindings,omitempty"`

	// Resources is the resolved requirement, computed once at enqueue.
	// Scheduling reads this rather than calling user code.
	Resources resource.Set `json:"resources,omitempty"`

	// ResourceLimits is the resolved enforcement ceiling.
	ResourceLimits resource.Set `json:"resource_limits,omitempty"`

	// ResourceClass is forwarded to the isolation backend uninterpreted.
	ResourceClass string `json:"resource_class,omitempty"`

	// InputBytes is the total size of the declared artifact inputs. It
	// is the estimator's primary feature and the measurement bucket key.
	InputBytes int64 `json:"input_bytes,omitempty"`

	// PrimaryInputHash is the content hash of the largest declared
	// input, used as the locality-scheduling signal. Often empty: the
	// artifact plane fills content_hash at first staging, not at
	// registration, so locality helps from an artifact's second use on.
	PrimaryInputHash string `json:"primary_input_hash,omitempty"`

	// LeaseEpoch is the fencing token for the current lease. It increments
	// on every grant and every reclamation. RenewLease, the grant inside
	// DequeueJobs, ReclaimExpiredLeases, and UpdateLeasedJob all check
	// it, so a worker holding a stale epoch fails to renew — and the pool
	// cancels the job within one heartbeat interval — or has its terminal
	// write refused outright. UpdateJob does not check it: that is a
	// whole-row write with no epoch predicate, so a caller that wants the
	// fence must use UpdateLeasedJob instead.
	//
	// Without the renewal check, a worker resuming from a long GC pause
	// would carry on renewing a lease on a job another worker now owns.
	LeaseEpoch int `json:"lease_epoch"`

	// LeaseExpiresAt is when the current lease lapses if not renewed.
	// Nil means no lease is held.
	LeaseExpiresAt *time.Time `json:"lease_expires_at,omitempty"`

	// LeaseTTL is how long each renewal extends the lease for this job,
	// copied at enqueue from the definition's WithLeaseTTL, or from the
	// enqueue site's when that supplies one. Zero means the pool's
	// default.
	//
	// This is what makes per-definition thresholds work: a 30-second job
	// and a six-hour job carry different values on their own rows, so one
	// reclaim query serves both.
	LeaseTTL time.Duration `json:"lease_ttl,omitempty"`

	// EvictCount is how many times this job has lost a worker to
	// infrastructure — a reclaimed lease, or later a graceful drain. It is
	// deliberately separate from RetryCount: a preempted job has not
	// failed, and charging preemptions to the retry budget would send a
	// healthy job to the DLQ having never once errored.
	EvictCount int `json:"evict_count"`
}

type Lease struct {
	// ExpiresAt is when the lease lapses if not renewed. A zero value
	// means no lease has been granted.
	ExpiresAt time.Time
}

type LeaseStore interface {
	// RenewLease extends the lease to leaseUntil, but only if the job is
	// still running, still assigned to workerID, and still at epoch.
	//
	// It returns ErrLeaseLost when that condition does not hold. That
	// return is the entire fencing mechanism: a worker that was reclaimed
	// while paused learns it no longer owns the job within one heartbeat
	// interval, instead of continuing to write for hours.
	RenewLease(
		ctx context.Context,
		jobID id.JobID,
		workerID id.WorkerID,
		epoch int,
		leaseUntil time.Time,
	) error

	// ReclaimExpiredLeases returns to pending every running job whose
	// lease has expired, clearing the worker assignment, incrementing
	// lease_epoch to fence the previous holder, and incrementing
	// evict_count. RetryCount is never touched — a lost lease is
	// infrastructure, not a handler failure.
	//
	// It also adopts a running job carrying no lease at all, but only
	// once that job has been silent for UnleasedReclaimGrace. See that
	// constant for why the exception exists and why it is gated on
	// silence rather than on the missing expiry alone.
	//
	// The claim and the read are one atomic statement, so two pools
	// reclaiming concurrently cannot both take the same job.
	//
	// A non-positive limit claims nothing and returns (nil, nil), checked
	// before any query runs. This matches DequeueOpts.Limit, so the two
	// methods on this interface agree.
	//
	// ORDER IS NOT PART OF THE CONTRACT, and the backends genuinely
	// differ. Postgres, SQLite and Mongo take the longest-expired jobs
	// first, because each already reads through an index on the expiry
	// and ordering it costs nothing. Memory iterates a map and Redis
	// walks an unordered set member list, so both return an arbitrary
	// subset.
	//
	// This only becomes visible when limit is smaller than the number of
	// jobs eligible at that instant, where the ordered backends drain
	// oldest-first and the other two drain arbitrarily. Nothing starves
	// either way, since each pass moves what it takes out of the eligible
	// set and the reaper runs on a timer.
	//
	// Redis is the reason this is documented rather than unified. It can
	// stop scanning as soon as it has claimed `limit` jobs; ordering would
	// force every call to read the entire job-id set first, turning a
	// bounded scan into a full one on the largest deployments, which is a
	// steep price for an ordering no caller has asked for.
	ReclaimExpiredLeases(ctx context.Context, limit int) ([]*Job, error)

	// UpdateLeasedJob persists j only while the caller still holds the
	// lease.
	//
	// The write applies iff the row is still running, still assigned to
	// workerID, and still at epoch. Otherwise it returns ErrLeaseLost and
	// leaves the row untouched — the lease has moved on, and a worker
	// that no longer owns a job has no coherent claim to make about it.
	//
	// It writes the same columns as UpdateJob EXCEPT the lease-owned
	// ones: lease_epoch, lease_expires_at, worker_id, and heartbeat_at
	// are never written here. Those have exactly three writers — the
	// grant in DequeueJobs, RenewLease, and ReclaimExpiredLeases — and
	// this method deliberately is not a fourth.
	//
	// j is the caller's claim-time (or otherwise stale) snapshot, so
	// j.LeaseExpiresAt is whatever the expiry was when this worker last
	// read the row — every renewal since has pushed the real value
	// forward. A whole-row write that passed the epoch predicate would
	// still roll the expiry backwards, shortening the current holder's
	// lease by however long this caller ran. Excluding the lease-owned
	// columns is what makes the epoch predicate a genuine fence rather
	// than a check that only looks like one.
	//
	// A missing row returns dispatch.ErrJobNotFound rather than
	// ErrLeaseLost: zero rows affected by the fence predicate means "the
	// lease moved on", but a row that no longer exists was never a
	// question of who holds it.
	UpdateLeasedJob(ctx context.Context, j *Job, workerID id.WorkerID, epoch int) error
}

type ListJobsOpts struct {
	States     []State
	Queue      string
	NamePrefix string
	ScopeAppID string
	ScopeOrgID string

	// Cursor is the ID of the last job on the previous page. Empty is the
	// first page. See package paging for the ordering contract.
	Cursor string

	// Limit is the page size; zero or less means paging.DefaultLimit.
	Limit int
}

type ListOpts struct {
	// Limit is the maximum number of jobs to return. Zero means no limit.
	Limit int
	// Offset is the number of jobs to skip.
	Offset int
	// Queue filters by queue name. Empty means all queues.
	Queue string
}

type Lister interface {
	ListJobs(ctx context.Context, opts ListJobsOpts) (Page, error)
}

type Option func(*Options)

type Options struct {
	// MaxRetries is the maximum number of retry attempts before sending to DLQ.
	MaxRetries int

	// Queue is the queue name this job should be enqueued to.
	Queue string

	// Priority determines dequeue ordering. Higher values are processed first.
	Priority int

	// Timeout is the maximum duration a job may run before being cancelled.
	Timeout time.Duration

	// LeaseTTL is how long this job's lease survives without renewal.
	// Zero means the worker pool's default. Set it above the expected gap
	// between heartbeats, not above the expected runtime — a lease is a
	// liveness window, not a time limit.
	LeaseTTL time.Duration

	// RunAt schedules the job for future execution. Zero means immediate.
	RunAt time.Time

	// Inputs declares the artifacts this job consumes. Declaring them,
	// rather than burying refs in the opaque payload, is what lets the
	// engine size the job before scheduling it, validate bindings at
	// enqueue, and stage the bytes before the handler runs.
	Inputs []artifact.InputSpec

	// Bindings maps declared input names to the artifacts supplied at
	// enqueue. The engine validates them against Inputs before the job is
	// persisted.
	Bindings map[string]artifact.Ref

	// Resources declares what this job needs. It is the floor: the
	// engine may raise it via ResourceFunc or a configured estimator,
	// and a per-enqueue WithResources call overrides both.
	Resources resource.Set

	// ResourceLimits is the enforcement ceiling. When unset, memory and
	// the other incompressible keys default to their request and CPU is
	// left unbounded.
	ResourceLimits resource.Set

	// ResourceFunc computes the requirement from the enqueue-time
	// request, for jobs whose footprint scales with their input. It runs
	// once, in the enqueuing process, and never on the scheduling path.
	ResourceFunc resource.ResourceFunc

	// ResourceClass is an opaque scheduling class the isolation backend
	// interprets. Core never reads it.
	ResourceClass string

	// Execution declares the minimum isolation this job's handler
	// requires. The zero value runs in-process, which is what every
	// existing definition gets.
	Execution exec.Policy
}

type Page struct {
	Jobs []*Job

	// NextCursor is the cursor for the following page, or empty when there
	// is none. It is always set when Complete is false.
	NextCursor string

	// Complete is false when a backend stopped scanning at its budget
	// before it could fill the page or prove there was nothing left. The
	// page is then not evidence that no more rows match: continue from
	// NextCursor.
	//
	// Only a backend that stops at a scan budget (redis) ever reports
	// false. The SQL and document backends always finish the search and
	// report true, and that is not a promise that the call is cheap: a
	// selective NamePrefix uses no index on postgres (starts_with), sqlite
	// (substr) or mongo (anchored regex), so a rare or prefix-only filter
	// can walk the whole table before it finds limit+1 matches or reaches
	// the end. Bound list calls with a context deadline.
	Complete bool
}

type Registrable interface {
	// Register adds this definition's handler to the registry.
	Register(r *Registry)

	// JobName returns the name the definition registers under.
	JobName() string

	// Policy returns the execution declaration, so a caller can check
	// that the deployment can satisfy it before registering anything.
	Policy() exec.Policy
}

type Registry struct {
	// contains filtered or unexported fields
}

type ResourceDecl struct {
	// Requests is the declared requirement. It is a floor: the engine may
	// raise it, and a per-enqueue override replaces it per key.
	Requests resource.Set
	// Limits is the declared enforcement ceiling, if any.
	Limits resource.Set
	// Func computes the requirement from the enqueue-time request.
	Func resource.ResourceFunc
	// Class is the opaque scheduling class for the isolation backend.
	Class string
}

type State string

type Store interface {
	// EnqueueJob persists a new job in pending state.
	EnqueueJob(ctx context.Context, j *Job) error

	// DequeueJobs atomically claims up to opts.Limit ready jobs from
	// opts.Queues that fit opts, sets them to running, and returns them
	// ordered by priority descending, then locality-preferred first, then
	// RunAt ascending.
	//
	// The fit test is part of the claim, not a filter over claimed rows.
	// A job that does not fit stays pending and untouched, available to
	// the next worker that does have room for it.
	//
	// When opts.Grants() the same statement also grants a lease: the
	// claimed rows get opts.WorkerID, opts.LeaseUntil, and an incremented
	// lease_epoch, and the returned jobs carry the epoch they were
	// granted. The grant travels in the claiming write itself, never as a
	// follow-up: a row left running with no expiry is outside the lease
	// machinery entirely, so a crash between two writes would leave the
	// job waiting out the far coarser UnleasedReclaimGrace instead of the
	// TTL it was given. See DequeueOpts.LeaseUntil.
	//
	// Opts that do not grant leave every lease column untouched. A grant
	// with no WorkerID is refused with ErrLeaseWithoutWorker and claims
	// nothing.
	//
	// Every backend must pass storetest.RunDequeueSuite, which is the
	// contract this signature only sketches.
	DequeueJobs(ctx context.Context, opts DequeueOpts) ([]*Job, error)

	// GetJob retrieves a job by ID.
	GetJob(ctx context.Context, jobID id.JobID) (*Job, error)

	// UpdateJob persists changes to an existing job.
	UpdateJob(ctx context.Context, j *Job) error

	// DeleteJob removes a job by ID.
	DeleteJob(ctx context.Context, jobID id.JobID) error

	// ListJobsByState returns jobs matching the given state.
	ListJobsByState(ctx context.Context, state State, opts ListOpts) ([]*Job, error)

	// HeartbeatJob updates the heartbeat timestamp for a running job,
	// indicating the worker is still alive.
	HeartbeatJob(ctx context.Context, jobID id.JobID, workerID id.WorkerID) error

	// ReapStaleJobs returns running jobs whose last heartbeat is older than
	// the given threshold, indicating the worker may have crashed.
	ReapStaleJobs(ctx context.Context, threshold time.Duration) ([]*Job, error)

	// CountJobs returns the number of jobs matching the given options.
	CountJobs(ctx context.Context, opts CountOpts) (int64, error)
}

type Usage struct {
	ID      id.UsageID `json:"id"`
	JobID   id.JobID   `json:"job_id"`
	Name    string     `json:"name"`
	Queue   string     `json:"queue"`
	Attempt int        `json:"attempt"`

	// Status is the terminal state this attempt reached. Failed attempts
	// are recorded too: a job that OOMs is the single most informative
	// data point about how much memory it needed.
	Status State `json:"status"`

	// InputBytes totals the artifacts bound to the job, or zero when it
	// declared none.
	InputBytes int64 `json:"input_bytes"`

	// Resources is what was predicted, empty when the resource model is
	// off.
	Resources resource.Set `json:"resources,omitempty"`

	// Measured consumption. Every field is zero when the rung that ran
	// the attempt could not account it: the in-process rung knows only
	// wall time, because a handler sharing the worker's address space has
	// no separable RSS.
	WallTime    time.Duration `json:"wall_time"`
	CPUTime     time.Duration `json:"cpu_time,omitempty"`
	PeakRSS     int64         `json:"peak_rss,omitempty"`
	DiskWritten int64         `json:"disk_written,omitempty"`

	// Executor names the rung that ran the attempt, so measurements taken
	// under different isolation are not silently pooled.
	Executor string `json:"executor,omitempty"`

	ScopeAppID string      `json:"scope_app_id,omitempty"`
	ScopeOrgID string      `json:"scope_org_id,omitempty"`
	WorkerID   id.WorkerID `json:"worker_id,omitempty"`
	RecordedAt time.Time   `json:"recorded_at"`
}

type UsageListOpts struct {
	// Name filters to one job definition. Empty means all.
	Name string
	// Since bounds the query to attempts recorded at or after this time.
	Since time.Time
	// Limit caps the rows returned. Zero means no limit.
	Limit int
	// Offset skips rows, for paging.
	Offset int
}

type UsageRecorder interface {
	// RecordJobUsage persists one attempt's measurements.
	RecordJobUsage(ctx context.Context, u *Usage) error

	// ListJobUsage returns recorded attempts, most recent first.
	ListJobUsage(ctx context.Context, opts UsageListOpts) ([]*Usage, error)

	// PurgeJobUsage deletes records older than before, up to limit rows,
	// and returns how many it removed.
	//
	// This table grows with every attempt the fleet makes, so something
	// has to bound it. Retention is the operator's call, which is why the
	// cutoff is a parameter rather than a policy baked in here.
	PurgeJobUsage(ctx context.Context, before time.Time, limit int) (int64, error)
}