Batches

Batches allow you to group multiple jobs together and monitor them as a collection. You can execute a set of jobs in parallel and receive callbacks when certain conditions are met.

Creating a Batch

import "github.com/mgomes/senna/client"

batch := client.NewBatch().
    WithDescription("Process uploaded spreadsheet").
    Add("process_row", map[string]any{"row_id": 1}).
    Add("process_row", map[string]any{"row_id": 2}).
    Add("process_row", map[string]any{"row_id": 3}).
    OnCompleteCallback("batch_finished")

// Enqueue the batch atomically
err := c.EnqueueBatch(ctx, batch)
if err != nil {
    log.Fatal(err)
}

fmt.Printf("Started Batch %s\n", batch.ID)

By default, all jobs in a batch are enqueued atomically - either all succeed or none are enqueued.

Large Initial Batches

Very large batches can use WithAutoflush to enqueue the initial jobs in bounded chunks:

batch := client.NewBatch().
    WithDescription("Process large import").
    WithAutoflush(1000)

for _, row := range rows {
    batch.Add("process_row", map[string]any{"row_id": row.ID})
}

err := c.EnqueueBatch(ctx, batch)

WithAutoflush keeps Redis command size bounded, but it changes the default atomicity behavior. Senna validates the job payloads before writing batch state, but if Redis fails after an earlier chunk has been queued, those earlier jobs may still run without batch tracking after Senna cleans up the batch state.

Adding Jobs to a Batch

Basic Addition

batch := client.NewBatch().
    Add("job_type", map[string]any{"key": "value"}).
    Add("job_type", map[string]any{"key": "value2"})

With Options

batch := client.NewBatch().
    Add("high_priority_job", args, client.WithQueue("critical")).
    Add("normal_job", args)

Note

Batch.Add does not support WithEncryption, WithUniqueKey, WithDelay, WithScheduleAt, WithBatch, or WithBulkChunkSize. Unsupported options cause EnqueueBatch to return an error before writing batch state.

Callbacks

Senna can notify you when a batch reaches certain states via callback jobs.

Callback Types

Callback	When it Fires
OnCompleteCallback	When all jobs have executed at least once (success or failure)
OnSuccessCallback	Only when ALL jobs complete successfully
OnDeathCallback	The first time ANY job dies (exhausts all retries)

Important

Death and Success callbacks are mutually exclusive. Once a death callback fires, the success callback will never fire for that batch without manual intervention.

Registering Callbacks

batch := client.NewBatch().
    Add("job1", args).
    Add("job2", args).
    OnCompleteCallback("on_complete").    // Always fires when all jobs run
    OnSuccessCallback("on_success").      // Only if all succeed
    OnDeathCallback("on_death")           // First time any job dies

err := c.EnqueueBatch(ctx, batch)

Callback Handlers

Callbacks are regular jobs. The batch ID is passed in job.Args["batch_id"]:

w.Register("on_complete", func(ctx context.Context, job *senna.Job) error {
    batchID := job.Args["batch_id"].(string)
    fmt.Printf("Batch %s completed\n", batchID)
    return nil
})

w.Register("on_success", func(ctx context.Context, job *senna.Job) error {
    batchID := job.Args["batch_id"].(string)
    fmt.Printf("Batch %s succeeded!\n", batchID)
    // All jobs finished successfully
    return nil
})

w.Register("on_death", func(ctx context.Context, job *senna.Job) error {
    batchID := job.Args["batch_id"].(string)
    fmt.Printf("Batch %s had a failure\n", batchID)
    // At least one job exhausted retries
    return nil
})

Batch jobs and callback jobs use the normal worker registration options for execution behavior:

w.Register("batch_job", batchHandler, worker.WithJobTimeout(10*time.Minute))
w.Register("on_complete", callbackHandler, worker.WithJobTimeout(time.Minute))

Callback Options

Pass additional arguments to callback jobs:

batch := client.NewBatch().
    Add("sync_user", map[string]any{"user_id": 123}).
    OnSuccessCallback("notify_user", map[string]any{
        "email":    "user@example.com",
        "template": "sync_complete",
    })

// In the callback handler:
w.Register("notify_user", func(ctx context.Context, job *senna.Job) error {
    batchID := job.Args["batch_id"].(string)
    email := job.Args["email"].(string)        // From callback options
    template := job.Args["template"].(string)  // From callback options
    // Send notification...
    return nil
})

