jobqueue

README

job-queue

To manage small jobs

Storage backends

store/gorm implements JobStorage on top of GORM and only depends on gorm.io/gorm core, not any specific driver — pass in any *gorm.DB and it works. It has been verified against:

• MySQL — schema: store/gorm/migration_mysql.sql, example: test/gorm
• Postgres — schema: store/gorm/migration_postgres.sql, example: test/gorm-postgres
• SQLite — schema: store/gorm/migration_sqlite.sql, example: test/gorm-sqlite (uses the pure-Go github.com/glebarez/sqlite driver, so no CGO/C compiler is required)

SQLite has no SELECT ... FOR UPDATE; the store detects the dialect (db.Name() == "sqlite") and skips that clause there, relying on SQLite's own transaction-level write serialization instead. Because of that, SQLite is best suited to single-process/low-concurrency deployments — concurrent claim attempts under real contention may surface as a "database is locked" error rather than blocking.

Duplicate-job locking

CheckDuplicateJob runs in its own short transaction, separate from the CreateJob/InjectJob insert that follows it — on purpose, so the dedup check never sits inside a long-lived job transaction and lock-waits against GetAndLockAvailableJob's claim scan (that combination is what used to cause deadlocks). The tradeoff is that CheckDuplicateJob by itself cannot prevent two concurrent callers from both passing the check and both inserting.

The actual guarantee comes from CreateJob/InjectJob, which take a cross-instance named lock scoped to (title, job_id) before re-checking and inserting:

• Postgres: pg_advisory_xact_lock — a separate lock-manager namespace from table/row locks, so it can't deadlock against the claim scan, and it auto-releases at transaction end (commit or rollback) regardless of who calls it. Fully race-free in every calling pattern.
• MySQL: GET_LOCK/RELEASE_LOCK — same deadlock-safety property, but since MySQL scopes the lock to the session/connection rather than the transaction, this store releases it explicitly right after its own insert. That's fully race-free for InjectJob and for CreateJob called with a plain (not already open) *gorm.DB, since the store then owns the whole transaction itself. If CreateJob is instead given a transaction the caller had already opened (e.g. to share job creation with its own writes) and holds open well beyond the insert before committing, there's a narrow window on MySQL specifically where a concurrent submission for the same key could still slip in — commit promptly in that pattern, or prefer Postgres, if that matters for your workload.
• SQLite: no named-lock primitive exists, but none is needed — SQLite already serializes writers at the transaction level.

CheckDuplicateJob itself is still called first by Foreman.AddJob* as a cheap best-effort pre-check (fail fast without opening a second transaction), but it is not what prevents the race — the lock-protected recheck inside CreateJob/InjectJob is. Separately, Foreman also stripes an in-process mutex over (title, job_id) around all AddJob* calls: it's a same-instance optimization only (fewer redundant round-trips to the DB lock when the same process double-submits), not a substitute for the DB-level lock, which is the only thing that protects against concurrent calls across multiple instances.

Running the store/gorm tests

store/gorm/store_test.go always runs against SQLite (a fresh temp file per test, no setup needed). MySQL and Postgres are included automatically when their DSN env vars are set, and silently skipped otherwise:

docker compose -f store/gorm/docker-compose.test.yml up -d
# wait for both to report healthy: docker compose -f store/gorm/docker-compose.test.yml ps

export JOBQUEUE_TEST_MYSQL_DSN="root:jobqueue@tcp(127.0.0.1:3307)/jobqueue?charset=utf8mb4&parseTime=True&loc=Local"
export JOBQUEUE_TEST_POSTGRES_DSN="host=127.0.0.1 port=5544 user=postgres password=jobqueue dbname=jobqueue sslmode=disable"
go test ./store/gorm/... -v

docker compose -f store/gorm/docker-compose.test.yml down -v

The compose file relaxes MySQL's fsync durability (innodb-flush-log-at-trx-commit=0, sync-binlog=0) purely for test-suite speed — real disk fsync under Docker Desktop's virtualized filesystem was measured at ~400ms/commit, enough on its own to make 30-way lock contention tests time out for reasons that have nothing to do with the code under test. Never use these settings against data you care about.

-race needs a C compiler, which this may not have on every machine (e.g. Windows without a configured gcc). To run the race detector anyway, use a Linux Go container on the same Docker network as the databases:

