queue

package
v1.3.0 Latest Latest
Warning

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

 Go to latest Published: Jul 16, 2026 License: MIT Imports: 23 Imported by: 0

README

Queue Package for Nimbus

Background job processing (Laravel-inspired). Queue is core—no plugin needed. Call queue.Boot() in your app bootstrap.

Installation

Queue is initialized in bin/server.go when creating a new app. Ensure you have:

import "github.com/CodeSyncr/nimbus/queue"

queue.Boot(&queue.BootConfig{RegisterJobs: start.RegisterQueueJobs})

Configuration

Variable Description Default
QUEUE_DRIVER sync, redis, database, sqs, kafka sync
REDIS_URL Redis URL (for redis driver) redis://localhost:6379
QUEUE_REDIS_VISIBILITY_TIMEOUT_SECONDS Redis in-flight lease timeout before reclaim 60
QUEUE_DB_LEASE_SECONDS Database processing lease before reclaim 120
QUEUE_BOOT_STRICT Fail boot on unknown driver values false
SQS_QUEUE_URL AWS SQS queue URL —
KAFKA_BROKERS Kafka brokers (comma-separated) —
KAFKA_TOPIC Kafka topic nimbus-queue
KAFKA_GROUP_ID Consumer group ID nimbus-queue

Drivers:

  • sync — Runs jobs immediately (no worker). Useful for dev.
  • redis — Redis lists. Requires REDIS_URL.
  • database — GORM. Uses database.Get().
  • sqs — AWS SQS.
  • kafka — Apache Kafka.

Defining jobs

Implement the queue.Job interface:

package jobs

import (
    "context"
    "github.com/CodeSyncr/nimbus/queue"
)

type SendEmail struct {
    UserID  int
    Subject string
}

func (j *SendEmail) Handle(ctx context.Context) error {
    // Send email...
    return nil
}

Optional Failed for cleanup when job fails permanently:

func (j *SendEmail) Failed(ctx context.Context, err error) {
    log.Printf("SendEmail failed for user %d: %v", j.UserID, err)
}

Dispatching jobs

import "github.com/CodeSyncr/nimbus/queue"

queue.Dispatch(&jobs.SendEmail{UserID: 12, Subject: "Welcome"}).Dispatch(ctx)

// Delayed
queue.Dispatch(&jobs.SendEmail{...}).Delay(5 * time.Minute).Dispatch(ctx)

// Specific queue
queue.Dispatch(&jobs.Report{}).OnQueue("reports").Dispatch(ctx)

Registering jobs

In start/jobs.go (or equivalent):

package start

import (
    "github.com/CodeSyncr/nimbus/queue"
    "myapp/jobs"
)

func RegisterQueueJobs() {
    queue.Register(&jobs.SendEmail{})
    queue.Register(&jobs.ProcessVideo{})
}

Running the worker

nimbus queue:work

Or from your app:

queue.RunWorker(ctx, "default")

Rate limiting

Pass RateLimitPerSec and RateLimitBurst in BootConfig to throttle job processing.

Use this section as a starting baseline for reliable queue processing.

1) Use a durable driver in production
  • Prefer redis or database (avoid sync in production).
  • Run multiple workers (nimbus queue:work) behind a process supervisor.
  • Set QUEUE_BOOT_STRICT=true to fail fast on invalid queue driver config.
2) Tune lease / visibility timeouts
  • Redis (QUEUE_REDIS_VISIBILITY_TIMEOUT_SECONDS): start at 60.
  • Database (QUEUE_DB_LEASE_SECONDS): start at 120.
  • Rule of thumb: timeout should be at least 2x your p95 job runtime.
  • Too low: duplicate work from premature reclaim.
  • Too high: slow recovery when workers crash.

You can also set these in code:

queue.Boot(&queue.BootConfig{
    Driver:                 "redis",
    RedisURL:               "redis://localhost:6379",
    RedisVisibilityTimeout: 60 * time.Second,
    DatabaseLeaseDuration:  120 * time.Second,
    RegisterJobs:           start.RegisterQueueJobs,
})
3) Retry policy
  • Default retry backoff is exponential with jitter.
  • Keep retries bounded (Retries(n) per job).
  • Ensure job handlers are idempotent (safe to run more than once).
4) Observe the right signals

