tasker

package module
v0.1.4 Latest Latest
Warning

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

 Go to latest Published: Sep 26, 2026 License: MIT Imports: 9 Imported by: 0

README

Tasker — Background Job System for Lemmego

tasker is a distributed, clusterable, multi-node background job management system for Go. It powers the queue plugin in the Lemmego framework.

Features

  • State machine: pending → scheduled → available → running → completed / retryable / failed / cancelled
  • Job hooks: BeforeHandle, AfterHandle, BeforeRetry, Failed (failure callback)
  • Retry strategies: Exponential, Fibonacci, Fixed backoff with configurable max attempts
  • Multi-node: Leader election, node registry, heartbeats, distributed locking, stale job re-queue
  • SQL driver: PostgreSQL, MySQL, SQLite support with dialect-aware queries
  • Redis driver: Node management, leader election, distributed locking
  • Atomic claiming: SELECT ... FOR UPDATE SKIP LOCKED (Postgres/MySQL), transactional (SQLite)
  • Multiple queues: Independent worker pools per queue with priority ordering
  • Auto-scaling: Horizon-style dynamic worker scaling based on queue backlog
  • Web dashboard: Real-time stats, job management, queue controls, worker monitoring
  • SSE events: Live stat updates via Server-Sent Events
  • CLI commands: Work, dispatch, job management

Architecture

┌──────────────┐  ┌──────────────┐  ┌──────────────┐
│   Node A     │  │   Node B     │  │   Node C     │
│  (Server)    │  │  (Worker)    │  │  (Worker)    │
│  Supervisor  │  │  Supervisor  │  │  Supervisor  │
│  └─Pool (3)  │  │  └─Pool (5)  │  │  └─Pool (10) │
│  └─Web UI    │  │              │  │              │
└──────┬───────┘  └──────┬───────┘  └──────┬───────┘
       │                 │                 │
       └─────────────┬───┴─────────────┬───┘
                     │                 │
             ┌───────▼─────────┐ ┌─────▼─────────┐
             │  PostgreSQL/SQL │ │   Redis        │
             │  (jobs table)   │ │ (pub/sub,      │
             │  SELECT FOR     │ │  locks,        │
             │  UPDATE SKIP    │ │  heartbeat)    │
             │  LOCKED         │ │                │
             └─────────────────┘ └───────────────┘

Quick Start

package main

import (
    "context"
    "fmt"
    "github.com/lemmego/tasker"
    "github.com/lemmego/tasker/driver/sqldriver"
)

type SendEmail struct {
    Email string `json:"email"`
    UserID int   `json:"user_id"`
}

func (j *SendEmail) Handle(ctx context.Context) error {
    fmt.Printf("Sending email to %s\n", j.Email)
    return nil
}

func main() {
    driver, _ := sqldriver.NewDriver(sqldriver.Config{
        DSN:        "postgres://localhost:5432/mydb?sslmode=disable",
        DriverName: "postgres",
    })

    mgr := tasker.NewConfiguredManager(tasker.Config{
        Driver:             driver,
        DefaultQueue:       "default",
        DefaultMaxAttempts: 3,
    })
    tasker.SetGlobal(mgr)

    tasker.RegisterJob("*main.SendEmail", func() tasker.Job {
        return &SendEmail{}
    })

    tasker.Dispatch(context.Background(), &SendEmail{
        Email:  "user@test.com",
        UserID: 1,
    })
}

Package Structure

