Enqueuing Jobs

The Zizq Node client exposes two enqueue methods:

client.enqueue(input) — enqueue a single job.
client.enqueueBulk(inputs) — enqueue many jobs in a single HTTP request.

JS:

await client.enqueue({
  type: "send_email",
  queue: "emails",
  payload: { userId: 42, template: "welcome" },
});

await client.enqueueBulk([
  { type: "send_email", queue: "emails", payload: { userId: 1 } },
  { type: "send_email", queue: "emails", payload: { userId: 2 } },
]);

Both accept enqueue inputs in the same shape.

Single enqueue

When enqueueing a single job, the enqueue() method returns the Job from the Zizq server, which provides all its metadata, such as id, status, readyAt etc. Note that payload is not part of the response.

JS:

const result = await client.enqueue({
  type: "send_email",
  queue: "emails",
  payload: { userId: 42, template: "welcome" },
});
result.id // "03fu0wm75gxgmfyfplwvazhex"

Bulk enqueue

Bulk enqueue works exactly the same as a single job enqueue, except that an array of inputs are provided, and an array of Job instances is returned in the order matching the inputs.

JS:

const results = await client.enqueueBulk([
  { type: "send_email", queue: "emails", payload: { userId: 1 } },
  { type: "send_email", queue: "emails", payload: { userId: 2 } },
]);
results.length // 2

Enqueue options

The following options are available on the inputs to client.enqueue() and client.enqueueBulk(). All of type, queue and payload are required inputs.

Tip

For more details on the jq query language, read the language specification on the jaq website or on jq.

