tasker

package module
v0.3.3 Latest Latest
Warning

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

 Go to latest Published: Nov 2, 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 }

Watching enqueues and scheduled runs

AddGlobalMiddleware wraps a job's execution. Two other things are observable, and they are different questions:

// Every enqueue, process-wide — Dispatch, DispatchBatch and Chain.
tasker.AddDispatchMiddleware(func(ctx context.Context, rows []*tasker.JobRow,
    next func(context.Context) error) error {
    for _, row := range rows {
        row.Metadata["request_id"] = requestIDFrom(ctx) // stamp before sending
    }
    return next(ctx)
})

// Every scheduled firing.
sched.Observe(func(ctx context.Context, run scheduler.Run) {
    slog.InfoContext(ctx, "scheduled", "id", run.ID, "err", run.Err)
})

A dispatch middleware is handed the rows before they are sent, and writing to them is allowed: Metadata is how a job carries the id of the request that queued it, which is what makes an enqueue here and an execution in a worker process the same story. Returning without calling next refuses the enqueue and the error reaches the caller of Dispatch, so a middleware meant only to watch must call next and return its error unchanged. Index 0 is outermost, as with JobMiddleware.

The two halves meet through the row. tasker.RowFromContext(ctx) returns the row being executed, installed by the worker before any middleware runs — so the id a dispatch middleware wrote into Metadata is readable in the process that runs the job. JobMiddleware is handed the decoded job and the raw payload, neither of which carries Metadata, and the two halves happen in different processes, so the row is the only thing that travels.

Without it, enqueue is invisible: execution middleware only sees a job run, so a job that was queued and never picked up is indistinguishable from a job nobody queued — which is the difference between "no worker is running" and "the dispatch call is not being reached".

scheduler.Run is reported separately because a job row answers nothing a schedule is asked: whether 4:20 fired, whether it fired twice, how long it has been failing. And a run that failed to enqueue leaves no row at all, so without the listener it did not happen as far as the queue is concerned.

Both contain a listener's panic. The scheduler's matters most: its cron has no recovery chain, so a panic on the firing goroutine would take down the one process that runs every scheduled job in the deployment.

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

View Source 
const NodeLivenessWindow = 30 * time.Second

NodeLivenessWindow is how recently a node must have reported to count as alive.

One constant because two answers to "is this worker up" is a dashboard contradicting itself: ListNodes filtered on a locally-written 30 seconds while GlobalStats.Processes counted DISTINCT node_id over *running jobs* — so an idle worker, which is what a healthy worker is almost all the time, made the overview say "no worker is reporting" directly above a table listing that worker with a heartbeat seconds old.

Three times the supervisor's default heartbeat interval, so one missed beat is not an outage.

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 AddDispatchMiddleware added in v0.2.0 

func AddDispatchMiddleware(mw DispatchMiddleware)

AddDispatchMiddleware installs a middleware around every enqueue, for the whole process. Additive, like AddGlobalMiddleware.

Process-wide rather than per-Manager, matching the execution side, because an observer wants every enqueue including the ones made through tasker.Global(). The cost is that it is global state a test can leave behind, which is what SetDispatchMiddlewares(nil) is for.

func AddGlobalMiddleware 

func AddGlobalMiddleware(mw JobMiddleware)

func ContextWithRow added in v0.2.1 

func ContextWithRow(ctx context.Context, row *JobRow) context.Context

ContextWithRow puts the row being executed on the context.

The worker calls this before it runs any middleware, so a JobMiddleware and a job's own Handle can reach the row. Exported because a caller that executes a job itself — a test, a one-off runner — should be able to put the same thing there.

func ListRegisteredJobs 

func ListRegisteredJobs() []string

func RegisterJob 

func RegisterJob(name string, factory JobFactory)

func SetDispatchMiddlewares added in v0.2.0 

func SetDispatchMiddlewares(mws []DispatchMiddleware)

SetDispatchMiddlewares replaces the installed set. Passing nil clears it, which is how a test stops leaking into the next one.

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 DispatchMiddleware added in v0.2.0 

type DispatchMiddleware func(ctx context.Context, rows []*JobRow, next func(ctx context.Context) error) error

DispatchMiddleware wraps an enqueue, the way JobMiddleware wraps an execution.

Until this existed, enqueue was not observable at all: AddGlobalMiddleware sees a job run, so a job that was queued and never picked up was indistinguishable from a job nobody queued — which is the difference between "no worker is running" and "the dispatch call is not being reached", and the most common question asked about a queue.

The rows are handed over before they are sent, and writing to them is allowed. Metadata is how a job carries the id of the request that queued it, which is what makes an enqueue and its execution in another process the same story; a hook that only reported the outcome could not put it there.

Returning an error without calling next refuses the enqueue, and the error reaches the caller of Dispatch. That is real power, so it is written down: a middleware meant only to watch must call next and return its error unchanged.

func GetDispatchMiddlewares added in v0.2.0 

func GetDispatchMiddlewares() []DispatchMiddleware

GetDispatchMiddlewares returns a copy of the installed set.

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"`
}

func RowFromContext added in v0.2.1 

func RowFromContext(ctx context.Context) (*JobRow, bool)

RowFromContext returns the row of the job being executed.

This is what closes the gap between an enqueue and an execution. A DispatchMiddleware can write to a row's Metadata — the id of the request that queued the job, say — and until now nothing on the execution side could read it back: JobMiddleware is handed the decoded job and the raw payload, neither of which carries Metadata. The two halves happen in different processes, so the row is the only thing that travels between them.

The row is the worker's own, not a copy. Read it; writing to it races the worker's own state transitions.

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

WithNodeID sets the manager's node identity.

An empty id is ignored rather than applied, and that is the whole point. The node id is the owner token for the leader lease, for every mutual-exclusion lock and for job claims — so "" is not a weaker identity, it is the *same* identity in every process. The compare-and-renew script then reads owner == "" as "this is mine" for everybody: every process wins the leader lease at once, every process can release every other's locks, and the whole deployment shares one heartbeat key, which the first shutdown deletes for all of them. After that each survivor's heartbeat fails forever with "node not found", because registration only happens at boot.

NewConfiguredManager passes whatever a caller's Config carries, and a Config assembled field by field carries "" unless someone remembered this one — which is how a real deployment arrived here. DefaultConfig already generates a unique id; the fix is to stop overwriting it with nothing.

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.