API Reference¶

This is the complete public API. Internal packages (internal/...) are not part of the public API and must not be imported directly. The public surface spans four packages: runner, contracts, redisstream, and election.

github.com/soroosh-tanzadeh/taskrunner
├── runner        Core engine: TaskRunner, Task, config, metrics, errors
├── contracts     Interfaces & value types: MessageQueue, Message, Lock
├── redisstream   Redis Streams implementation of contracts.MessageQueue
└── election      Standalone leader election (usable without the runner)

`runner` package¶

`TaskRunner`¶

The central type. Construct with NewTaskRunner; never instantiate by hand.

func NewTaskRunner(cfg TaskRunnerConfig, client *redis.Client, queue contracts.MessageQueue) *TaskRunner

Lifecycle¶

Method	Signature	Description
`Start`	`(ctx context.Context) error`	Start fetchers, workers, election, replication, and the timing aggregator. Blocks until `ctx` is canceled; drains the worker pool (30s timeout) on exit. Returns `ErrTaskRunnerAlreadyStarted` on double-start, `ErrRaceOccuredOnStart` on a CAS race.
`StartElection`	`(ctx context.Context)`	Start leader election. Called automatically by `Start`; exposed so you can re-arm it if needed.
`StartDelayedSchedule`	`(ctx context.Context, batchSize int) error`	Start the delayed-task scheduler loop (leader-only dispatch). `batchSize` is the page size and dispatch-pool capacity. Blocks until `ctx` is canceled.

Producing work¶

Method	Signature	Description
`Dispatch`	`(ctx context.Context, taskName string, payload any) error`	Enqueue a task immediately. Looks up the task, builds a `TaskMessage`, marshals, and `XADD`s. For unique tasks, acquires the distributed lock first.
`DispatchDelayed`	`(ctx context.Context, taskName string, payload any, d time.Duration) error`	Enqueue for ~`d` in the future (min 5s). Writes to the delayed ZSET.
`ScheduleFor`	`(ctx context.Context, taskName string, payload any, executionTime time.Time) error`	Enqueue for an absolute wall-clock time (must be >5s away).
`RegisterTask`	`(task *Task)`	Add a task to the registry, keyed by `task.Name`.

Introspection¶

Method	Signature	Description
`IsLeader`	`() bool`	Whether this instance currently holds the leadership lease. Atomic.
`ConsumerGroup`	`() string`	The configured consumer group.
`GetQueue`	`() contracts.MessageQueue`	The queue backend in use.
`ErrorChannel`	`() chan error`	Buffered (cap 1) channel of runtime errors. Drain it for the lifetime of the runner; it is closed on shutdown.
`GetTimingStatistics`	`() (Stats, error)`	On-demand snapshot of per-task timings, predicted wait, TPS. Reads Redis (one `HGET` per task + `XLEN`).
`GetNumberOfReplications`	`() (int, error)`	Auto-detected instance count from replication keys (cached 10s).

`TaskRunnerConfig`¶

See Configuration Reference for prose. Field list:

type TaskRunnerConfig struct {
    Host                     string
    BatchSize                int
    ConsumerGroup            string
    ConsumersPrefix          string
    NumWorkers               int
    NumFetchers              int
    MinWorkers               int
    MaxWorkers               int
    DesiredWaitTime          time.Duration
    DesiredWaitTimeTolerance time.Duration
    TuningCooldownSeconds    int
    ReplicationFactor        int            // deprecated
    FailedTaskHandler        FailedTaskHandler
    LongQueueHook            LongQueueHook
    LongQueueThreshold       time.Duration
    BlockDuration            time.Duration
    MetricsResetInterval     time.Duration
}

`Task`¶

type Task struct {
    Name              string
    MaxRetry          int
    ReservationTimeout time.Duration
    Action            TaskAction        // func(ctx context.Context, payload any) error
    Unique            bool
    UniqueKey         UniqueKeyFunc     // func(payload any) string
    UniqueFor         int64             // seconds
}

Method:

Method	Signature	Description
`CreateMessage`	`(payload any) TaskMessage`	Build the on-the-wire message. Called internally by `Dispatch`.

Value types¶

type TaskMessage struct {
    ID                 string `json:"-"`
    Unique             bool   `json:"unique"`
    UniqueFor          int64  `json:"unique_for"`
    UniqueKey          string `json:"unique_key"`
    UniqueLockValue    string `json:"unique_lock_value"`
    ReservationTimeout int64  `json:"reservation_timeout"`
    TaskName           string `json:"task_name"`
    Payload            any    `json:"payload"`
}

type DelayedTask struct {
    Id      string `json:"id"`
    Task    string `json:"task"`
    Time    int64  `json:"time"`
    Payload any    `json:"payload"`
}

type Stats struct {
    PerTaskTiming     map[string]int64
    PredictedWaitTime float64
    AvgTiming         time.Duration
    AvgScheduleTiming int64
    TPS               float64
}

type TaskAction     func(ctx context.Context, payload any) error
type UniqueKeyFunc  func(payload any) string
type LongQueueHook  func(Stats)
type FailedTaskHandler func(ctx context.Context, task TaskMessage, err error) error

