Enqueue Many Jobs

How to insert a batch of standalone jobs in one transaction.

When to Batch Enqueue

Use a batch enqueue when:

• You create many standalone jobs at one time. A bulk notification and a data-processing queue are examples.
• Separate Enqueue calls in a loop make too many round trips.
• You need all of the standalone jobs, or none.

cyclone.EnqueueMany uses pipelined inserts. It uses two server round trips for any batch size. That is faster than a loop of Enqueue for a large batch.

EnqueueMany: Batch Step Enqueue

inputs := []NotifyInput{
    {UserID: 1, Message: "Your order shipped"},
    {UserID: 2, Message: "Your order shipped"},
    {UserID: 3, Message: "Your order shipped"},
}

runIDs, err := cyclone.EnqueueMany(ctx, client, NotifyStep, inputs)
if err != nil {
    return fmt.Errorf("batch enqueue: %w", err)
}
fmt.Printf("enqueued %d jobs\n", len(runIDs))
// runIDs[i] matches inputs[i].

The returned []string has the same order as the input slice. runIDs[i] is the run ID for inputs[i].

EnqueueActionMany

EnqueueActionMany does the same for an action:

inputs := []SendEmailInput{
    {To: "alice@example.com", Subject: "Welcome"},
    {To: "bob@example.com",   Subject: "Welcome"},
}

runIDs, err := cyclone.EnqueueActionMany(ctx, client, SendEmailAction, inputs)

Transactional Batch Enqueue

Use cyclone.EnqueueManyTx or cyclone.EnqueueActionManyTx to insert the batch in your transaction:

tx, err := pool.Begin(ctx)
if err != nil {
    return err
}
defer tx.Rollback(ctx)

// Your business write:
if _, err := tx.Exec(ctx, `UPDATE campaigns SET status = 'launched' WHERE id = $1`, campaignID); err != nil {
    return err
}

// Batch enqueue in the same transaction:
runIDs, err := cyclone.EnqueueManyTx(ctx, client, tx, SendEmailStep, emailInputs)
if err != nil {
    return err
}

// Both the campaign update and all job rows commit together.
return tx.Commit(ctx)

If the transaction rolls back, Cyclone creates no standalone jobs.

Workers claim the batch within one poll interval of the commit. To wake them at once, pass a TrackWakeups transaction. Call PublishWakeups after the commit. The batch records one wakeup for its queue. See Transactional Enqueueing.

Performance Characteristics

Method	Round trips	Notes
Enqueue in a loop (N calls)	2N	Opens and commits N transactions
EnqueueMany (N items)	2	One transaction, pipelined INSERT
EnqueueManyTx (N items)	1 extra	Inserts into the caller transaction

For 1000 standalone jobs on a 1ms connection, a loop takes ~2 seconds. A batch takes ~2ms.

Error Behavior

EnqueueMany inserts all of the standalone jobs, or none. If one item fails to marshal, Cyclone rolls back the batch. If one insert fails, Cyclone rolls back the batch. Cyclone returns an error.

runIDs, err := cyclone.EnqueueMany(ctx, client, NotifyStep, inputs)
if err != nil {
    // None of the jobs were enqueued.
    return err
}
// All jobs enqueued successfully.

The error names the item that failed. A marshal error on item 5 is an example.

Unique Jobs in Batches

WithUnique applies to each item. A batch fails when a key conflicts with a standalone job that is already active.

Two items in one call can have the same key. Cyclone rejects the second item as a duplicate. When you expect duplicates, remove them before you call EnqueueMany. Or handle the *DuplicateError.

Testing Batch Inserts

func TestBatchEnqueue(t *testing.T) {
    env := cyclonetest.New(t, cyclonetest.WithWorkers(4))
    env.Mux().Register(NotifyStep)

    inputs := make([]NotifyInput, 10)
    for i := range inputs {
        inputs[i] = NotifyInput{UserID: i + 1, Message: "hello"}
    }

    runIDs, err := cyclone.EnqueueMany(context.Background(), env.Client(), NotifyStep, inputs)
    require.NoError(t, err)
    require.Len(t, runIDs, 10)

    for _, runID := range runIDs {
        env.RunUntilComplete(runID)
        env.RequireCompleted(runID)
    }
}

Operational Caveats

You can divide a batch of more than 10,000 items into batches of 1000 to 5000 items. A transaction that inserts 100,000 rows holds locks for a long time. That transaction can delay autovacuum.

Cyclone keeps every input item in memory before the transaction starts. For a large input with a large payload, that memory can be large.

A large batch transaction holds a table-level intent lock. Other inserts at the same time can wait for a short time. Keep a batch under 5000 items for a high-traffic table.

Related

• Enqueue Standalone Jobs
• Transactional Enqueueing