Nimbus now exports queue counters (via Horizon metrics):

  • nimbus_queue_jobs_dispatched_total
  • nimbus_queue_jobs_processed_total
  • nimbus_queue_jobs_failed_total
  • nimbus_queue_jobs_retried_total
  • nimbus_queue_jobs_reclaimed_total

Prometheus-formatted endpoint (when Horizon is enabled):

  • GET /horizon/api/metrics/prometheus
5) Suggested alert thresholds (starting point)
  • Retry spike: retried/processed ratio > 5% for 5-10 minutes.
  • Reclaim activity: reclaimed > 0 sustained for 10+ minutes.
  • Failure rate: failed/processed ratio > 1-2% for 5+ minutes.
  • Backlog growth: queue length rising continuously without recovery.

Tune thresholds per workload after collecting a week of baseline data.

Documentation

Index

Constants

This section is empty.

Variables

This section is empty.

Functions

func AcquireUniqueLock 

func AcquireUniqueLock(job UniqueJob) bool

AcquireUniqueLock acquires the unique lock for a job.

func AddObserver 

func AddObserver(o Observer)

AddObserver appends an observer (e.g. Telescope alongside Horizon).

func AfterBatch 

func AfterBatch(fn func(context.Context, *Batch))

AfterBatch registers a callback invoked after a batch finishes all jobs (after Then / Finally hooks on the batch itself). Useful for observability (e.g. Telescope). Multiple subscribers are allowed.

func DispatchUnique 

func DispatchUnique(ctx context.Context, job UniqueJob) error

DispatchUnique dispatches a job only if it's not already queued.

func IsUniqueLocked 

func IsUniqueLocked(job UniqueJob) bool

IsUniqueLocked checks if a unique job is currently locked.

func Register 

func Register(job Job)

Register registers a job type with the global manager. Call at startup, typically from a central RegisterQueueJobs function in your application. 

queue.Register(&jobs.SendEmail{})

func ReleaseUniqueLock 

func ReleaseUniqueLock(job UniqueJob)

ReleaseUniqueLock releases the unique lock for a job.

func RunWorker 

func RunWorker(ctx context.Context, queueName string)

RunWorker runs the queue worker loop. Call from queue:work command.

func SetFailedJobStore 

func SetFailedJobStore(s FailedJobStore)

SetFailedJobStore sets the global failed job store (e.g. from Horizon plugin).

func SetGlobal 

func SetGlobal(m *Manager)

SetGlobal sets the global manager.

func SetObserver 

func SetObserver(o Observer)

SetObserver replaces the observer list with a single observer (Horizon).

Types

type Adapter 

type Adapter interface {
	// Push adds a job to the queue. delay is 0 for immediate.
	Push(ctx context.Context, payload *JobPayload) error

	// Pop blocks until a job is available or ctx is done. Returns nil when done.
	Pop(ctx context.Context, queue string) (*JobPayload, error)

	// Len returns approximate number of pending jobs (best-effort).
	Len(ctx context.Context, queue string) (int, error)
}

Adapter enqueues and dequeues jobs for processing.

type Batch 

type Batch struct {
	ID string
	// contains filtered or unexported fields
}

Batch allows dispatching a group of jobs and tracking their completion, with optional callbacks.

func NewBatch 

func NewBatch(jobs ...Job) *Batch

NewBatch creates a new job batch.

func (*Batch) Catch 

func (b *Batch) Catch(fn func(ctx context.Context, batch *Batch, err error)) *Batch

Catch sets a callback for when any job in the batch fails.

func (*Batch) Dispatch 

func (b *Batch) Dispatch(ctx context.Context) error

Dispatch runs all jobs in the batch concurrently.

func (*Batch) Errors 

func (b *Batch) Errors() []error

Errors returns all errors from failed jobs.

func (*Batch) FailedJobs 

func (b *Batch) FailedJobs() int32

FailedJobs returns the number of failed jobs.

func (*Batch) Finally 

func (b *Batch) Finally(fn func(ctx context.Context, batch *Batch)) *Batch

Finally sets a callback that runs after all jobs complete (success or failure).

func (*Batch) Finished 

func (b *Batch) Finished() bool

Finished returns true when all jobs have completed.

func (*Batch) HasFailures 

func (b *Batch) HasFailures() bool

HasFailures returns true if any jobs failed.

func (*Batch) OnQueue 

func (b *Batch) OnQueue(name string) *Batch

