sidekiq

package
v1.12.0 Latest Latest
Warning

This package is not in the latest version of its module.

 Go to latest Published: Nov 6, 2026 License: Apache-2.0 Imports: 9 Imported by: 2

Documentation

Index

Constants

View Source 
const (
	DispatchInterval  = 30 * time.Second
	HeartbeatInterval = 1 * time.Minute
	StaleAfter        = 15 * time.Minute
	// MaxAttempts caps re-claims before a job is permanently failed, preventing poison-job loops.
	MaxAttempts = 5
)

Variables

This section is empty.

Functions

func IsDuplicateJobError added in v1.6.1 

func IsDuplicateJobError(err error) bool

IsDuplicateJobError reports whether err is a primary-key collision on the job row, i.e. another node enqueued the same deterministic job ID first. Producers that derive the ID from the work itself (an archive window, a scheduled rotation) treat this as "someone else owns it", not as a failure. Matching on message text is the only option: the store surfaces the driver's error verbatim and defines no sentinel for it.

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 JobSummary added in v1.9.0 

type JobSummary struct {
	Done    int64
	Total   int64
	Message string
}

JobSummary is the kind-independent view of a job's progress, derived from its opaque metadata.

type ProgressFunc 

type ProgressFunc func(metadata string) error

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 

func New(store Store, logger schemas.Logger, maxConcurrent int, runnerID string) *Runner

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 

func (r *Runner) Cancel(ctx context.Context, id string) (bool, error)

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 

func (r *Runner) Enqueue(ctx context.Context, id, kind, metadata, createdBy string) error

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) QuietOnSuccessKinds added in v1.11.0 

func (r *Runner) QuietOnSuccessKinds() []string

QuietOnSuccessKinds returns the kinds registered with RegisterQuietOnSuccess, sorted.

func (*Runner) Register 

func (r *Runner) Register(kind string, fn HandlerFunc)

Register binds a handler to a job kind. Call before enqueuing.

func (*Runner) RegisterQuietOnSuccess added in v1.11.0 

func (r *Runner) RegisterQuietOnSuccess(kind string)

RegisterQuietOnSuccess marks a kind as routine: a job of it that completed is left out of job listings, while one waiting, running, failed or cancelled is still shown. For work queued on a timer, which would otherwise fill a listing with runs nobody has to act on.

func (*Runner) RegisterSummarizer added in v1.9.0 

func (r *Runner) RegisterSummarizer(kind string, fn SummarizeFunc)

RegisterSummarizer binds a progress extractor to a job kind so generic views can show done/total without knowing the kind's metadata shape. Kinds without one report no progress.

func (*Runner) Shutdown 

func (r *Runner) Shutdown()

Shutdown cancels the background context and waits for in-flight goroutines to return.

func (*Runner) StartDispatcher 

func (r *Runner) StartDispatcher(interval, staleAfter time.Duration) (stop func())

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.

func (*Runner) Summarize added in v1.9.0 

func (r *Runner) Summarize(kind, metadata string) (JobSummary, bool)

Summarize returns the progress of a job of the given kind and whether the kind has a summarizer.

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.

type SummarizeFunc added in v1.9.0 

type SummarizeFunc func(metadata string) JobSummary

SummarizeFunc extracts a JobSummary from a job's metadata JSON. It must tolerate malformed input.

Source Files

View all Source files

Jump to

Keyboard shortcuts

? : This menu
/ : Search site
f or F : Jump to
y or Y : Canonical URL
go.dev uses cookies from Google to deliver and enhance the quality of its services and to analyze traffic. Learn more.