tasker/
├── tasker.go           # Manager — public API entry point
├── types.go            # Core types: JobRow, State, JobFilter, QueueStats, etc.
├── job.go              # Job interface + optional interfaces
├── driver.go           # Driver interface — implemented by sqldriver/redisdriver
├── state_machine.go    # State transitions + backoff computation
├── pool.go             # Pool + Worker — job claiming and execution
├── registry.go         # Job type registry + global middlewares
├── middleware.go       # Built-in middleware (logging, recovery)
├── provider.go         # Config struct + NewConfiguredManager
├── errors.go           # Sentinel errors
│
├── driver/sqldriver/   # SQL driver (Postgres, MySQL, SQLite)
│   ├── driver.go       #    Full implementation — enqueue, claim, complete, stats, etc.
│   ├── dialect.go      #    Dialect abstraction for Postgres/MySQL/SQLite
│   └── migrations/     #    SQL migration files
│
├── driver/redisdriver/ # Redis driver (node mgmt, leader election, locks)
│   └── driver.go
│
├── supervisor/         # Process lifecycle + auto-scaling
│   └── supervisor.go
│
├── coordinator/        # Multi-node coordination
│   └── coordinator.go
│
├── scheduler/          # Cron-based recurring jobs
│   └── scheduler.go
│
├── batch/              # Batch operations
│   └── batch.go
│
├── web/                # Web UI dashboard
│   ├── web.go          #    HTTP handlers + API routes
│   └── dashboard.go    #    Dashboard HTML template
│
└── cmd/                # CLI commands
    ├── work.go         #    tasker:work
    ├── queue.go        #    tasker:queue
    └── job.go          #    tasker:job

Job Interface

// Required — every job must implement Handle
type Job interface {
    Handle(ctx context.Context) error
}

// Optional interfaces:
type ShouldQueue interface { Queue() QueueName }
type ShouldDelay interface { Delay() time.Duration }
type ShouldRetry interface { MaxAttempts() int }
type ShouldFail interface { Failed(ctx, payload, err) error }
type ShouldTag interface { Tags() []string }
type ShouldBatch interface { BatchID() string }
type ShouldLock interface { LockKey() string; LockTimeout() time.Duration }
type ShouldTimeout interface { Timeout() time.Duration }
type ShouldMiddleware interface { Middleware() []JobMiddleware }

// Lifecycle hooks:
type BeforeHandleHook interface { BeforeHandle(ctx) error }
type AfterHandleHook interface { AfterHandle(ctx) error }
type BeforeRetryHook interface { BeforeRetry(ctx, attempt, err) error }

Job States

pending ──► available ──► running ──► completed
                 ▲            │
                 │            ├──► retryable ──► available (backoff)
                 │            │
                 │            └──► failed (max attempts)
                 │
                 └── cancelled

Driver Configuration

SQL (Postgres / MySQL / SQLite)
sqldriver.Config{
    DSN:          "postgres://localhost:5432/mydb?sslmode=disable",
    DriverName:   "postgres",
    TablePrefix:  "tasker_",
    MaxOpenConns: 25,
    MaxIdleConns: 10,
}
Redis
redisdriver.Config{
    Addr:      "localhost:6379",
    Password:  "",
    DB:        0,
    KeyPrefix: "tasker:",
}

Web UI API Endpoints

Management routes are disabled by default. Before mounting the handler, install an explicit authentication middleware with server.UseAuth(...). For browser session authentication, also call server.UseCSRF(); state-changing requests must then send the X-CSRF-Token header matching the tasker_csrf cookie.

Method Path Description
GET /tasker/ Dashboard HTML
GET /tasker/api/stats Global stats
GET /tasker/api/jobs Paginated job list
GET /tasker/api/jobs/{id} Job detail
POST /tasker/api/jobs/{id}/retry Retry a job
POST /tasker/api/jobs/{id}/cancel Cancel a job
POST /tasker/api/jobs/batch/retry Batch retry
POST /tasker/api/jobs/batch/cancel Batch cancel
GET /tasker/api/queues Queue metrics
POST /tasker/api/queues/{queue}/pause Pause a queue
POST /tasker/api/queues/{queue}/resume Resume a queue
GET /tasker/api/workers Connected nodes
GET /tasker/api/metrics/jobs Per-job-type metrics
GET /tasker/api/metrics/queues Per-queue metrics
GET /tasker/api/events SSE real-time events
POST /tasker/api/prune Prune old jobs

CLI Commands

tasker:work             # Start worker process
  --queue=default,email #   Queues to listen on
  --workers=3           #   Workers per queue

tasker:dispatch         # Dispatch test jobs
  --count=10            #   Number of jobs
  --queue=default       #   Target queue
  --fail-rate=30        #   % chance of failure
  --delay=500           #   Simulated processing delay (ms)