Callback options cannot use Senna-owned identity keys such as batch_id or parent_id. Those keys are reserved for the batch IDs Senna passes to callback jobs.

Callback Queue

Route callback jobs to a specific queue:

batch := client.NewBatch().
    Add("job1", args).
    OnCompleteCallback("on_complete").
    WithCallbackQueue("critical")  // Callbacks go to "critical" queue

Batch Status

Query the status of a batch programmatically:

status := c.BatchStatus(batchID)

// Refresh from Redis
if err := status.Refresh(ctx); err != nil {
    log.Fatal(err)
}

fmt.Printf("Total: %d\n", status.Total())       // Total jobs in batch
fmt.Printf("Pending: %d\n", status.Pending())   // Not yet complete
fmt.Printf("Successes: %d\n", status.Successes())
fmt.Printf("Failures: %d\n", status.Failures()) // Failed but may retry
fmt.Printf("Complete: %v\n", status.Complete()) // All jobs have run once
fmt.Printf("Dead: %v\n", status.Dead())         // Any job has died

Refresh loads the batch counters and metadata. If you also need failed job IDs for a full status view, use RefreshFull to fetch the batch state and failed ID set in one Redis pipeline:

if err := status.RefreshFull(ctx); err != nil {
    log.Fatal(err)
}

failedJIDs, err := status.FailedJIDs(ctx)
if err != nil {
    log.Fatal(err)
}
fmt.Printf("Failed jobs: %v\n", failedJIDs)

Waiting for Completion

Block until a batch completes:

status := c.BatchStatus(batchID)

// Wait up to 5 minutes
ctx, cancel := context.WithTimeout(ctx, 5*time.Minute)
defer cancel()

if err := status.Join(ctx); err != nil {
    if err == context.DeadlineExceeded {
        log.Println("Batch didn't complete in time")
    }
    log.Fatal(err)
}

fmt.Println("Batch completed!")

Dynamic Job Addition

Jobs within a batch can add more jobs to the same batch:

w.Register("parent_job", func(ctx context.Context, job *senna.Job) error {
    // Get batch handle from context
    batch := worker.BatchFromContext(ctx)
    if batch == nil {
        return nil // Not running in a batch
    }

    fmt.Printf("Working in batch %s\n", batch.BID())

    // Add child jobs to this batch
    childIDs := []int{1, 2, 3, 4, 5}
    for _, id := range childIDs {
        err := batch.Add(ctx, "child_job", map[string]any{"child_id": id})
        if err != nil {
            return err
        }
    }

    return nil
})

This is useful for:

• Processing hierarchical data (parent → children)
• Pagination (process page → enqueue next page)
• Recursive operations

Warning

You can dynamically add jobs to a batch but only from within another job in that batch. It is not safe to modify a batch from outside the batch or within its callbacks. client.WithBatch only writes batch metadata onto a standalone job. It does not add the job to the batch membership set and will not grant BatchFromContext access.

Nested Batches

While dynamic job addition adds jobs to the same batch, nested batches create a parent-child hierarchy. Child batches are independent batches that are tracked as part of a parent batch. The parent batch won't complete until all its child batches complete.

// Create a child batch linked to a parent
childBatch := client.NewBatch().
    WithParent(parentBatchID).
    OnSuccessCallback("child_done", map[string]any{"data": "value"})

childBatch.Add("child_job", map[string]any{"key": "value"})

err := c.EnqueueBatch(ctx, childBatch)

When to use each approach:

Approach	Use Case
Dynamic Job Addition	Add more work to the current batch (e.g., pagination, discovered subtasks)
Nested Batches	Create sequential workflow steps that must complete before the next begins

Workflow Orchestration

Nested batches enable complex multi-step workflows where each step can contain parallel jobs, but steps execute sequentially. This pattern gives you the best of both worlds: parallel execution within steps and sequential ordering between steps.

The key insight: callbacks can add new child batches to the parent, enabling step-by-step workflow progression.

[Overall Batch]
    │
    ├── Step 1: [Job A]
    │       ↓ (on success callback)
    ├── Step 2: [Job B, Job C, Job D] (parallel)
    │       ↓ (on success callback)
    └── Step 3: [Job E]
            ↓
    Overall Success Callback

