Example: Delayed Scheduler

This example schedules tasks to run in the future. The full source is at examples/scheduler.

package main

import (
    "context"
    "fmt"
    "sync"
    "time"

    "github.com/redis/go-redis/v9"
    "github.com/soroosh-tanzadeh/taskrunner/redisstream"
    "github.com/soroosh-tanzadeh/taskrunner/runner"
)

func main() {
    rdb := redis.NewClient(&redis.Options{Addr: "127.0.0.1:6379", DB: 0, PoolSize: 50})
    queue := redisstream.NewRedisStreamMessageQueueWithOptions(
        rdb,
        redisstream.WithPrefix("example"),
        redisstream.WithQueue("tasks"),
        redisstream.WithReClaimDelay(30*time.Second),
        redisstream.WithDeleteOnAck(true),
    )

    tr := runner.NewTaskRunner(runner.TaskRunnerConfig{
        BatchSize:       10,
        ConsumerGroup:   "example",
        ConsumersPrefix: "default",
        NumWorkers:      8,
        NumFetchers:     2,
    }, rdb, queue)

    tr.RegisterTask(&runner.Task{
        Name:     "runLater",
        MaxRetry: 3,
        Action: func(ctx context.Context, payload any) error {
            fmt.Printf("executed: %v\n", payload)
            return nil
        },
    })

    ctx, cancel := context.WithCancel(context.Background())
    defer cancel()
    wg := sync.WaitGroup{}
    wg.Add(2)
    go func() { defer wg.Done(); _ = tr.Start(ctx) }()                   // (1) workers + election
    go func() { defer wg.Done(); _ = tr.StartDelayedSchedule(ctx, 1000) }() // (2) scheduler

    _ = tr.DispatchDelayed(context.Background(), "runLater", "I run in ~5s", 5*time.Second)
    _ = tr.DispatchDelayed(context.Background(), "runLater", "I run in ~10s", 10*time.Second)

    time.Sleep(12 * time.Second)
    cancel()
    wg.Wait()
}

Walkthrough

Start two loops

go func() { _ = tr.Start(ctx) }()                      // workers + election
go func() { _ = tr.StartDelayedSchedule(ctx, 1000) }() // scheduler

• Start runs the fetchers, workers, replication registrar, timing aggregator, and leader election.
• StartDelayedSchedule(ctx, batchSize) runs the leader-polled scheduler loop. The batchSize (here 1000) is both the page size for due jobs and the capacity of the scheduler's internal dispatch pool.

Only the leader actually dispatches due jobs, but every instance should call StartDelayedSchedule so that a leadership failover picks up the scheduler without intervention.

Schedule with a relative delay

_ = tr.DispatchDelayed(context.Background(), "runLater", "I run in ~5s", 5*time.Second)

This ZADDs a DelayedTask into taskrunner:<group>:delayed_tasks with score = time.Now().Add(5s).Unix(). The minimum effective delay is 5s; shorter delays return ErrDurationIsSmallerThanCheckCycle.

Or schedule for an absolute time

when := time.Now().Add(1 * time.Hour)
_ = tr.ScheduleFor(context.Background(), "runLater", "hourly", when)

ScheduleFor is the absolute-time counterpart to DispatchDelayed's relative duration.

What the leader does every 5 seconds

sequenceDiagram
    participant L as Leader
    participant Z as delayed_tasks ZSET
    participant S as Stream
    L->>Z: ZCOUNT score in [0, now]
    Z-->>L: N due
    L->>Z: ZRANGEBYSCORE (paged, batchSize)
    Z-->>L: due DelayedTask JSON
    loop each due job
        L->>S: Dispatch -> XADD
        L->>Z: ZREM member
    end

Because ZREM happens after XADD, a crash between the two means a job could be dispatched twice on recovery. Keep your Action idempotent.

Run it

go run examples/scheduler/main.go
# executed: I run in ~5s        (~5s in)
# executed: I run in ~10s       (~10s in)

The test (examples/scheduler/scheduler_test.go) uses miniredis:

go test ./examples/scheduler/...

When to reach for delayed vs. retry

• Delayed = you know at dispatch time the job should run later (reminders, scheduled maintenance, backoff windows you compute yourself).
• Retry (via MaxRetry + the pending reclaim cycle) = the job failed and should be redelivered automatically. See Delayed Scheduling.