tasker:job retry <id>   # Retry a specific job
tasker:job cancel <id>  # Cancel a specific job
tasker:job clear        # Clear jobs by state
  --states=failed,cancelled

tasker:queue list       # List queue stats

Running Tests

# Core tests
go test ./...

# SQL driver tests (uses SQLite in-memory)
go test ./driver/sqldriver/...

# With race detector
go test -race ./...

License

MIT

Documentation

Index

Constants

This section is empty.

Variables

View Source 
var (
	ErrJobNotFound        = errors.New("job not found")
	ErrJobAlreadyExists   = errors.New("job already exists")
	ErrInvalidTransition  = errors.New("invalid state transition")
	ErrJobNotClaimed      = errors.New("job not claimed by this worker")
	ErrQueueNotFound      = errors.New("queue not found")
	ErrWorkerPoolStopped  = errors.New("worker pool is stopped")
	ErrInvalidPayload     = errors.New("invalid job payload")
	ErrJobTimedOut        = errors.New("job execution timed out")
	ErrJobPanicked        = errors.New("job execution panicked")
	ErrMaxRetriesExceeded = errors.New("max retries exceeded")
	ErrNodeNotFound       = errors.New("node not found")
)

Functions

func AddGlobalMiddleware 

func AddGlobalMiddleware(mw JobMiddleware)

func ListRegisteredJobs 

func ListRegisteredJobs() []string

func RegisterJob 

func RegisterJob(name string, factory JobFactory)

func SetGlobal 

func SetGlobal(m *Manager)

func SetGlobalMiddlewares 

func SetGlobalMiddlewares(mws []JobMiddleware)

func ShouldRequeue 

func ShouldRequeue(nodeTimeout time.Duration) bool

func ValidateTransition 

func ValidateTransition(from, to State) error

Types

type AfterHandleHook 

type AfterHandleHook interface {
	AfterHandle(ctx context.Context) error
}

type AttemptError 

type AttemptError struct {
	Attempt   int       `json:"attempt"`
	Error     string    `json:"error"`
	Timestamp time.Time `json:"timestamp"`
	Stack     string    `json:"stack,omitempty"`
}

type BackoffConfig 

type BackoffConfig struct {
	Strategy BackoffStrategy `json:"strategy"`
	Base     time.Duration   `json:"base"`
	MaxDelay time.Duration   `json:"max_delay"`
}

func DefaultBackoff 

func DefaultBackoff() BackoffConfig

func FibonacciBackoff 

func FibonacciBackoff() BackoffConfig

type BackoffStrategy 

type BackoffStrategy string
const (
	BackoffExponential BackoffStrategy = "exponential"
	BackoffFibonacci   BackoffStrategy = "fibonacci"
	BackoffFixed       BackoffStrategy = "fixed"
)

type BeforeHandleHook 

type BeforeHandleHook interface {
	BeforeHandle(ctx context.Context) error
}

type BeforeRetryHook 

type BeforeRetryHook interface {
	BeforeRetry(ctx context.Context, attempt int, err error) error
}

type Config 

type Config struct {
	Driver             Driver
	DefaultQueue       QueueName
	DefaultMaxAttempts int
	NodeID             NodeID
}

func DefaultAppConfig 

func DefaultAppConfig() Config

type DispatchOpt 

type DispatchOpt func(*dispatchOptions)

func OnQueue 

func OnQueue(name QueueName) DispatchOpt

func WithBatchID 

func WithBatchID(id string) DispatchOpt

func WithDelay 

func WithDelay(d time.Duration) DispatchOpt

func WithMetadata 

func WithMetadata(k, v string) DispatchOpt

func WithPriority 

func WithPriority(p int) DispatchOpt

func WithTags 

func WithTags(tags ...string) DispatchOpt

type Driver 