Option	Description
`type` string	The type that identifies this job.
`queue` string	The name of the queue onto which this job is enqueued.
`payload` object	Any valid JSON-serializable type understood by the handler that will run this job.
`priority` number?	Optional priority value between `0` and `65536`. When not specified, the default priority from the server applies.
`readyAt` number?	Optional milliseconds since the Unix epoch at which this job becomes ready for processing. When set at a future time, the job is enqueued with the `scheduled` status. Otherwise the job is `ready` immediately.
`retryLimit` number?	Optional retry limit override, which defines the number of retries that can occur before the Zizq server marks the job `dead` and stops retrying. When not specified, the server's default retry limit applies.
`backoff` BackoffConfig?	Optional backoff policy specific to this job. When not specified the server's default backoff policy applies. When specified, all fields must be present as they form a single backoff curve formula.
`backoff.baseMs` number	Number of milliseconds used at the mimimum delay in all exponential backoff calculations.
`backoff.exponent` number	The power curve steepness of the exponential backoff formula. The number of job attempts is raised to this power and added onto the `baseMs`. Floating point values are acceptable.
`backoff.jitterMs` number	A random jitter delay used to avoid cascades of failures all retrying at the same time. A random number between `0` and `jitterMs` is picked, then multiplied by the number of job attempts. The result is then added onto the total delay which creates a natural spread.
`retention` RetentionConfig?	Optional retention policy specific to this job. When not specified the server's default retention policy applies.
`retention.deadMs` number?	Number of milliseconds for which this job should be retained after entering the `dead` status. When not specified the server's default applies.
`retention.completedMs` number?	Number of milliseconds for which this job should be retained after entering the `completed` status. When not specified the server's default applies.
`uniqueKey` (string \| (EnqueueInput) => string)?	Optional unique key used to handle enqueue-time de-duplication of jobs. Can be a function receiving the full enqueue input object and returning a string. Generally applications will use `payloadHasher()` to produce a configurable hash function here. A job that is unique across all of its payload uses simply `{ uniqueKey: payloadHasher() }`. requires a pro license on the server.
`uniqueWhile` ("queued" \| "active" \| "exists")?	Optional unique scope for which uniqueness is enforced on this job after it is enqueued. One of: `queued` — duplicates will be prevented while this job is in the `scheduled` or `ready` states. `active` — duplicates will be prevented while this job is in the `scheduled`, `ready` or `in_flight` states. `exists` — duplicates will be prevented for as long as this job is present on the server.
`batch` BatchConfig?	Optional batched jobs configuration for this job. Batched jobs allow multiple jobs to be folded/coalesced together into a larger batch job. Generally applications will use the `batchConfig()` helper rather than construct this object manually. For example, a job that holds an array in its `items` key could be configured to accumulate batches of up to 1000 items using `{ batch: batchConfig(1000, '.items') }`. requires a pro license on the server.
`batch.key` string \| (EnqueueInput) => string	Shared batch key used to identify jobs that can be folded together. Can be a function receiving the full enqueue input object and returning a string. Generally applications will use `payloadHasher()` to produce a configurable hash function here.
`batch.when` string	A `jq` expression evaluated whenever a subsequent enqueue occurs using the same `batch.key`. This acts as a predicate returning a boolean-ish result indicating whether the new job can be folded into the existing job, or a new job should be enqueued, starting a new batch and sealing the existing batch. The expression has two implicitly bound variables `$existing` and `$new`. These are bound to the existing job's payload, and the incoming job's payload. The typical use case is to return true if the combined payload length is below a desired threshold. If the expression returns a truthy value, Zizq evaluates the `batch.fold` expression to derive the folded payload. For example, a job that holds an array in its `items` key could be configured to accumulate up to 1000 items before being sealed and starting a new batch by using the expression `($existing.items + $new.items) \| length <= 1000`. As a defensive measure against writing an invalid `jq` expression, this expression is validated by binding both `$existing` and `$new` to the incoming payload before the job is enqueued.
`batch.fold` string	A `jq` expression evaluated whenever a subsequent enqueue occurs using the same `batch.key` and `batch.when` evaluates to a truthy value. This acts as a reducer expression combining the existing job's payload with the new payload. The expression has two implicitly bound variables `$existing` and `$new`. These are bound to the existing job's payload, and the incoming job's payload. The typical use case is to concatenate two arrays together, either for the entire payload, or at a sub-path of the payload, though any reasonably complex logic may be applied here. For example, a job that holds an array in its `items` key could be configured to fold new items into itself using the expression `$existing \| .items + $new.items`. As a defensive measure against writing an invalid `jq` expression, this expression is validated by binding both `$existing` and `$new` to the incoming payload before the job is enqueued.
`budgets` BudgetBindingInput[]?	Budgets this job is bound to and must satisfy before it dispatches. This is used for rate limiting and concurrency control. See Concurrency & Rate Limiting for full details on this feature. Requires a pro license on the server.
`budgets[].key` string	The key that identifies this budget.
`budgets[].cost` number?	The number of tokens this job draws from the budget when it dispatches. Default to `1` when not specified.
`budgets[].createWith` BudgetPolicy?	Optional definition of the budget's policy to be used if the budget does not already exist.
`budgets[].createWith.allocation` number	The total number of tokens made available by the budget. Must not be less than the job's `cost`.
`budgets[].createWith.strategy` BudgetStrategy	The definition for how this budget's tokens are managed.
`budgets[].createWith.strategy.type` "while_in_flight" \| "time_based"	Enum specifying which strategy to use. A `while_in_flight` strategy implements concurrency control — jobs debit tokens on dispatch, and release them on completion. A `time_based` strategy implements a rate limit — jobs debit tokens on dispatch, and those tokens are automatically released on a timer. Tokens continue accruing until the `allocation` is full, or the `burst` is reached if set.
`budgets[].createWith.strategy.durationMs` number?	Required for a `time_based` strategy. Specifies the period of time over which tokens are replenished.
`budgets[].createWith.strategy.burst` number?	Only valid for a `time_based` strategy. Specifies the maximum number of tokens the budget may accrue at once. The default value is the same as the `allocation`.

Dynamic Job Configuration

The inputs to enqueue jobs are plain JavaScript objects. Applications can implement helper functions to provide enqueue inputs for jobs dynamically. For example changing the priority based on the time of day, or based on details in the job payload.

JS:

import type { EnqueueInput } from "@zizq-labs/zizq";

export type SendEmailPayload = {
  to: string;
  subject: string;
};

export function sendEmailJob(payload: SendEmailPayload): EnqueueInput {
  return {
    type: "send_email",
    queue: "emails",
    priority: payload.to.endsWith("@important.com") ? 10 : 100,
    payload,
  };
}

Just wrap the payload with the job of the appropriate type.

JS:

import { Client } from "@zizq-labs/zizq";
import { sendEmailJob } from "./jobs";

const client = new Client({ url: "http://localhost:7890" });

await client.enqueue(sendEmailJob({
  to: "example@important.com",
  subject: "Important email",
}));

Keyboard shortcuts

Zizq Node Client

Enqueuing Jobs

Single enqueue

Bulk enqueue

Enqueue options

Dynamic Job Configuration