OnQueue sets the queue for all jobs in the batch.

func (*Batch) PendingJobs 

func (b *Batch) PendingJobs() int32

PendingJobs returns the number of jobs still pending.

func (*Batch) Then 

func (b *Batch) Then(fn func(ctx context.Context, batch *Batch)) *Batch

Then sets a callback for when all non-failed jobs complete.

func (*Batch) TotalJobs 

func (b *Batch) TotalJobs() int32

TotalJobs returns the total number of jobs in the batch.

type BootConfig 

type BootConfig struct {
	Driver   string // sync, redis, database, sqs, kafka
	RedisURL string
	// RedisVisibilityTimeout controls Redis in-flight lease timeout.
	RedisVisibilityTimeout time.Duration
	// DatabaseLeaseDuration controls how long processing DB jobs are leased before reclaim.
	DatabaseLeaseDuration time.Duration
	SQSQueueURL           string
	KafkaBrokers          string
	KafkaTopic            string
	KafkaGroupID          string
	RateLimitPerSec       float64
	RateLimitBurst        int
	Strict                bool
	RegisterJobs          func()
}

BootConfig configures queue boot. Pass nil for env-based config.

type Chain 

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

Chain dispatches a sequence of jobs one after another. If any job in the chain fails (exhausts retries), the rest are skipped and the optional OnFailure callback is called.

func NewChain 

func NewChain(jobs ...Job) *Chain

NewChain creates a new job chain.

func (*Chain) Dispatch 

func (c *Chain) Dispatch(ctx context.Context) error

Dispatch executes the chain sequentially.

func (*Chain) DispatchAsync 

func (c *Chain) DispatchAsync(ctx context.Context) error

DispatchAsync dispatches the chain as a single wrapper job via queue.

func (*Chain) OnFailure 

func (c *Chain) OnFailure(fn func(ctx context.Context, failedJob Job, err error)) *Chain

OnFailure sets a callback when a chain job fails.

func (*Chain) OnQueue 

func (c *Chain) OnQueue(name string) *Chain

OnQueue sets the queue for all jobs in the chain.

type CompletableAdapter 

type CompletableAdapter interface {
	Adapter
	Complete(ctx context.Context, payload *JobPayload) error
}

CompletableAdapter optionally deletes/acks a message after successful processing (e.g. SQS).

type DatabaseAdapter 

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

DatabaseAdapter uses a SQL database for job storage.

func NewDatabaseAdapter 

func NewDatabaseAdapter(db *lucid.DB) *DatabaseAdapter

NewDatabaseAdapter creates a database adapter.

func (*DatabaseAdapter) Complete 

func (d *DatabaseAdapter) Complete(ctx context.Context, payload *JobPayload) error

Complete marks a processing job as done after successful or terminal handling.

func (*DatabaseAdapter) EnsureTable 

func (d *DatabaseAdapter) EnsureTable(ctx context.Context) error

EnsureTable creates the queue_jobs table if not exists.

func (*DatabaseAdapter) Len 

func (d *DatabaseAdapter) Len(ctx context.Context, queue string) (int, error)

Len returns the number of pending jobs.

func (*DatabaseAdapter) Pop 

func (d *DatabaseAdapter) Pop(ctx context.Context, queue string) (*JobPayload, error)

Pop blocks until a job is available. Uses polling with SELECT FOR UPDATE SKIP LOCKED (Postgres/MySQL).

func (*DatabaseAdapter) Push 

func (d *DatabaseAdapter) Push(ctx context.Context, payload *JobPayload) error

Push adds a job to the queue.

func (*DatabaseAdapter) SetLeaseDuration 

func (d *DatabaseAdapter) SetLeaseDuration(v time.Duration)

SetLeaseDuration sets how long a processing job can remain unacked before reclaim.

type DispatchBuilder 

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

DispatchBuilder allows chaining dispatch options.

func Dispatch 

func Dispatch(job Job) *DispatchBuilder

Dispatch enqueues a job using the global manager. It is the primary entry point for application code to push work onto a queue.

The returned DispatchBuilder can be used to set options before the job is actually queued: 

queue.Dispatch(&jobs.SendEmail{UserID: 12}).
    Delay(5 * time.Minute).
    Dispatch(ctx)

queue.Dispatch(&jobs.Report{}).
    OnQueue("reports").
    Retries(5).
    Dispatch(ctx)

