Documentation
¶
Index ¶
- Constants
- type HandlerFunc
- type ProgressFunc
- type Runner
- func (r *Runner) Cancel(ctx context.Context, id string) (bool, error)
- func (r *Runner) Enqueue(ctx context.Context, id, kind, metadata, createdBy string) error
- func (r *Runner) EnqueuePartitioned(ctx context.Context, id, kind, partitioningKey, metadata, createdBy string) error
- func (r *Runner) Register(kind string, fn HandlerFunc)
- func (r *Runner) Shutdown()
- func (r *Runner) StartDispatcher(interval, staleAfter time.Duration) (stop func())
- type Store
Constants ¶
Variables ¶
This section is empty.
Functions ¶
This section is empty.
Types ¶
type HandlerFunc ¶
type HandlerFunc func(ctx context.Context, job tables.TableSidekiqJob, progress ProgressFunc) (finalMetadata string, err error)
HandlerFunc processes one job. Receives the job row (read Metadata for the resume cursor) and a progress callback. Returns final metadata and an error. Nil error completes the job; non-nil fails it. The context is cancelled if this node loses ownership or on shutdown.
type ProgressFunc ¶
ProgressFunc persists a checkpoint and bumps the heartbeat. Handlers call it after each unit of work.
type Runner ¶
type Runner struct {
// contains filtered or unexported fields
}
Runner owns the handler registry and job goroutine lifecycle.
func New ¶
New creates a Runner. maxConcurrent bounds simultaneous job goroutines (<=0 defaults to 4). Pass the node ID as runnerID in cluster mode. Pass "" to make any running job immediately re-claimable on restart (no stale window).
func (*Runner) Cancel ¶ added in v1.5.9
Cancel stops a pending or running job. It flips the durable row to cancelled first — so the dispatcher will not re-claim it and the reaper will not fail it, whatever happens next — and then cancels the handler's context if the job happens to be running on this node, which stops the work within one checkpoint instead of one heartbeat.
A job owned by another node is stopped by that node's next heartbeat, which is fenced on status = running and therefore reads the cancellation as lost ownership. Cancelling an unknown or already-terminal job is a no-op: it returns false with no error, since the caller's intent ("this job should not be running") already holds.
func (*Runner) Enqueue ¶
Enqueue persists a new pending job and starts it as soon as a concurrency slot is free. Returns once the DB row is committed so the caller can respond immediately.
func (*Runner) EnqueuePartitioned ¶ added in v1.5.5
func (r *Runner) EnqueuePartitioned(ctx context.Context, id, kind, partitioningKey, metadata, createdBy string) error
EnqueuePartitioned is Enqueue for a job that must run serially within partitioningKey: jobs sharing a non-empty key run one-at-a-time in FIFO (created_at) order across the cluster (see ClaimPartitionedSidekiqJob). An empty key behaves exactly like Enqueue. The eager spawn is safe either way — a not-yet-runnable partitioned job simply fails to claim and is picked up later by the dispatcher.
func (*Runner) Register ¶
func (r *Runner) Register(kind string, fn HandlerFunc)
Register binds a handler to a job kind. Call before enqueuing.
func (*Runner) Shutdown ¶
func (r *Runner) Shutdown()
Shutdown cancels the background context and waits for in-flight goroutines to return.
func (*Runner) StartDispatcher ¶
StartDispatcher scans for claimable jobs on an interval. Uses non-blocking semaphore acquisition so it never spawns more goroutines than available concurrency slots — remaining jobs are left for the next tick. Runs one scan immediately on start.
type Store ¶
type Store interface {
CreateSidekiqJob(ctx context.Context, job *tables.TableSidekiqJob) error
GetSidekiqJob(ctx context.Context, id string) (*tables.TableSidekiqJob, error)
// ClaimSidekiqJob atomically claims a job for runnerID; returns true only for the winner.
ClaimSidekiqJob(ctx context.Context, id, runnerID string, staleBefore time.Time) (bool, error)
// ClaimPartitionedSidekiqJob claims a partitioning-keyed job only when it is next in
// line for its key (no other same-key job running, no older same-key job pending).
ClaimPartitionedSidekiqJob(ctx context.Context, id, runnerID string, staleBefore time.Time, partitioningKey string, createdAt time.Time) (bool, error)
// HeartbeatSidekiqJob bumps updated_at for a job still owned by runnerID; returns false on lost ownership.
HeartbeatSidekiqJob(ctx context.Context, id, runnerID string) (bool, error)
UpdateSidekiqJobProgress(ctx context.Context, id, runnerID, metadata string) error
CompleteSidekiqJob(ctx context.Context, id, runnerID, metadata string) error
FailSidekiqJob(ctx context.Context, id, runnerID, metadata, lastErr string) error
// CancelSidekiqJob flips a pending or running job to cancelled. Returns true only
// when this call performed the transition (false = unknown or already terminal).
CancelSidekiqJob(ctx context.Context, id string) (bool, error)
// FinalizeCancelledSidekiqJob stores the last metadata snapshot of a cancelled job
// so its partial progress survives.
FinalizeCancelledSidekiqJob(ctx context.Context, id, runnerID, metadata string) error
// ListClaimableSidekiqJobs returns pending jobs and running jobs whose heartbeat is older than staleBefore.
ListClaimableSidekiqJobs(ctx context.Context, staleBefore time.Time) ([]tables.TableSidekiqJob, error)
}
Store is the narrow subset of configstore the runner needs.
Source Files