docker run --rm --network jobqueue-store-test_default \
  -v "$PWD:/src" -w /src \
  -e JOBQUEUE_TEST_MYSQL_DSN="root:jobqueue@tcp(jobqueue-store-test-mysql-1:3306)/jobqueue?charset=utf8mb4&parseTime=True&loc=Local" \
  -e JOBQUEUE_TEST_POSTGRES_DSN="host=jobqueue-store-test-postgres-1 port=5432 user=postgres password=jobqueue dbname=jobqueue sslmode=disable" \
  golang:1.24 go test -race ./store/gorm/... -v

This combination (real MySQL/Postgres via Docker, concurrent test, -race) is what caught a real bug during development: the MySQL lock-release call was originally issued on the same *gorm.DB handle that had just committed, and a *sql.Tx can't be reused after Commit() - so RELEASE_LOCK silently failed and every lock leaked after its first use, serializing all later callers behind GET_LOCK's timeout. Fixed by acquiring MySQL's lock on its own dedicated *sql.Conn, decoupled from the data transaction's lifecycle, so release always executes regardless of when the data transaction commits.

Documentation

Index

• Constants
• Variables
• type Foreman
  • func NewForeman(config *QueueConfig, store JobStorage, logger Logger) Foreman
• type Governor
  • func NewLocalOndemandGovernor(max, min, maxworker int, jd map[string]JobDescription) Governor
• type HandlerFunc
• type Job
• type JobDescription
• type JobStorage
• type LocalOndemandGovernor
  • func (g LocalOndemandGovernor) AddJob(title string)
  • func (g LocalOndemandGovernor) DelJob(title string)
  • func (g LocalOndemandGovernor) GetCounter() int
  • func (g LocalOndemandGovernor) NoJob()
  • func (g LocalOndemandGovernor) Spawn() (bool, []string)
• type Logger
• type MiddlewareFunc
• type QueueConfig
• type ScheduleJob

Constants

const ContextJobKey = "job-key"

const ContextQueueKey = "queue-key"

const MaxTime int64 = 9223372036854775807

Variables

var JobError = joberror{
	INVALID_JD:     errors.New("invalid job description"),
	INVALID_JOB:    errors.New("invalid job"),
	INVALID_WORKER: errors.New("invalid worker"),
	INVALID_PL:     errors.New("invalid production line"),
	NOT_FOUND:      errors.New("not found"),
	TERMINATING:    errors.New("terminating"),
	DUPLICATE_JOB:  errors.New("duplicate job"),
}

var JobStatus = jobstatus{
	INIT:       "init",
	PROCESSING: "processing",
	RETRY:      "retry",
	ERROR:      "error",
	MAX_RETRY:  "max_retry",
	DONE:       "done",
}

var StoreError = storeerror{
	INVALID_GORM_TX: errors.New("invalid gorm transaction"),
	INVALID_GORM_DB: errors.New("invalid gorm session"),
	LOCK_TIMEOUT:    errors.New("timed out waiting for job lock"),
}

Types

type Foreman

type Foreman interface {
	//AddWorker add a worker function for handler
	AddWorker(title string, worker HandlerFunc, midl ...MiddlewareFunc) error

	//AddJobTransaction add a new job to the queue
	AddJobTransaction(ctx context.Context, job *Job) error

	//AddJob add a new job to the queue without context, dont recommend but handy for multiple context
	AddJob(job *Job) error

	//AddJobWithSchedule add a new job to the queue to run at a schedule
	AddJobWithSchedule(ctx context.Context, job *Job, runat int64) error

	//Serve start the worker service
	Serve() error

	//Strike graceful shutdown service. Reject new jobs and wait for running one to finish. Force kill at TTL.
	Strike(ttl int) error // ttl in seconds

	//AddMiddleware add global middleware, order of adding matters
	AddMiddleware(midl ...MiddlewareFunc)

	//GetCounter return worker counter for debug purpose
	GetCounter() int
}

func NewForeman

func NewForeman(config *QueueConfig, store JobStorage, logger Logger) Foreman

type Governor

type Governor interface {
	AddJob(string)
	DelJob(string)
	NoJob()
	Spawn() (bool, []string)
	GetCounter() int
}

func NewLocalOndemandGovernor

func NewLocalOndemandGovernor(max, min, maxworker int, jd map[string]JobDescription) Governor

OndemandGovernor ondemand governor

type HandlerFunc

type HandlerFunc func(ctx context.Context) error

type Job

