Queues and Priority

How to send work to a named queue. How to control claim order in one queue.

The Default Queue

Every step uses the "default" queue unless you set another queue. A client that you start with WithWorkers(n) starts n workers on the default queue.

WithQueue: Assign a Step to a Queue

Set the queue on the step. That queue then applies to every enqueue of this step.

var ReportStep = cyclone.NewStep("report", reportFn,
    cyclone.WithQueue("bulk"),
)

Set a different queue on one enqueue:

cyclone.Enqueue(ctx, client, reportStep, in, cyclone.WithQueue("priority"))

The per-enqueue value wins when both are set.

A queue name has 1–256 bytes. A queue name can contain ASCII letters, digits, ., _, or -. Examples are default, bulk, high-priority, and tenant.eu.

WithWorkerQueues: Configure Workers per Queue

client, _ := cyclone.NewClient(pool, mux,
    cyclone.WithWorkerQueues(map[string]int{
        "default":  8,
        "bulk":     2,
        "priority": 4,
    }),
)

Each queue has its own worker pool. The check requires MaxConns >= txSlots + 6. By default, txSlots equals the sum of the workers.

A queue that WithWorkerQueues does not list has no workers on this client. No worker claims a step on that queue until you start a client that lists the queue.

Queue Isolation

Workers claim steps only from their configured queues. A slow step on bulk cannot take a worker from priority. Those steps are in separate queues.

Separate queues keep workloads apart. Put user-facing work on one queue. Put high-throughput background work on another queue.

Priority Within a Queue

In one queue, a worker claims a higher-priority step before a lower-priority step. The default priority is 0. A higher value is claimed earlier.

// High priority for user-initiated exports:
cyclone.Enqueue(ctx, client, exportStep, in, cyclone.WithPriority(10))

// Normal background sync:
cyclone.Enqueue(ctx, client, syncStep, in) // priority = 0

// Low priority for analytics batch:
cyclone.Enqueue(ctx, client, analyticsStep, in, cyclone.WithPriority(-10))

Set the priority on the step. That value is the default.

var UrgentStep = cyclone.NewStep("urgent", urgentFn,
    cyclone.WithQueue("default"),
    cyclone.WithPriority(100),
)

Tie-Breaking by scheduled_at then ID

When several steps have the same priority, workers claim them in FIFO order. The first key is scheduled_at. The second key is the row ID. For delayed steps with the same priority, the worker claims each step when its delay ends. The worker follows that time order.

Sizing Worker Pools

A starting formula:

pool MaxConns = txSlots + 6
txSlots = workers (default) — reduce if handlers rarely use ctx.CompletionTx() concurrently
workers = target concurrent executions

For a CPU-bound handler, set the worker count near the number of cores. For an I/O-bound handler, the worker count can be much higher. An HTTP call is I/O-bound. A database query is I/O-bound.

For mixed work, use one queue for each kind of work. Size each queue on its own.

// 4 CPU-bound workers + 16 I/O-bound workers
cyclone.WithWorkerQueues(map[string]int{
    "cpu-work": 4,
    "io-work":  16,
})
// Default txSlots=20, so Pool MaxConns >= 20 + 6 = 26.

WithTxSlots: Decouple Worker Count from DB Concurrency

txSlots limits how many completion transactions can be open at one time. By default, txSlots equals the total worker count. A CPU-bound handler completes quickly. Those handlers rarely overlap in ctx.CompletionTx(). You can set a lower txSlots. A lower value reduces the pool requirement.

client, _ := cyclone.NewClient(pool, mux,
    cyclone.WithWorkers(20),
    cyclone.WithTxSlots(5), // at most 5 handlers in ctx.CompletionTx() at once
    // pool only needs 5 + 6 = 11 connections instead of the default 20 + 6 = 26
)

Queue Naming Rules

• Use ASCII letters in upper or lower case, digits, dots ., hyphens -, and underscores _.
• Use 1 to 256 bytes.
• Do not use a space.
• Do not use any other character.
• "default" is the queue when no queue option is set. "default" is otherwise an ordinary queue name.

Operational Metrics by Queue

stats, _ := client.Stats(ctx)
for _, q := range stats.Queues {
    fmt.Printf("queue=%-20s available=%d running=%d\n", q.Queue, q.Available, q.Running)
    if q.OldestAvailable != nil {
        fmt.Printf("  oldest_available_age=%s\n", *q.OldestAvailable)
    }
    if q.LeaseExpired > 0 {
        fmt.Printf("  lease_expired=%d\n", q.LeaseExpired)
    }
}

Or read the same figures in SQL:

SELECT queue, available, running, oldest_available_age, lease_expired
FROM cyclone_v_queues
ORDER BY available DESC;

Common Queue Layouts

Two queues, one for user-facing work and one for background work:

cyclone.WithWorkerQueues(map[string]int{
    "default": 4, // user-facing jobs; fast
    "bulk":    2, // background jobs; can be slow
})

A layout that separates priority:

cyclone.WithWorkerQueues(map[string]int{
    "critical": 8,  // SLA-bound
    "default":  4,  // normal
    "bulk":     2,  // best-effort
})

One queue for each integration:

cyclone.WithWorkerQueues(map[string]int{
    "stripe":   2,  // rate-limited; cap workers to stay in rate limit
    "sendgrid": 4,
    "default":  8,
})

Related

• Queue Sizing
• Observability