Documentation
¶
Overview ¶
Package queue persists scheduler runs and coordinates their execution and recovery.
Index ¶
- Variables
- type Queue
- func (q *Queue) Cancel(ctx context.Context, environmentID, jobID, runID string) (st.Run, error)
- func (q *Queue) Checkpoint(ctx context.Context, jobID, schedule string, next time.Time) error
- func (q *Queue) Get(ctx context.Context, environmentID, jobID, runID string) (st.Run, error)
- func (q *Queue) List(ctx context.Context, environmentID, jobID string, page, limit int) (st.RunList, error)
- func (q *Queue) Records(ctx context.Context) ([]st.QueueRecord, error)
- func (q *Queue) Resolve(ctx context.Context, environmentID, jobID, runID, resolvedBy string) (st.Run, error)
- func (q *Queue) Retry(ctx context.Context, environmentID, jobID, runID string) (st.Run, error)
- func (q *Queue) SetObserver(observer st.RunObserver)
- func (q *Queue) Start(ctx context.Context) error
- func (q *Queue) Stop(ctx context.Context) error
- func (q *Queue) Submit(ctx context.Context, request st.Request) (st.Run, error)
- func (q *Queue) UpdateRun(ctx context.Context, run st.Run, change func(*st.Run) error) error
Constants ¶
This section is empty.
Variables ¶
var ErrRunConflict = errors.Sentinel("job run state changed")
ErrRunConflict indicates that execution ownership or state changed.
var ErrRunNotFound = errors.Sentinel("job run not found")
ErrRunNotFound indicates that neither a run nor its receipt exists.
Functions ¶
This section is empty.
Types ¶
type Queue ¶
type Queue struct {
// contains filtered or unexported fields
}
Queue persists work before dispatching it to the owning domain.
func New ¶
func New(store *kv.KVService, execute, reconcile func(context.Context, st.Run) (st.Outcome, error)) *Queue
New uses the application's existing KV store; no in-memory admission exists.
func (*Queue) Cancel ¶
Cancel marks an unstarted queued run canceled. Runs already attempted locally or dispatched remotely must be reconciled rather than canceled as pending work.
func (*Queue) Checkpoint ¶
Checkpoint records the next due occurrence atomically with overdue recovery.
func (*Queue) Get ¶
Get retrieves a run by environment, job, and run ID, including compacted idempotency receipts. It returns ErrRunNotFound when no matching record exists.
func (*Queue) List ¶
func (q *Queue) List(ctx context.Context, environmentID, jobID string, page, limit int) (st.RunList, error)
List returns retained run history newest first, excluding compacted receipts. An empty job ID includes every job in the environment; pages are one-based.
func (*Queue) Records ¶
Records reads all persisted queue records, including run history and receipts. Storage or decoding errors fail the read instead of returning an incomplete snapshot.
func (*Queue) Resolve ¶
func (q *Queue) Resolve(ctx context.Context, environmentID, jobID, runID, resolvedBy string) (st.Run, error)
Resolve releases reviewed, inactive work without discarding its execution evidence.
func (*Queue) Retry ¶
Retry requeues a failed, partial, or attention-required run without changing its ID or dropping target progress. Unconfirmed remote work cannot be retried here.
func (*Queue) SetObserver ¶
func (q *Queue) SetObserver(observer st.RunObserver)
SetObserver connects activity projection before the queue starts accepting work.
func (*Queue) Start ¶
Start validates persisted state and starts dispatching on ctx. Repeated calls are harmless, but a stopped queue cannot be restarted; create a new instance.
func (*Queue) Stop ¶
Stop rejects new submissions, cancels execution, and waits for workers until ctx expires. Unfinished claims remain persisted for startup reconciliation.
Source Files