type Job struct {
	ID        string `mapstructure:"id" yaml:"id" json:"id"`
	JobID     string `mapstructure:"job_id" yaml:"job_id" json:"job_id"`
	Title     string `mapstructure:"title" yaml:"title" json:"title"`
	Payload   string `mapstructure:"payload" yaml:"payload" json:"payload"`
	Try       int    `mapstructure:"try" yaml:"try" json:"try"`
	Priority  int    `mapstructure:"priority" yaml:"priority" json:"priority"`
	Status    string `mapstructure:"status" yaml:"status" json:"status"`
	Result    string `yaml:"result" mapstructure:"result" json:"result"`
	Message   string `yaml:"message" mapstructure:"message" json:"message"`
	UpdatedAt int64  `yaml:"updated_at" mapstructure:"updated_at" json:"updated_at"`
}

type JobDescription

type JobDescription struct {
	Title      string `mapstructure:"title" yaml:"title" json:"title"`
	TTL        int    `mapstructure:"ttl" yaml:"ttl" json:"ttl"`                      // seconds
	Concurrent int    `mapstructure:"concurrent" yaml:"concurrent" json:"concurrent"` // 0 is invalid
	Priority   int    `mapstructure:"priority" yaml:"priority" json:"priority"`       // 0 is invalid here, 1 is highest
	MaxRetry   int    `mapstructure:"max_retry" yaml:"max_retry" json:"max_retry"`    // number of retry
	Secure     bool   `mapstructure:"secure" yaml:"secure" json:"secure"`
}

Config models

type JobStorage

type JobStorage interface {
	CheckDuplicateJob(job Job) error
	CreateJob(ctx context.Context, job Job) error
	GetAndLockAvailableJob(jd map[string]JobDescription, ignorelist ...string) (*Job, error)
	UpdateJobResult(job Job) error

	InjectJob(Job) error
	CreateScheduleJob(ctx context.Context, job ScheduleJob) error
	GetScheduledJob(from, to int64) ([]*ScheduleJob, error)
	UpdateScheduledJob(ScheduleJob) error
}

type LocalOndemandGovernor

type LocalOndemandGovernor struct {
	MaxSleep       int
	MinSleep       int
	CurSleep       *int
	WorkerCounter  *int
	Locker         *sync.Mutex
	JobCounter     map[string]*int
	MaxWorker      int
	JobDescription map[string]JobDescription
}

func (LocalOndemandGovernor) AddJob

func (g LocalOndemandGovernor) AddJob(title string)

func (LocalOndemandGovernor) DelJob

func (g LocalOndemandGovernor) DelJob(title string)

func (LocalOndemandGovernor) GetCounter

func (g LocalOndemandGovernor) GetCounter() int

func (LocalOndemandGovernor) NoJob

func (g LocalOndemandGovernor) NoJob()

func (LocalOndemandGovernor) Spawn

func (g LocalOndemandGovernor) Spawn() (bool, []string)

type Logger

type Logger interface {
	Debug(msg string)
	Info(msg string)
	Warn(msg string)
	Error(msg string)
	Fatal(msg string)
	Panic(msg string)
}

type MiddlewareFunc

type MiddlewareFunc func(HandlerFunc) HandlerFunc

type QueueConfig

type QueueConfig struct {
	JobDescription map[string]JobDescription `mapstructure:"job_description" yaml:"job_description" json:"job_description"`
	CleanError     bool                      `yaml:"clean_error" mapstructure:"clean_error" json:"clean_error"` // will the GC clear maxed try job
	Concurrent     int                       `yaml:"concurrent" mapstructure:"concurrent" json:"concurrent"`    // number of workers can be run, will ignore JD.Concurrent if it >
	BreakTime      int                       `mapstructure:"break_time" yaml:"break_time" json:"break_time"`    // miliseconds, break between loop
	RampTime       int                       `mapstructure:"ramp_time" yaml:"ramp_time" json:"ramp_time"`
}

type ScheduleJob

type ScheduleJob struct {
	ID        string `mapstructure:"id" yaml:"id" json:"id"`
	JobID     string `mapstructure:"job_id" yaml:"job_id" json:"job_id"`
	Title     string `mapstructure:"title" yaml:"title" json:"title"`
	Payload   string `mapstructure:"payload" yaml:"payload" json:"payload"`
	Priority  int    `mapstructure:"priority" yaml:"priority" json:"priority"`
	Status    string `mapstructure:"status" yaml:"status" json:"status"`
	UpdatedAt int64  `yaml:"updated_at" mapstructure:"updated_at" json:"updated_at"`
	Schedule  int64  `json:"schedule"`
	ExecuteID string `json:"execute_id"`
	Log       string `json:"log"`
}