type Driver interface {
	Enqueue(ctx context.Context, job *JobRow) error
	EnqueueBatch(ctx context.Context, jobs []*JobRow) error
	Claim(ctx context.Context, queue QueueName, nodeID NodeID, max int) ([]*JobRow, error)
	Complete(ctx context.Context, id JobID, output []byte) error
	Fail(ctx context.Context, id JobID, err error) error
	ScheduleRetry(ctx context.Context, id JobID, err error, scheduledAt time.Time) error
	Retry(ctx context.Context, id JobID) (*JobRow, error)
	RetryBatch(ctx context.Context, ids []JobID) error
	Cancel(ctx context.Context, id JobID) (*JobRow, error)
	CancelBatch(ctx context.Context, ids []JobID) error
	GetByID(ctx context.Context, id JobID) (*JobRow, error)
	QueryJobs(ctx context.Context, filter JobFilter) ([]*JobRow, int64, error)
	QueueStats(ctx context.Context, queue QueueName) (*QueueStats, error)
	GlobalStats(ctx context.Context) (*GlobalStats, error)
	JobStats(ctx context.Context, kind string) (*JobTypeStats, error)
	Ping(ctx context.Context) error
	Close() error
	RegisterNode(ctx context.Context, node NodeInfo, ttl time.Duration) error
	DeregisterNode(ctx context.Context, nodeID NodeID) error
	Heartbeat(ctx context.Context, nodeID NodeID, ttl time.Duration) error
	ListNodes(ctx context.Context) ([]NodeInfo, error)
	LeaderElection(ctx context.Context, nodeID NodeID, ttl time.Duration) (bool, error)
	IsLeader(ctx context.Context, nodeID NodeID) (bool, error)
	ResignLeadership(ctx context.Context, nodeID NodeID) error
	AcquireLock(ctx context.Context, key string, nodeID NodeID, ttl time.Duration) (bool, error)
	ReleaseLock(ctx context.Context, key string, nodeID NodeID) error
	RequeueStale(ctx context.Context, timeout time.Duration) (int64, error)
	Prune(ctx context.Context, before time.Time, states []State) (int64, error)
}

type GlobalStats 

type GlobalStats struct {
	Status        string `json:"status"`
	JobsPerMinute int64  `json:"jobs_per_minute"`
	Processes     int    `json:"processes"`
	FailedJobs    int64  `json:"failed_jobs"`
	RecentJobs    int64  `json:"recent_jobs"`
	PausedMasters int    `json:"paused_masters"`
}

type Job 

type Job interface {
	Handle(ctx context.Context) error
}

type JobFactory 

type JobFactory func() Job

func GetRegisteredJob 

func GetRegisteredJob(name string) (JobFactory, bool)

type JobFilter 

type JobFilter struct {
	States  []State     `json:"states,omitempty"`
	Queues  []QueueName `json:"queues,omitempty"`
	Kinds   []string    `json:"kinds,omitempty"`
	Tags    []string    `json:"tags,omitempty"`
	Search  string      `json:"search,omitempty"`
	Limit   int         `json:"limit,omitempty"`
	Offset  int         `json:"offset,omitempty"`
	OrderBy string      `json:"order_by,omitempty"`
	Order   string      `json:"order,omitempty"`
}

type JobID 

type JobID uint64

type JobMiddleware 

type JobMiddleware func(ctx context.Context, job Job, payload []byte, next func(ctx context.Context) error) error

func GetGlobalMiddlewares 

func GetGlobalMiddlewares() []JobMiddleware

func LoggingMiddleware 

func LoggingMiddleware() JobMiddleware

func RecoveryMiddleware 

func RecoveryMiddleware() JobMiddleware

type JobRow 

type JobRow struct {
	ID          JobID             `json:"id"`
	UUID        string            `json:"uuid"`
	Queue       QueueName         `json:"queue"`
	Kind        string            `json:"kind"`
	Payload     []byte            `json:"payload"`
	State       State             `json:"state"`
	Priority    int               `json:"priority"`
	Attempt     int               `json:"attempt"`
	MaxAttempts int               `json:"max_attempts"`
	AttemptedBy []NodeID          `json:"attempted_by"`
	AttemptedAt *time.Time        `json:"attempted_at"`
	Errors      []AttemptError    `json:"errors"`
	Output      []byte            `json:"output,omitempty"`
	Tags        []string          `json:"tags"`
	ScheduledAt time.Time         `json:"scheduled_at"`
	CreatedAt   time.Time         `json:"created_at"`
	StartedAt   *time.Time        `json:"started_at"`
	CompletedAt *time.Time        `json:"completed_at"`
	FinalizedAt *time.Time        `json:"finalized_at"`
	NodeID      NodeID            `json:"node_id,omitempty"`
	BatchID     string            `json:"batch_id,omitempty"`
	Timeout     time.Duration     `json:"timeout,omitempty"`
	Metadata    map[string]string `json:"metadata,omitempty"`
	UniqueKey   string            `json:"unique_key,omitempty"`
}

