Documentation
¶
Index ¶
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) 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) 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)
// 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
// 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