Errors¶

type TaskRunnerError string          // implements error
type TaskExecutionError struct{ ... } // wraps an Action error with task name

Sentinel constants:

Constant	Meaning
`ErrTaskRunnerAlreadyStarted`	`Start()` called twice.
`ErrRaceOccuredOnStart`	CAS race during `Start()`.
`ErrTaskNotFound`	Dispatch of an unregistered task.
`ErrInvalidTaskPayload`	Message JSON unmarshal failure.
`ErrTaskMaxRetryExceed`	Passed to `FailedTaskHandler` on exhaustion.
`ErrUniqueForIsRequired`	Unique task with `UniqueFor == 0`.
`ErrDurationIsSmallerThanCheckCycle`	Delay < 5s.
`ErrTaskAlreadyDispatched`	Unique lock contention.
`ErrTaskUnlockFailed`	Lock release returned false.

TaskExecutionError accessors:

func (t TaskExecutionError) Error() string
func (t TaskExecutionError) GetError() error
func (t TaskExecutionError) GetTaskName() string

`contracts` package¶

The interfaces a queue backend and locker must satisfy. Implement these to plug in a non-Redis backend.

type MessageQueue interface {
    Len() (int64, error)
    Add(ctx context.Context, message *Message) error
    Receive(ctx context.Context, pullDuration time.Duration, batchSize int, group, consumerName string) ([]Message, error)
    Consume(ctx context.Context, readBatchSize int, blockDuration time.Duration, group, consumerName string, errorChannel chan error, consumer StreamConsumeFunc)
    Delete(ctx context.Context, id string) error
    Purge(ctx context.Context) error
    Ack(ctx context.Context, group, messageId string) error
    RequireHeartHeartBeat() bool
    HeartBeat(ctx context.Context, group, consumer, messageId string) error
}

type Message struct {
    ID           string
    Payload      string
    receiveCount int64            // unexported; set via NewMessage
}

func NewMessage(id, payload string, receiveCount int64) Message

func (m Message) GetId() string
func (m Message) GetPayload() string
func (m Message) GetReceiveCount() int64

Locks (for the unique-jobs path):

type LockOptions struct {
    Expiry     time.Duration
    RetryDelay time.Duration
    Value      string
    Retries    int
}

type Lock interface {
    TryLock() error
    Lock() error
    Unlock() (bool, error)
    Value() string
}

type DistributedLocker interface {
    CreateMutexLock(name string, lockOptions LockOptions) Lock
}

type HeartBeatFunc     func(ctx context.Context) error
type StreamConsumeFunc func(context.Context, Message, HeartBeatFunc) error

Errors:

type QueueError string  // implements error

const (
    ErrNoNewMessage       QueueError = "no new message"
    MessageNotFoundError  QueueError = "message not found"
    NotInitializedError   QueueError = "you should initialize rmq first"
)

`redisstream` package¶

The shipped Redis Streams backend.

Constructors¶

func NewRedisStreamMessageQueue(redisClient *redis.Client, prefix, queue string, reClaimDelay time.Duration, deleteOnAck bool) *RedisStreamMessageQueue

func NewRedisStreamMessageQueueWithOptions(redisClient *redis.Client, options ...Option) *RedisStreamMessageQueue

Options¶

func WithPrefix(prefix string) Option
func WithQueue(queue string) Option
func WithReClaimDelay(reClaimDelay time.Duration) Option
func WithRedisVersion(version string) Option
func WithDeleteOnAck(deleteOnAck bool) Option

Methods (beyond `contracts.MessageQueue`)¶

Method	Signature	Description
`GetMetrics`	`(ctx context.Context) (map[string]any, error)`	Queue size, per-group pending/lag, per-consumer idle, and message-claim-delay distribution from the ring buffer.

The type implements contracts.MessageQueue in full; see the Architecture for the Redis commands behind each method.

`election` package¶

Standalone leader election, usable without the runner.

type Opts struct {
    Redis    redis.Scripter   // *redis.Client or *redis.ClusterClient
    TTL      time.Duration
    Wait     time.Duration
    JitterMS int
    Key      string
}

func NewElector(opts Opts) (leader *Elector, onPromote <-chan time.Time, onDemote <-chan time.Time, onError <-chan error)

Method	Signature	Description
`Start`	`()`	Begin competing for leadership. Idempotent.
`Stop`	`() error`	Resign (release the lease if held) and stop.
`IsLeader`	`(context.Context) error`	Returns nil if this instance currently leads, non-nil otherwise.

The three returned channels fire on leadership transitions and Redis errors. See Leader Election.

Importing¶

import (
    "github.com/redis/go-redis/v9"
    "github.com/soroosh-tanzadeh/taskrunner/contracts"
    "github.com/soroosh-tanzadeh/taskrunner/election"
    "github.com/soroosh-tanzadeh/taskrunner/redisstream"
    "github.com/soroosh-tanzadeh/taskrunner/runner"
)

Module path: github.com/soroosh-tanzadeh/taskrunner. Go 1.24.9+.

API Reference¶

runner package¶

TaskRunner¶