If no global manager has been configured via queue.SetGlobal (usually done by queue.Boot), Dispatch returns a no-op builder so that calls are safe in tests or environments where the queue subsystem is disabled.

func (*DispatchBuilder) Delay 

func (b *DispatchBuilder) Delay(d time.Duration) *DispatchBuilder

Delay sets the delay before the job runs.

func (*DispatchBuilder) Dispatch 

func (b *DispatchBuilder) Dispatch(ctx context.Context) error

Dispatch executes the dispatch.

func (*DispatchBuilder) OnQueue 

func (b *DispatchBuilder) OnQueue(name string) *DispatchBuilder

OnQueue sets the queue name.

func (*DispatchBuilder) Priority 

func (b *DispatchBuilder) Priority(n int) *DispatchBuilder

Priority sets job priority (1=highest, 10=lowest).

func (*DispatchBuilder) Retries 

func (b *DispatchBuilder) Retries(n int) *DispatchBuilder

Retries sets max retry attempts.

type FailedJob 

type FailedJob interface {
	Job
	Failed(ctx context.Context, err error)
}

FailedJob is optionally implemented for cleanup when job fails permanently.

type FailedJobRecord 

type FailedJobRecord struct {
	ID         string    `json:"id"`
	UUID       string    `json:"uuid"`
	Queue      string    `json:"queue"`
	JobName    string    `json:"job_name"`
	Payload    []byte    `json:"payload"`
	Exception  string    `json:"exception"`
	FailedAt   time.Time `json:"failed_at"`
	Attempts   int       `json:"attempts"`
	MaxRetries int       `json:"max_retries"`
}

FailedJobRecord is a single failed job entry for listing and retry.

type FailedJobStore 

type FailedJobStore interface {
	// Push adds a failed job to the store.
	Push(ctx context.Context, payload *JobPayload, errMsg string) error
	// List returns all failed job records (e.g. for dashboard).
	List(ctx context.Context) ([]FailedJobRecord, error)
	// Get returns one record by ID/UUID.
	Get(ctx context.Context, id string) (*FailedJobRecord, error)
	// Forget removes a single failed job.
	Forget(ctx context.Context, id string) error
	// ForgetAll removes all failed jobs.
	ForgetAll(ctx context.Context) error
	// Retry re-enqueues the job and removes it from failed store.
	Retry(ctx context.Context, id string, enqueue func(ctx context.Context, payload *JobPayload) error) error
}

FailedJobStore persists failed jobs for Horizon dashboard (list, forget, retry).

func GetFailedJobStore 

func GetFailedJobStore() FailedJobStore

GetFailedJobStore returns the global failed job store.

type Job 

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

Job is the interface for queue jobs.

type JobFunc 

type JobFunc func(ctx context.Context) error

JobFunc adapts a function to Job.

func (JobFunc) Handle 

func (f JobFunc) Handle(ctx context.Context) error

type JobPayload 

type JobPayload struct {
	ID         string                 `json:"id"`
	JobName    string                 `json:"job"`
	Queue      string                 `json:"queue"`
	Payload    []byte                 `json:"payload"`
	Attempts   int                    `json:"attempts"`
	MaxRetries int                    `json:"max_retries"`
	Delay      time.Duration          `json:"delay"`
	RunAt      time.Time              `json:"run_at"`
	Meta       map[string]interface{} `json:"meta,omitempty"`
}

JobPayload is the serialized form of a job for storage.

type KafkaAdapter 

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

KafkaAdapter uses Kafka for job storage.

func NewKafkaAdapter 

func NewKafkaAdapter(cfg KafkaConfig) *KafkaAdapter

NewKafkaAdapter creates a Kafka adapter.

func (*KafkaAdapter) Close 

func (k *KafkaAdapter) Close() error

Close closes the writer and reader.

func (*KafkaAdapter) Len 

func (k *KafkaAdapter) Len(ctx context.Context, queue string) (int, error)