type JobTypeStats 

type JobTypeStats struct {
	Kind         string  `json:"kind"`
	Throughput   float64 `json:"throughput"`
	AvgRuntimeMs float64 `json:"avg_runtime_ms"`
	FailedCount  int64   `json:"failed_count"`
	TotalCount   int64   `json:"total_count"`
	Samples      int64   `json:"samples"`
}

type Manager 

type Manager struct {
	// contains filtered or unexported fields
}

func Global 

func Global() *Manager

func NewConfiguredManager 

func NewConfiguredManager(cfg Config) *Manager

func NewManager 

func NewManager(opts ...ManagerOption) *Manager

func (*Manager) Cancel 

func (m *Manager) Cancel(ctx context.Context, id JobID) (*JobRow, error)

func (*Manager) CancelBatch 

func (m *Manager) CancelBatch(ctx context.Context, ids []JobID) error

func (*Manager) Chain 

func (m *Manager) Chain(ctx context.Context, jobs []Job, opts ...DispatchOpt) ([]*JobRow, error)

func (*Manager) Config 

func (m *Manager) Config() ManagerConfig

func (*Manager) Dispatch 

func (m *Manager) Dispatch(ctx context.Context, job Job, opts ...DispatchOpt) (*JobRow, error)

func (*Manager) DispatchBatch 

func (m *Manager) DispatchBatch(ctx context.Context, jobs []Job, opts ...DispatchOpt) ([]*JobRow, error)

func (*Manager) Driver 

func (m *Manager) Driver() Driver

func (*Manager) GetJob 

func (m *Manager) GetJob(ctx context.Context, id JobID) (*JobRow, error)

func (*Manager) GlobalStats 

func (m *Manager) GlobalStats(ctx context.Context) (*GlobalStats, error)

func (*Manager) JobStats 

func (m *Manager) JobStats(ctx context.Context, kind string) (*JobTypeStats, error)

func (*Manager) QueryJobs 

func (m *Manager) QueryJobs(ctx context.Context, filter JobFilter) ([]*JobRow, int64, error)

func (*Manager) QueueStats 

func (m *Manager) QueueStats(ctx context.Context, queue QueueName) (*QueueStats, error)

func (*Manager) Retry 

func (m *Manager) Retry(ctx context.Context, id JobID) (*JobRow, error)

func (*Manager) RetryBatch 

func (m *Manager) RetryBatch(ctx context.Context, ids []JobID) error

func (*Manager) Stop 

func (m *Manager) Stop(ctx context.Context) error

func (*Manager) Work 

func (m *Manager) Work(ctx context.Context, queue QueueName, concurrency int) error

type ManagerConfig 

type ManagerConfig struct {
	DefaultQueue       QueueName
	DefaultMaxAttempts int
	DefaultTimeout     time.Duration
	NodeID             NodeID
}

func DefaultConfig 

func DefaultConfig() ManagerConfig

type ManagerOption 

type ManagerOption func(*Manager)

func WithDefaultQueue 

func WithDefaultQueue(name QueueName) ManagerOption

func WithDefaultTimeout added in v0.1.3 

func WithDefaultTimeout(timeout time.Duration) ManagerOption

func WithDriver 

func WithDriver(d Driver) ManagerOption

func WithMaxAttempts 

func WithMaxAttempts(n int) ManagerOption

func WithNodeID 

func WithNodeID(id NodeID) ManagerOption

type NodeID 

type NodeID string

type NodeInfo 

