Example: Unique Jobs

This example shows unique-task dispatch: the second attempt to enqueue the same job (within the UniqueFor window) is rejected. The full source is at examples/unique.

package main

import (
    "context"
    "fmt"
    "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:       5,
        ConsumerGroup:   "example-unique",
        ConsumersPrefix: "default",
        NumWorkers:      2,
        NumFetchers:     1,
    }, rdb, queue)

    tr.RegisterTask(&runner.Task{
        Name:      "sendEmail",
        MaxRetry:  1,
        Unique:    true,            // (1) enable uniqueness
        UniqueFor: 60,              // (2) lock TTL, seconds
        UniqueKey: func(payload any) string { return payload.(string) }, // (3) per-key scope
        Action: func(ctx context.Context, payload any) error {
            fmt.Printf("sending email: %s\n", payload)
            return nil
        },
    })

    ctx, cancel := context.WithCancel(context.Background())
    defer cancel()
    go tr.Start(ctx)

    _ = tr.Dispatch(context.Background(), "sendEmail", "order-123")
    if err := tr.Dispatch(context.Background(), "sendEmail", "order-123"); err != nil {
        fmt.Printf("duplicate blocked: %v\n", err)
    }

    time.Sleep(2 * time.Second)
    cancel()
}

Expected output:

sending email: order-123
duplicate blocked: ErrTaskAlreadyDispatched

Walkthrough

Make the task unique

Three fields turn a normal task into a unique one:

Unique:    true,
UniqueFor: 60,                                               // seconds
UniqueKey: func(payload any) string { return payload.(string) },

Field	Effect
Unique: true	Dispatch acquires a distributed lock before XADD.
UniqueFor: 60	The lock's TTL in seconds. Required when Unique is true.
UniqueKey	Derives the lock scope from the payload. Same key => same lock => second dispatch is blocked.

The lock key

For payload "order-123" and task name "sendEmail", the Redis lock key is:

taskrunner:unique:sendEmail:order-123

If UniqueKey returns "" or is nil, the key collapses to taskrunner:unique:sendEmail -- meaning only one sendEmail of any kind can be in flight at a time. The example scopes per-order so different orders proceed independently.

Dispatch behavior

_ = tr.Dispatch(context.Background(), "sendEmail", "order-123") // acquires lock, XADDs -> nil
err := tr.Dispatch(context.Background(), "sendEmail", "order-123")
// err == runner.ErrTaskAlreadyDispatched

The lock is held from Dispatch until the worker's afterProcess deferred call runs. If the worker crashes, the lock expires after UniqueFor seconds and a new dispatch becomes possible.

Run it

go run examples/unique/main.go

The test (examples/unique/unique_test.go) uses miniredis:

go test ./examples/unique/...

Practical guidance

• Set UniqueFor above worst-case runtime (including retries). If the task can take 45s, set UniqueFor >= 90. Otherwise the lock expires mid-run and a duplicate slips in. See Unique Jobs.
• Make UniqueKey deterministic. It must return the same string for payloads you consider "the same job". For a struct payload, hash the identity fields.
• Handle ErrTaskAlreadyDispatched gracefully -- it's expected, not an error. Log it and move on; don't retry in a tight loop.
• Nil UniqueKey => global uniqueness for that task. Useful for "only one nightly reconciliation in flight", but it serializes all dispatches of that task behind a single lock, so it limits throughput.