Example: Video Processing Pipeline

This example processes an uploaded video through validation, transcoding, thumbnail generation, and publishing:

// Start the workflow by creating the overall batch
func StartVideoProcessing(ctx context.Context, c *client.Client, videoID string) error {
    overall := client.NewBatch().
        WithDescription(fmt.Sprintf("Process video %s", videoID)).
        OnSuccessCallback("video_complete", map[string]any{"video_id": videoID}).
        OnDeathCallback("video_failed", map[string]any{"video_id": videoID})

    // The initial job kicks off the workflow
    overall.Add("start_video_workflow", map[string]any{"video_id": videoID})

    return c.EnqueueBatch(ctx, overall)
}

The workflow starter creates the first step as a child batch:

w.Register("start_video_workflow", func(ctx context.Context, job *senna.Job) error {
    videoID := job.Args["video_id"].(string)
    parentID := job.BatchID // The overall batch

    // Step 1: Validate the video
    step1 := client.NewBatch().
        WithParent(parentID).
        OnSuccessCallback("validation_done", map[string]any{"video_id": videoID})

    step1.Add("validate_video", map[string]any{"video_id": videoID})

    return c.EnqueueBatch(ctx, step1)
})

w.Register("validate_video", func(ctx context.Context, job *senna.Job) error {
    videoID := job.Args["video_id"].(string)
    // Check format, duration, file integrity...
    return nil
})

Each step's callback creates the next step. Note how parent_id is accessed from job.Args:

w.Register("validation_done", func(ctx context.Context, job *senna.Job) error {
    videoID := job.Args["video_id"].(string)
    parentID := job.Args["parent_id"].(string) // The overall batch

    // Step 2: Transcode to multiple formats in parallel
    step2 := client.NewBatch().
        WithParent(parentID).
        OnSuccessCallback("transcoding_done", map[string]any{"video_id": videoID})

    step2.Add("transcode", map[string]any{"video_id": videoID, "format": "720p"})
    step2.Add("transcode", map[string]any{"video_id": videoID, "format": "1080p"})
    step2.Add("transcode", map[string]any{"video_id": videoID, "format": "4k"})

    return c.EnqueueBatch(ctx, step2)
})

w.Register("transcode", func(ctx context.Context, job *senna.Job) error {
    videoID := job.Args["video_id"].(string)
    format := job.Args["format"].(string)
    // Transcode video to the specified format...
    return nil
})

Continue the chain for remaining steps:

w.Register("transcoding_done", func(ctx context.Context, job *senna.Job) error {
    videoID := job.Args["video_id"].(string)
    parentID := job.Args["parent_id"].(string)

    // Step 3: Generate thumbnails at different timestamps
    step3 := client.NewBatch().
        WithParent(parentID).
        OnSuccessCallback("thumbnails_done", map[string]any{"video_id": videoID})

    for _, ts := range []int{10, 30, 60} {
        step3.Add("generate_thumbnail", map[string]any{
            "video_id":  videoID,
            "timestamp": ts,
        })
    }

    return c.EnqueueBatch(ctx, step3)
})

w.Register("thumbnails_done", func(ctx context.Context, job *senna.Job) error {
    videoID := job.Args["video_id"].(string)
    parentID := job.Args["parent_id"].(string)

    // Step 4: Publish to CDN
    step4 := client.NewBatch().
        WithParent(parentID)

    step4.Add("publish_to_cdn", map[string]any{"video_id": videoID})

    return c.EnqueueBatch(ctx, step4)
})

The overall batch callback fires when the entire workflow completes:

w.Register("video_complete", func(ctx context.Context, job *senna.Job) error {
    videoID := job.Args["video_id"].(string)
    // Mark video as ready, notify user, update database...
    log.Printf("Video %s processing complete!", videoID)
    return nil
})

w.Register("video_failed", func(ctx context.Context, job *senna.Job) error {
    videoID := job.Args["video_id"].(string)
    // Handle failure, notify admin, cleanup partial work...
    log.Printf("Video %s processing failed", videoID)
    return nil
})

Workflow Tips

Tip

Rule of thumb: Jobs read their batch metadata from job.BatchID and mutate their own batch through worker.BatchFromContext(ctx), which is only populated after verified batch membership. Callbacks access the parent batch via job.Args["parent_id"].