type NodeInfo struct {
	ID            NodeID      `json:"id"`
	Host          string      `json:"host"`
	Port          int         `json:"port"`
	Queues        []QueueName `json:"queues"`
	Workers       int         `json:"workers"`
	Status        string      `json:"status"`
	StartedAt     time.Time   `json:"started_at"`
	LastHeartbeat time.Time   `json:"last_heartbeat"`
	Version       string      `json:"version"`
}

type Pool 

type Pool struct {
	// contains filtered or unexported fields
}

func NewPool 

func NewPool(mgr *Manager, queue QueueName, maxWorkers int) *Pool

func (*Pool) Queue 

func (p *Pool) Queue() QueueName

func (*Pool) Scale 

func (p *Pool) Scale(target int) error

func (*Pool) Start 

func (p *Pool) Start(ctx context.Context) error

func (*Pool) Stop 

func (p *Pool) Stop(ctx context.Context) error

func (*Pool) WorkerCount 

func (p *Pool) WorkerCount() int

type PoolManager 

type PoolManager struct {
	// contains filtered or unexported fields
}

func NewPoolManager 

func NewPoolManager(mgr *Manager) *PoolManager

func (*PoolManager) Pools 

func (pm *PoolManager) Pools() map[QueueName]*Pool

func (*PoolManager) ScalePool 

func (pm *PoolManager) ScalePool(queue QueueName, target int) error

func (*PoolManager) StartPool 

func (pm *PoolManager) StartPool(ctx context.Context, queue QueueName, concurrency int) (*Pool, error)

func (*PoolManager) StopAll 

func (pm *PoolManager) StopAll(ctx context.Context) error

func (*PoolManager) StopPool 

func (pm *PoolManager) StopPool(ctx context.Context, queue QueueName) error

func (*PoolManager) WorkerCount 

func (pm *PoolManager) WorkerCount() int

type QueueName 

type QueueName string

type QueueStats 

type QueueStats struct {
	Queue            QueueName `json:"queue"`
	Available        int64     `json:"available"`
	Running          int64     `json:"running"`
	Completed        int64     `json:"completed"`
	Failed           int64     `json:"failed"`
	Retryable        int64     `json:"retryable"`
	Scheduled        int64     `json:"scheduled"`
	AvgRuntimeMs     float64   `json:"avg_runtime_ms"`
	ThroughputPerMin float64   `json:"throughput_per_min"`
	WaitTimeMs       float64   `json:"wait_time_ms"`
}

type ShouldBatch 

type ShouldBatch interface {
	BatchID() string
}

type ShouldDelay 

type ShouldDelay interface {
	Delay() time.Duration
}

type ShouldFail 

type ShouldFail interface {
	Failed(ctx context.Context, payload []byte, err error) error
}

type ShouldLock 

type ShouldLock interface {
	LockKey() string
	LockTimeout() time.Duration
}

type ShouldMiddleware 

type ShouldMiddleware interface {
	Middleware() []JobMiddleware
}

type ShouldQueue 

type ShouldQueue interface {
	Queue() QueueName
}

type ShouldRetry 

type ShouldRetry interface {
	MaxAttempts() int
}

type ShouldRetryUntil 

type ShouldRetryUntil interface {
	RetryUntil() time.Time
}

type ShouldRetryWithBackoff 

type ShouldRetryWithBackoff interface {
	ShouldRetry
	RetryBackoff() BackoffConfig
}

type ShouldTag 

type ShouldTag interface {
	Tags() []string
}

type ShouldTimeout 

type ShouldTimeout interface {
	Timeout() time.Duration
}

type State 

type State string
const (
	StatePending   State = "pending"
	StateScheduled State = "scheduled"
	StateAvailable State = "available"
	StateRunning   State = "running"
	StateCompleted State = "completed"
	StateRetryable State = "retryable"
	StateFailed    State = "failed"
	StateCancelled State = "cancelled"
)

func ComputeNextState 

func ComputeNextState(job *JobRow, err error) (State, time.Duration)

func (State) Terminal 

func (s State) Terminal() bool

func (State) Valid 

func (s State) Valid() bool

type Transition 

type Transition struct {
	From State
	To   State
}

func (Transition) Valid 

func (t Transition) Valid() bool

Source Files

View all Source files

Directories

Path Synopsis
driver
redisdriver
sqldriver

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.