Unique jobs

Jobs can be made unique, such that River guarantees that only one exists for a given set of properties. Jobs can be made unique by args, kind, period, queue, and state.

Unique properties

It’s occasionally useful to ensure that background work is only performed once for a given set of conditions. For example, there might be a job that does a daily reconciliation of a user’s account, but performs heavy lifting across many accounts, so ideally it only runs once a day per account to save on worker resources. River can guarantee job uniqueness along dimensions based on a combination of job properties like arguments and insert period.

type ReconcileAccountArgs struct {
    AccountID int `json:"account_id"`
}

func (ReconcileAccountArgs) Kind() string { return "reconcile_account" }

// InsertOpts returns custom insert options that every job of this type will
// inherit, including unique options.
func (ReconcileAccountArgs) InsertOpts() river.InsertOpts {
    return river.InsertOpts{
        UniqueOpts: river.UniqueOpts{
            ByArgs:   true,
            ByPeriod: 24 * time.Hour,
        },
    }
}

...

// First job insertion for account 1.
_, err = riverClient.Insert(ctx, ReconcileAccountArgs{AccountID: 1}, nil)
if err != nil {
    panic(err)
}

// Job is inserted a second time, but it doesn't matter because its unique
// args cause the insertion to be skipped because it's meant to only run
// once per account per 24 hour period.
_, err = riverClient.Insert(ctx, ReconcileAccountArgs{AccountID: 1}, nil)
if err != nil {
    panic(err)
}

// Because the job is unique ByArgs, another job for account 2 _is_ allowed.
_, err = riverClient.Insert(ctx, ReconcileAccountArgs{AccountID: 2}, nil)
if err != nil {
    panic(err)
}

opts = River::UniqueOpts.new(by_args: true, by_period: 24 * 3600)
client.insert ReconcileAccountArgs.new(account_id: 1), unique_opts: opts

# Returns the existing job during the same period.
client.insert ReconcileAccountArgs.new(account_id: 1), unique_opts: opts

# Different arguments allow another job.
client.insert ReconcileAccountArgs.new(account_id: 2), unique_opts: opts

#[derive(Clone, Debug, serde::Deserialize, riverqueue::JobArgs, serde::Serialize)]
#[river(kind = "reconcile_account", unique(by_args, by_period = "24h"))]
struct ReconcileAccountArgs {
    account_id: i64,
}

client.insert(ReconcileAccountArgs { account_id: 1 }).await?;

// Returns the existing job during the same period.
client.insert(ReconcileAccountArgs { account_id: 1 }).await?;

// Different arguments allow another job.
client.insert(ReconcileAccountArgs { account_id: 2 }).await?;

const reconcileAccount = defineJob({
  kind: "reconcile_account",
  schema: z.object({ account_id: z.number().int() }),
  defaults: { unique: { byArgs: true, byPeriod: { hours: 24 } } },
});

await client.insert(reconcileAccount, { account_id: 1 });

// Returns the existing job during the same period.
await client.insert(reconcileAccount, { account_id: 1 });

// Different arguments allow another job.
await client.insert(reconcileAccount, { account_id: 2 });

Each unique dimension increases the specificity of the key, allowing more distinct jobs. Kind is included unless explicitly excluded. With a daily period alone, one job of that kind is allowed per period. Adding arguments allows one per set of encoded arguments per period. With no unique options, no uniqueness is enforced.

Unique by args

Uniqueness by arguments compares the encoded JSON. {"account_id":1} can coexist with {"account_id":2}, but a second job with {"account_id":1} conflicts.

Top-level keys are sorted before hashing; nested keys are not sorted recursively. When inserting from different languages, agree on JSON field names, values, number representations, and nested key ordering so equivalent jobs produce the same unique key.

Using a subset of args

type ReconcileAccountArgs struct {
    CustomerID int    `json:"customer_id" river:"unique"`
    TraceID    string `json:"trace_id"`
}

client.insert args, unique_opts: River::UniqueOpts.new(by_args: [:customer_id])

#[derive(Clone, Debug, serde::Deserialize, riverqueue::JobArgs, serde::Serialize)]
#[river(kind = "reconcile_account", unique(by_args))]
struct ReconcileAccountArgs {
    #[river(unique)]
    customer_id: i64,
    trace_id: String,
}

await client.insert(reconcileAccount, args, {
  unique: { byArgs: ["customer_id"] },
});

Unique by period

Uniqueness by period compares the job kind and a time interval. On insertion, the current time is rounded down to the nearest multiple of the given period, and a job is only inserted if there isn’t already an existing job that has been created at or scheduled to run between this lower bound and the next multiple of the period.

For example, if a job is inserted with a fifteen-minute unique period and the current time is 15:21:00, it’ll be unique for the interval of 15:15:00 to 15:30:00. A new job inserted at 15:28:00 will be skipped on grounds of uniqueness, but one inserted at 15:31:00 would be allowed.

The period is the most commonly used unique property, and other properties are most likely to be specified along with it, rather than be configured by themselves.

Unique by queue

Uniqueness by queue includes the queue name in the unique key.

For example, if a job with kind reconcile_account is inserted into queue default, a new insertion of reconcile_account would be skipped on grounds of uniqueness, but a reconcile_account inserted to queue high_priority would be allowed.

Unique by state

Uniqueness by state controls which job states prevent another insertion. The default includes available, completed, pending, running, retryable, and scheduled; it excludes cancelled and discarded.

This default is usually the right setting for most unique jobs, but a custom value might be useful in tweaking behavior. For example, removing completed from the set above would mean that uniqueness would be enforced within active job states (i.e. being run or available to be run), so that each time a job with this kind completes, a new one is allowed to be enqueued.

If retryable is removed from the list, it’s possible for an erroring job to hit a conflict when it is retried (because a duplicate has since been inserted). In this scenario, River will move the conflicting job to discarded since it cannot be retried.

Job retention horizons

When thinking about job state, remember that completed jobs aren’t retained permanently. The default retention time for completed jobs is 24 hours, so with a default unique state list, even with no unique period set a new job would be allowed to be inserted every 24 hours as the previous completed job is pruned.

Checking for skipped inserts

insertRes, err := riverClient.Insert(ctx, SortArgs{...}, nil)
insertRes.UniqueSkippedAsDuplicate // true if job was skipped

result = client.insert args, unique_opts: opts

result.unique_skipped_as_duplicate? # true if an existing job matched
result.job # the inserted or existing job

let result = client.insert(args).await?;

println!("Duplicate: {}", result.unique_skipped_as_duplicate);
println!("Job ID: {}", result.id());

const result = await client.insert(reconcileAccount, args);

console.log(result.status === "duplicate");
console.log(result.job.id); // The inserted or existing job.

At least once

While River can ensure that a unique job is only inserted once, it can’t guarantee that it will be worked exactly once. A unique job could work successfully, but fail to have its completed status persisted to the database, which would require that it be worked again for River to be sure it went through. Like other jobs, River provides an at-least-once guarantee for unique jobs.

Unique index

Job uniqueness is enforced with a special partial unique index on the river_job table. The unique index only applies to jobs whose state is within its specified list of unique states, which can be customized on a per-job basis.