Note

Callback ordering: Child batch success callbacks always fire before the parent batch success callback. This guarantees your step callbacks execute in order before the overall workflow callback.

Warning

Callbacks cannot modify their own batch. Once a batch's callbacks are running, that batch is complete. Callbacks should only create new child batches on the parent batch, not add jobs to their own batch.

Batch Invalidation

You can invalidate (cancel) a batch so remaining jobs skip execution:

w.Register("check_job", func(ctx context.Context, job *senna.Job) error {
    batch := worker.BatchFromContext(ctx)

    // Something went wrong - invalidate the batch
    if somethingBadHappened {
        if err := batch.Invalidate(ctx); err != nil {
            return err
        }
    }

    return nil
})

Jobs should check validity before doing expensive work:

w.Register("cancelable_job", func(ctx context.Context, job *senna.Job) error {
    // Check if batch is still valid
    valid, err := worker.ValidWithinBatch(ctx)
    if err != nil {
        return err
    }
    if !valid {
        return nil // Skip execution, batch was invalidated
    }

    // Do expensive work...

    return nil
})

Iterating Batches

List and iterate through all batches:

// Active batches
batchSet := senna.NewBatchSet(redisClient, "myapp")
err := batchSet.Each(ctx, func(status *senna.BatchStatus) error {
    fmt.Printf("Batch %s: %d pending\n", status.BID(), status.Pending())
    return nil
})

// Dead batches (those with failed jobs)
deadSet := senna.NewDeadBatchSet(redisClient, "myapp")
err = deadSet.Each(ctx, func(status *senna.BatchStatus) error {
    failedJIDs, _ := status.FailedJIDs(ctx)
    fmt.Printf("Dead batch %s: %v\n", status.BID(), failedJIDs)
    return nil
})

Example: Processing a CSV File

// Enqueue the batch
batch := client.NewBatch().
    WithDescription("Process sales_report.csv")

for rowNum := 0; rowNum < totalRows; rowNum++ {
    batch.Add("process_csv_row", map[string]any{
        "file_id": fileID,
        "row":     rowNum,
    })
}

batch.OnCompleteCallback("csv_import_complete", map[string]any{
    "file_id": fileID,
}).OnDeathCallback("csv_import_failed", map[string]any{
    "file_id":   fileID,
    "notify":    "admin@example.com",
})

err := c.EnqueueBatch(ctx, batch)

// Handler for individual rows
w.Register("process_csv_row", func(ctx context.Context, job *senna.Job) error {
    fileID := job.Args["file_id"].(string)
    row := int(job.Args["row"].(float64))

    // Check if batch is still valid
    valid, _ := worker.ValidWithinBatch(ctx)
    if !valid {
        return nil // Batch was cancelled
    }

    // Process the row...

    return nil
})

// Completion callback
w.Register("csv_import_complete", func(ctx context.Context, job *senna.Job) error {
    fileID := job.Args["file_id"].(string)
    fmt.Printf("CSV %s import complete!\n", fileID)
    return nil
})

Important Notes

Caution

Never disable retries in batch jobs. If a job fails without retrying, it disappears and the batch will never complete. Always allow at least some retries:

// BAD - batch may hang forever
batch.Add("job", args, client.WithRetry(0))

// GOOD - allow retries
batch.Add("job", args, client.WithRetry(3))

Warning

Be careful using job uniqueness with batches. If an error is raised while defining the batch jobs, the unique lock will remain in Redis but the job may never be pushed. Consider whether you really need uniqueness for batch jobs.

Note

Batch expiration: Batches expire after 30 days if not completed. Make sure your jobs complete within this window.

Note

Empty batches: Empty batches (with zero jobs) are valid and will immediately fire callbacks:

batch := client.NewBatch().
    OnCompleteCallback("empty_batch_complete")

c.EnqueueBatch(ctx, batch) // Callback fires immediately

Tip

If you find batches stuck with pending jobs, especially around deployments, verify you are gracefully shutting down workers. See the Workers documentation for proper shutdown handling.

Tip

Batches can scale to millions of jobs. If you're pushing the envelope, verify your Redis instance sizing and consider using maxmemory-policy noeviction to prevent data loss.

Next Steps

• Use Rate Limiters to control batch throughput
• Handle Errors in batch jobs
• Add Middleware for logging batch progress