-
Notifications
You must be signed in to change notification settings - Fork 0
Enqueuing Jobs
This guide covers all the ways to enqueue jobs and the available options.
import (
"github.com/mgomes/senna"
"github.com/mgomes/senna/client"
)
c, err := client.New(&client.Config{
Redis: senna.RedisConfig{
Addr: "localhost:6379",
Password: "",
DB: 0,
},
Namespace: "myapp",
Settings: client.Settings{
DefaultQueue: "default", // Queue used when none specified
DefaultRetry: 25, // Default retry count
},
})
if err != nil {
log.Fatal(err)
}
defer c.Close()Enqueue a job to run as soon as a worker is available:
job, err := c.Enqueue(ctx, "send_email", map[string]any{
"to": "user@example.com",
"subject": "Welcome!",
"body": "Thanks for signing up.",
})Run a job after a specified duration:
// Run in 5 minutes
job, err := c.EnqueueIn(ctx, 5*time.Minute, "send_reminder", map[string]any{
"user_id": 123,
})
// Run in 1 hour
job, err = c.EnqueueIn(ctx, time.Hour, "cleanup", nil)Run a job at a specific time:
// Run at midnight on New Year's
runAt := time.Date(2025, 1, 1, 0, 0, 0, 0, time.UTC)
job, err := c.EnqueueAt(ctx, runAt, "new_year_notification", nil)
// Run tomorrow at 9am
tomorrow := time.Now().Add(24 * time.Hour)
scheduled := time.Date(
tomorrow.Year(), tomorrow.Month(), tomorrow.Day(),
9, 0, 0, 0, time.Local,
)
job, err = c.EnqueueAt(ctx, scheduled, "daily_report", nil)Scheduled jobs are stored in a Redis sorted set with the execution timestamp as the score. A background scheduler in the worker checks for due jobs and moves them to their target queues.
┌─────────────┐ EnqueueIn/At ┌──────────────────┐
│ Client │ ─────────────────► │ Scheduled Set │
└─────────────┘ │ (sorted by time)│
└────────┬─────────┘
│
Scheduler checks every
ScheduledPollInterval (5s)
│
▼
┌──────────────────┐
│ Job Queue │
└────────┬─────────┘
│
▼
┌──────────────────┐
│ Worker │
└──────────────────┘
Note
Scheduled jobs have second-level precision. The scheduler checks for due jobs every ScheduledPollInterval (default 5 seconds), so jobs may run up to 5 seconds after their scheduled time.
Tip
When running multiple workers, scheduled jobs are processed atomically—only one worker will move each job to the queue, preventing duplicates.
You can adjust how frequently workers check for due scheduled jobs:
w, err := worker.New(&worker.Config{
Redis: redisConfig,
Namespace: "myapp",
Settings: senna.WorkerSettings{
ScheduledPollInterval: 1 * time.Second, // Check every second (default: 5s)
},
})Lower intervals mean scheduled jobs run closer to their target time, but increase Redis load. For most applications, the default 5-second interval is sufficient.
Note
Job arguments are stored as JSON. Numbers become float64, so use type assertions carefully in handlers.
Job arguments are passed as map[string]any. In your handler, you'll need to type-assert values:
// Enqueue with various argument types
c.Enqueue(ctx, "process_order", map[string]any{
"order_id": 12345,
"amount": 99.99,
"items": []string{"item1", "item2"},
"customer": map[string]any{"id": 1, "name": "Alice"},
"priority": true,
})
// In the handler
w.Register("process_order", func(ctx context.Context, job *senna.Job) error {
// Numbers are float64 when unmarshaled from JSON
orderID := int(job.Args["order_id"].(float64))
amount := job.Args["amount"].(float64)
priority := job.Args["priority"].(bool)
// Slices and maps need casting
items := job.Args["items"].([]any)
customer := job.Args["customer"].(map[string]any)
return nil
})Send a job to a specific queue:
// Send to "critical" queue
c.Enqueue(ctx, "process_payment", args, client.WithQueue("critical"))
// Send to "low" priority queue
c.Enqueue(ctx, "generate_report", args, client.WithQueue("low"))Override the default retry count:
// Allow up to 5 retries
c.Enqueue(ctx, "send_webhook", args, client.WithRetry(5))
// No retries (dangerous for batch jobs!)
c.Enqueue(ctx, "one_shot_job", args, client.WithRetry(0))Caution
Setting WithRetry(0) disables retries. If the job fails, it's immediately moved to the dead queue. Never disable retries for batch jobs—the batch may never complete.
Add a delay before execution (alternative to EnqueueIn):
c.Enqueue(ctx, "job_type", args, client.WithDelay(10*time.Minute))Prevent duplicate jobs using a unique key:
// Only one sync job per user within the TTL
_, err := c.Enqueue(ctx, "sync_user_data", map[string]any{
"user_id": 123,
}, client.WithUniqueKey("sync:user:123", time.Hour))
// Second enqueue with same key returns error
_, err = c.Enqueue(ctx, "sync_user_data", map[string]any{
"user_id": 123,
}, client.WithUniqueKey("sync:user:123", time.Hour))
if err != nil {
var dupErr *senna.DuplicateJobError
if errors.As(err, &dupErr) {
log.Println("Job already enqueued, skipping")
}
}The unique key is automatically cleared when the job completes or fails permanently.
Warning
Be careful using unique keys with Batches. If an error occurs while defining batch jobs, the unique lock may remain in Redis but the job might not be enqueued.
Encrypt sensitive job arguments at rest:
// First, configure encryption in the client
key := make([]byte, 32) // Load from secure storage
c, err := client.New(&client.Config{
Redis: redisConfig,
Namespace: "myapp",
Encryption: &senna.EncryptionSettings{
Enabled: true,
Key: key,
},
})
// Enqueue with encryption
c.Enqueue(ctx, "process_pii", map[string]any{
"ssn": "123-45-6789",
"card_number": "4111111111111111",
}, client.WithEncryption())Arguments are encrypted with AES-GCM and automatically decrypted when the worker processes the job. See Encryption for setup, key handling, and rotation guidance.
client.WithEncryption() requires the client to be configured with encryption.
Without that configuration, enqueue returns client.ErrEncryptionUnavailable
and does not write the job.
Important
Store encryption keys securely (environment variables, secrets manager). Never commit keys to version control. The worker must use the same key to decrypt arguments.
Options can be combined:
c.Enqueue(ctx, "sensitive_sync", args,
client.WithQueue("critical"),
client.WithRetry(3),
client.WithUniqueKey("sync:123", time.Hour),
client.WithEncryption(),
)When you need to enqueue many jobs of the same type, use bulk enqueuing to reduce Redis network round trips. Instead of making N individual calls, bulk enqueuing pushes jobs in bounded Redis chunks. The default chunk size is 1000 jobs.
// Create a list of arguments for each job
argsList := []map[string]any{
{"user_id": 1},
{"user_id": 2},
{"user_id": 3},
// ... up to thousands of jobs
}
// Enqueue all jobs with bounded Redis command size
jobs, err := c.EnqueueBulk(ctx, "sync_user", argsList)
if err != nil {
log.Fatal(err)
}
fmt.Printf("Enqueued %d jobs\n", len(jobs))Schedule multiple jobs to run after a delay or at a specific time:
// Run all jobs in 1 hour
jobs, err := c.EnqueueBulkIn(ctx, time.Hour, "send_reminder", argsList)
// Run all jobs at midnight
midnight := time.Date(2025, 1, 2, 0, 0, 0, 0, time.Local)
jobs, err = c.EnqueueBulkAt(ctx, midnight, "nightly_sync", argsList)All jobs in a bulk operation share the same options:
jobs, err := c.EnqueueBulk(ctx, "process_order", argsList,
client.WithQueue("critical"),
client.WithRetry(3),
client.WithEncryption(),
client.WithBulkChunkSize(500),
)Warning
Unique keys (WithUniqueKey) are not supported with bulk enqueue. Attempting to use them will return an error.
WithBulkChunkSize controls the maximum number of jobs sent in each Redis command. Use a smaller value when job payloads are large or Redis command latency matters more than total round trips.
EnqueueBulk returns a slice of successfully enqueued jobs. Each job has its own unique ID:
jobs, err := c.EnqueueBulk(ctx, "job_type", argsList)
for _, job := range jobs {
fmt.Printf("Enqueued job %s with args %v\n", job.ID, job.Args)
}If any job arguments fail to serialize or encrypt, EnqueueBulk returns an error before writing jobs to Redis.
If Redis fails after an earlier chunk was accepted, EnqueueBulk returns the accepted job prefix along with a *client.BulkPartialError:
jobs, err := c.EnqueueBulk(ctx, "job_type", argsList)
if err != nil {
var partial *client.BulkPartialError
if errors.As(err, &partial) {
log.Printf("Redis accepted %d of %d jobs", partial.Enqueued, partial.Total)
// jobs contains the accepted prefix
}
}Bulk enqueuing eliminates the network round-trip latency for each job. For example, enqueuing 1000 jobs individually might take 1000+ round trips, while bulk enqueuing does it in one chunk by default.
// Efficient: one chunk for 1000 jobs
argsList := make([]map[string]any, 1000)
for i := range argsList {
argsList[i] = map[string]any{"record_id": i}
}
jobs, _ := c.EnqueueBulk(ctx, "process_record", argsList)For very large inputs, still page records from your source system instead of building the entire dataset in memory. WithBulkChunkSize bounds each Redis command, but EnqueueBulk still returns a []*senna.Job for every input job.
// Process 100,000 records in chunks of 1000
const chunkSize = 1000
for offset := 0; offset < totalRecords; offset += chunkSize {
// Query for this chunk of records
records := db.Query("SELECT id FROM users LIMIT ? OFFSET ?", chunkSize, offset)
// Build arguments list
argsList := make([]map[string]any, len(records))
for i, record := range records {
argsList[i] = map[string]any{"user_id": record.ID}
}
// Enqueue this chunk
jobs, err := c.EnqueueBulk(ctx, "sync_user", argsList, client.WithQueue("bulk"))
if err != nil {
log.Printf("Failed to enqueue chunk at offset %d: %v", offset, err)
continue
}
log.Printf("Enqueued %d jobs (offset %d)", len(jobs), offset)
}Bulk enqueue works with batches for tracking completion:
// Create a batch first
batch := client.NewBatch().
WithDescription("Sync all users").
OnCompleteCallback("sync_complete")
// Add jobs to the batch using bulk enqueue
argsList := []map[string]any{
{"user_id": 1},
{"user_id": 2},
{"user_id": 3},
}
// Note: You'll need to add jobs to the batch manually or use EnqueueBatch
// For bulk operations within a running batch job, use batch.Add() in a loopNote
For batch operations, consider using EnqueueBatch with multiple Add() calls, which provides atomic enqueuing and automatic callback handling. Bulk enqueue is best for high-volume scenarios where you don't need batch tracking.
Enqueue returns a *senna.Job with the following fields:
type Job struct {
ID string // Unique job ID (UUID)
Type string // Job type (handler name)
Args map[string]any // Job arguments
Queue string // Target queue
Retry int // Max retry count
CreatedAt time.Time // When the job was created
EnqueuedAt time.Time // When the job was enqueued
}Full Redis configuration options:
senna.RedisConfig{
Addr: "localhost:6379", // Redis address
Password: "secret", // Redis password (if any)
DB: 0, // Redis database number
PoolSize: 100, // Connection pool size
MinIdleConns: 10, // Minimum idle connections
DialTimeout: 5 * time.Second, // Connection timeout
ReadTimeout: 3 * time.Second, // Read timeout
WriteTimeout: 3 * time.Second, // Write timeout
}client.Settings{
DefaultQueue: "default", // Default queue for jobs
DefaultRetry: 25, // Default retry count
}