Len returns 0 (Kafka doesn't provide simple count).

func (*KafkaAdapter) Pop 

func (k *KafkaAdapter) Pop(ctx context.Context, queue string) (*JobPayload, error)

Pop blocks until a job is available.

func (*KafkaAdapter) Push 

func (k *KafkaAdapter) Push(ctx context.Context, payload *JobPayload) error

Push adds a job to the queue.

type KafkaConfig 

type KafkaConfig struct {
	Brokers  []string // e.g. []string{"localhost:9092"}
	Topic    string   // queue topic
	GroupID  string   // consumer group for workers
	MinBytes int      // min bytes to fetch (default 1)
	MaxBytes int      // max bytes to fetch (default 1e6)
}

KafkaConfig holds Kafka adapter configuration.

type Manager 

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

Manager manages adapters and job dispatch.

func Boot 

func Boot(cfg *BootConfig) *Manager

Boot initializes the queue manager from config/env and sets it globally.

func BootWithError 

func BootWithError(cfg *BootConfig) (*Manager, error)

BootWithError initializes the queue manager from config/env and returns an explicit error when configuration/adapter setup fails.

func GetGlobal 

func GetGlobal() *Manager

GetGlobal returns the global manager.

func NewManager 

func NewManager(adapter Adapter) *Manager

NewManager creates a manager with the given adapter. Pass nil to use SyncAdapter.

func (*Manager) Adapter 

func (m *Manager) Adapter() Adapter

Adapter returns the underlying queue adapter (for Horizon retry, etc.).

func (*Manager) Dispatch 

func (m *Manager) Dispatch(job Job) *DispatchBuilder

Dispatch enqueues a job. Returns a DispatchBuilder for options.

func (*Manager) Process 

func (m *Manager) Process(ctx context.Context, queue string) error

Process pops a job from the adapter, deserializes, and runs it.

func (*Manager) Register 

func (m *Manager) Register(job Job)

Register registers a job type for deserialization. Call with a zero-value instance. 

queue.Register(&jobs.SendEmail{})

func (*Manager) RegisterFunc 

func (m *Manager) RegisterFunc(name string, fn func() Job)

RegisterFunc registers a job by name with a constructor.

type Observer 

type Observer interface {
	JobDispatched(payload *JobPayload)
	JobProcessed(payload *JobPayload, err error)
}

Observer can be used to observe queue lifecycle events (for dashboards like Horizon). It is optional and only called when set.

type ObserverV2 

type ObserverV2 interface {
	JobProcessedV2(payload *JobPayload, duration time.Duration, err error)
}

ObserverV2 is an optional extension interface for richer job lifecycle metadata. Observers can implement this in addition to Observer.

type Queue 

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

Queue is a legacy in-memory worker pool. Prefer queue.Dispatch() with a configured Manager (Redis/Database) for production.

func NewQueue 

func NewQueue(workers int) *Queue

NewQueue creates a queue with n workers. Start with Run().

func (*Queue) Push 

func (q *Queue) Push(job Job)

Push enqueues a job (non-blocking if buffer full; can backpressure).

func (*Queue) Run 

func (q *Queue) Run(ctx context.Context)

Run starts the worker pool. Call in a goroutine or block.

func (*Queue) Stop 

func (q *Queue) Stop()

Stop stops the queue.

type QueueJob 

type QueueJob struct {
	ID        string    `gorm:"primaryKey;size:36"`
	Queue     string    `gorm:"index;size:64;not null"`
	Payload   []byte    `gorm:"type:text;not null"` // JSON of JobPayload
	RunAt     time.Time `gorm:"index;not null"`
	Status    string    `gorm:"size:16;default:pending"` // pending, processing, failed, done
	CreatedAt time.Time
	UpdatedAt time.Time
}

QueueJob is the database model for jobs.

func (QueueJob) TableName 

func (QueueJob) TableName() string

type RateLimitAdapter 

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

RateLimitAdapter wraps an adapter with rate limiting.

func NewRateLimitAdapter 

func NewRateLimitAdapter(inner Adapter, limitPerSec float64, burst int) *RateLimitAdapter

NewRateLimitAdapter wraps an adapter with a rate limiter. limitPerSec: max jobs per second (e.g. 10) burst: max burst size (e.g. 20)

func (*RateLimitAdapter) Complete 

func (r *RateLimitAdapter) Complete(ctx context.Context, payload *JobPayload) error

Complete delegates if inner supports it.

func (*RateLimitAdapter) Len 

func (r *RateLimitAdapter) Len(ctx context.Context, queue string) (int, error)

Len delegates to inner.

func (*RateLimitAdapter) Pop 

func (r *RateLimitAdapter) Pop(ctx context.Context, queue string) (*JobPayload, error)

Pop waits for rate limit before returning a job.

func (*RateLimitAdapter) Push 

func (r *RateLimitAdapter) Push(ctx context.Context, payload *JobPayload) error

Push delegates to inner (no rate limit on push).

type RateLimitConfig 

type RateLimitConfig struct {
	// Limit is the max number of jobs per second (e.g. 10 = 10 jobs/sec).
	Limit float64
	// Burst allows short bursts above the limit.
	Burst int
}

RateLimitConfig configures rate limiting per queue.

type ReclaimObserver 

type ReclaimObserver interface {
	JobsReclaimed(queue string, count int)
}

ReclaimObserver can observe reclaimed in-flight jobs.

type RedisAdapter 

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

RedisAdapter uses Redis lists for job storage. Supports delayed jobs via sorted sets.

func NewRedisAdapter 

func NewRedisAdapter(client *redis.Client) *RedisAdapter

NewRedisAdapter creates a Redis adapter. Pass a configured redis.Client.

func NewRedisAdapterFromURL 

func NewRedisAdapterFromURL(url string) (*RedisAdapter, error)

NewRedisAdapterFromURL creates adapter from REDIS_URL (e.g. redis://localhost:6379).

func (*RedisAdapter) Complete 

func (r *RedisAdapter) Complete(ctx context.Context, payload *JobPayload) error

Complete acknowledges and removes a processed in-flight Redis message.

func (*RedisAdapter) Len 

func (r *RedisAdapter) Len(ctx context.Context, queue string) (int, error)

Len returns the number of pending jobs.

func (*RedisAdapter) Pop 

func (r *RedisAdapter) Pop(ctx context.Context, queue string) (*JobPayload, error)

Pop blocks until a job is available. Processes delayed jobs when ready.

func (*RedisAdapter) Push 

func (r *RedisAdapter) Push(ctx context.Context, payload *JobPayload) error

Push adds a job to the queue.

func (*RedisAdapter) SetVisibilityTimeout 

func (r *RedisAdapter) SetVisibilityTimeout(v time.Duration)

SetVisibilityTimeout sets the in-flight lease before jobs are reclaimed.

type RedisFailedStore 

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

RedisFailedStore stores failed jobs in Redis for Horizon dashboard.

func NewRedisFailedStore 

func NewRedisFailedStore(client *redis.Client) *RedisFailedStore

NewRedisFailedStore creates a failed job store using the given Redis client.

func (*RedisFailedStore) Forget 

func (r *RedisFailedStore) Forget(ctx context.Context, id string) error

Forget removes a single failed job.

func (*RedisFailedStore) ForgetAll 

func (r *RedisFailedStore) ForgetAll(ctx context.Context) error

ForgetAll removes all failed jobs.

func (*RedisFailedStore) Get 

func (r *RedisFailedStore) Get(ctx context.Context, id string) (*FailedJobRecord, error)

Get returns one record by ID.

func (*RedisFailedStore) List 

func (r *RedisFailedStore) List(ctx context.Context) ([]FailedJobRecord, error)

List returns all failed job records (newest last).

func (*RedisFailedStore) Push 

func (r *RedisFailedStore) Push(ctx context.Context, payload *JobPayload, errMsg string) error

Push adds a failed job to the store.

func (*RedisFailedStore) Retry 

func (r *RedisFailedStore) Retry(ctx context.Context, id string, enqueue func(ctx context.Context, payload *JobPayload) error) error

Retry re-enqueues the job and removes it from the failed store.

type RedisQueueWorkload 

type RedisQueueWorkload struct {
	Name       string `json:"name"`
	Pending    int64  `json:"pending"`    // jobs waiting in the main list
	Delayed    int64  `json:"delayed"`    // jobs in delayed sorted set
	Processing int64  `json:"processing"` // jobs leased to workers
	InFlight   int64  `json:"in_flight"`  // in-flight lease tracking (sorted set members)
}

RedisQueueWorkload holds Redis list/set sizes for one logical queue (Redis driver key layout).

func RedisQueueWorkloads 

func RedisQueueWorkloads(ctx context.Context, c *redis.Client, names []string) ([]RedisQueueWorkload, error)

RedisQueueWorkloads returns live depth metrics from Redis for the given queue names. Uses the same key prefixes as RedisAdapter. Pass nil or empty names to probe only "default".

type RetryObserver 

type RetryObserver interface {
	JobRetried(payload *JobPayload, nextDelay time.Duration)
}

RetryObserver can observe retry scheduling events.

type SQSAdapter 

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

SQSAdapter uses AWS SQS for job storage.

func NewSQSAdapter 

func NewSQSAdapter(ctx context.Context, queueURL string) (*SQSAdapter, error)

NewSQSAdapter creates an SQS adapter. queueURL is the full SQS queue URL.

func NewSQSAdapterFromConfig 

func NewSQSAdapterFromConfig(client *sqs.Client, queueURL string) *SQSAdapter

NewSQSAdapterFromConfig creates adapter with custom config.

func (*SQSAdapter) Complete 

func (s *SQSAdapter) Complete(ctx context.Context, payload *JobPayload) error

Complete deletes the SQS message after successful processing.

func (*SQSAdapter) Len 

func (s *SQSAdapter) Len(ctx context.Context, queue string) (int, error)

Len returns approximate message count (SQS ApproximateNumberOfMessages).

func (*SQSAdapter) Pop 

func (s *SQSAdapter) Pop(ctx context.Context, queue string) (*JobPayload, error)

Pop blocks until a job is available (long polling).

func (*SQSAdapter) Push 

func (s *SQSAdapter) Push(ctx context.Context, payload *JobPayload) error

Push adds a job to the queue.

type Scheduler 

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

Scheduler runs jobs on a schedule.

func NewScheduler 

func NewScheduler(m *Manager) *Scheduler

NewScheduler creates a scheduler that dispatches to the given manager.

func (*Scheduler) Cron 

func (s *Scheduler) Cron(expr string, job Job) error

Cron adds a job with a cron expression. Examples: "0 0 * * *" = midnight daily, "*/5 * * * *" = every 5 min.

func (*Scheduler) Every 

func (s *Scheduler) Every(d time.Duration, job Job) error

Every adds a job that runs at a fixed interval.

func (*Scheduler) Start 

func (s *Scheduler) Start(ctx context.Context)

Start runs the scheduler. Blocks until ctx is done.

func (*Scheduler) Stop 

func (s *Scheduler) Stop()

Stop stops the scheduler.

type Silenced 

type Silenced interface {
	Job
	// Silenced returns true to hide this job from the completed list.
	Silenced() bool
}

Silenced is optionally implemented to hide the job from Horizon's completed jobs list.

type SyncAdapter 

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

SyncAdapter runs jobs immediately in the same process. No persistence. Use for development/testing when you don't need a separate worker.

func NewSyncAdapter 

func NewSyncAdapter(m *Manager) *SyncAdapter

NewSyncAdapter creates a sync adapter. Requires manager for job deserialization.

func (*SyncAdapter) Len 

func (s *SyncAdapter) Len(ctx context.Context, queue string) (int, error)

Len always returns 0.

func (*SyncAdapter) Pop 

func (s *SyncAdapter) Pop(ctx context.Context, queue string) (*JobPayload, error)

Pop always blocks (sync has no persisted jobs). Use Redis/Database for workers.

func (*SyncAdapter) Push 

func (s *SyncAdapter) Push(ctx context.Context, payload *JobPayload) error

Push runs the job immediately in-process. It mirrors the behavior of Manager.Process enough to keep observers (e.g. Horizon) informed.

type Tagger 

type Tagger interface {
	Job
	Tags() []string
}

Tagger is optionally implemented to assign tags for Horizon dashboard (Laravel-style).

type UniqueJob 

type UniqueJob interface {
	Job

	// UniqueID returns a unique identifier for this job instance.
	// Jobs with the same UniqueID will not be dispatched while one is pending.
	UniqueID() string

	// UniqueFor returns the duration to hold the unique lock.
	// After this duration, the same job can be dispatched again.
	UniqueFor() time.Duration
}

UniqueJob interface — jobs implementing this will be deduplicated.

type WithoutOverlapping 

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

WithoutOverlapping wraps a Job so that concurrent executions of the same job (by UniqueID) are prevented. Only one instance runs at a time.

func NewWithoutOverlapping 

func NewWithoutOverlapping(job Job, uniqueID string) *WithoutOverlapping

NewWithoutOverlapping wraps a job to prevent overlapping execution.

func (*WithoutOverlapping) Handle 

func (w *WithoutOverlapping) Handle(ctx context.Context) error

Handle runs the inner job if no other instance with the same ID is running.

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.