Unique Jobs

Sometimes you want to guarantee that the same job is not enqueued twice while a first copy is still pending or running. Classic examples:

• "Send the welcome email for user 42" -- you don't want two of those queued if the user clicks signup twice.
• "Reconcile account 1234" -- only one in flight at a time per account.

TaskRunner implements this with a distributed lock held from Dispatch until the worker finishes processing. The lock is keyed by task name (and optionally a payload-derived key), so uniqueness can be global to a task or scoped per-key.

How it works

sequenceDiagram
    participant P as Producer
    participant TR as TaskRunner
    participant LK as redsync lock<br/>(Redis SET NX PX)
    participant Q as Stream

    P->>TR: Dispatch("sendEmail", "order-123")
    TR->>TR: task.Unique == true
    TR->>LK: acquire taskrunner:unique:sendEmail:order-123<br/>TTL = UniqueFor seconds
    alt lock acquired
        LK-->>TR: ok
        TR->>Q: XADD message (with UniqueLockValue)
        Q-->>P: dispatched
    else lock held by another
        LK-->>TR: ErrLockNotObtained
        TR-->>P: ErrTaskAlreadyDispatched
    end
    Note over LK: lock auto-expires after UniqueFor seconds
    participant W as Worker
    W->>Q: XREADGROUP -> message
    W->>W: run task.Action
    W->>LK: release with stored UniqueLockValue
    LK-->>W: released (or expired)
    W->>Q: XACK

The lock is taken at dispatch time, before the XADD. The lock value (a random token) is embedded in the TaskMessage so the worker that eventually processes the message can release it. Release uses a value-guarded delete (via redsync), so only the worker holding the correct token can unlock.

Configuring a unique task

A task becomes unique by setting three fields:

tr.RegisterTask(&runner.Task{
    Name:      "sendEmail",
    MaxRetry:  3,
    Unique:    true,        // (1) enable uniqueness
    UniqueFor: 60,          // (2) lock TTL in seconds (REQUIRED when Unique)
    UniqueKey: func(payload any) string {   // (3) scope per-key (optional)
        return payload.(string)             // e.g. the email/order ID
    },
    Action: func(ctx context.Context, payload any) error {
        return sendEmail(payload.(string))
    },
})

Field	Required	Meaning
Unique	yes	Enables the lock-acquire path in Dispatch.
UniqueFor	yes (when Unique)	Lock TTL in seconds. Dispatch fails with ErrUniqueForIsRequired if this is 0.
UniqueKey	no	A func(payload any) string deriving the scope key from the payload.

The lock key

The Redis key is taskrunner:unique:<taskName>:<uniqueKey>. If UniqueKey is nil, the key collapses to just taskrunner:unique:<taskName> -- meaning only one message of that task name can be in flight at a time, regardless of payload. A warning is logged in this case.

// runner/task.go
func (t *Task) lockKey(payload any) string {
    var lockKey string
    if t.UniqueKey != nil {
        if lockKey = t.UniqueKey(payload); len(lockKey) == 0 {
            lockKey = t.Name
        }
    }
    if len(lockKey) != 0 {
        lockKey = t.Name + ":" + lockKey
    } else {
        lockKey = t.Name
        log.Warnf("The unique key for Task %s is `nil`...", t.Name)
    }
    return lockKey
}

Dispatching unique jobs

// First dispatch succeeds -- acquires the lock and XADDs.
err := tr.Dispatch(context.Background(), "sendEmail", "order-123")
// err == nil

// Second dispatch within 60s: lock is held -> ErrTaskAlreadyDispatched.
err = tr.Dispatch(context.Background(), "sendEmail", "order-123")
// err == runner.ErrTaskAlreadyDispatched

// Different key: succeeds independently.
err = tr.Dispatch(context.Background(), "sendEmail", "order-456")
// err == nil

Lifecycle of the lock

The lock is released by the worker in afterProcess -- a deferred call that runs when the worker finishes (success or failure):

func (t *TaskRunner) afterProcess(message TaskMessage) {
    if message.Unique {
        if err := t.releaseLock(message.UniqueKey, message.UniqueLockValue); err != nil {
            t.captureError(err)
        }
    }
}

If the worker crashes before releasing, the lock simply expires after UniqueFor seconds, at which point a new dispatch is allowed again. This is the intended behaviour -- uniqueness is a time-windowed guarantee, not a permanent one.

UniqueFor vs your task runtime

UniqueFor must be longer than the worst-case task runtime (including retries), otherwise the lock can expire while the first job is still processing, allowing a duplicate to slip in.

timeline
    title UniqueFor too short -> duplicate slips in
    t0 : Dispatch A : lock acquired (TTL = UniqueFor)
    t1 : Worker starts A (A is slow)
    t2 : UniqueFor expires : lock released, but A still running
    t3 : Dispatch B : lock re-acquired : B enqueued
    t4 : A finishes : B also runs : DUPLICATE WORK

Rule of thumb: set UniqueFor to comfortably exceed your task's p99 runtime multiplied by its MaxRetry. If a task can take up to 45s with up to 3 retries, set UniqueFor to at least 2-3 minutes.

Error reference

Error	When
runner.ErrUniqueForIsRequired	Dispatch of a unique task with UniqueFor == 0.
runner.ErrTaskAlreadyDispatched	Dispatch could not acquire the lock (held by another in-flight job).
runner.ErrTaskUnlockFailed	Worker's release call returned false (lock already expired or value mismatch). Treated as best-effort; the expiry is the safety net.

A note on redsync semantics

The underlying locker (internal/locker, wrapping go-redsync) implements the Distributed Mutex (Redlock) algorithm over a single Redis. The acquire uses SET NX PX with retries; the release uses a Lua value-compare-and-delete. Because TaskRunner targets a single Redis (not a multi-node Redlock quorum), the guarantees are those of a single Redis instance: correct under the Redis single-threaded model, with the usual caveats about failover during a partition.