batch-operations.md

Batch Operations

Hammerwork supports batch operations for bulk job processing: a JobBatch is inserted in a single transaction, tracked as a unit, and configured with a failure handling mode.

Table of Contents

• Overview
• Creating Job Batches
• Batch Configuration
• Failure Handling Modes
• Worker Batch Processing
• Monitoring Batch Progress
• CLI Batch Commands
• Best Practices

Overview

Batch operations provide:

• Bulk insertion: all jobs of a batch are inserted in one transaction instead of one round trip each
• Atomic enqueue: a batch is stored completely or not at all
• Progress tracking: get_batch_status reports pending, completed and failed counts
• Flexible failure handling: choose what happens when a job of the batch fails
• Batch claims: workers can claim many jobs per round trip and keep per-batch processing statistics

A batch holds up to 10,000 jobs, all on the same queue.

Creating Job Batches

Basic Batch Creation

# #[allow(unused_imports)] use hammerwork::queue::DatabaseQueue;
# #[allow(unused_variables, unused_mut, dead_code, unreachable_code)]
# async fn doc(queue: std::sync::Arc<hammerwork::JobQueue<sqlx::Postgres>>, handler: hammerwork::worker::JobHandler, jobs: Vec<hammerwork::Job>, worker: hammerwork::Worker<sqlx::Postgres>, batch_id: hammerwork::BatchId) -> std::result::Result<(), Box<dyn std::error::Error>> {
use hammerwork::{batch::{JobBatch, PartialFailureMode}, Job, queue::DatabaseQueue};
use serde_json::json;

let jobs = vec![
    Job::new("email_queue".to_string(), json!({
        "to": "user1@example.com",
        "subject": "Welcome!"
    })),
    Job::new("email_queue".to_string(), json!({
        "to": "user2@example.com",
        "subject": "Newsletter"
    })),
    Job::new("email_queue".to_string(), json!({
        "to": "user3@example.com",
        "subject": "Updates"
    })),
];

let batch = JobBatch::new("welcome_emails")
    .with_jobs(jobs)
    .with_batch_size(100)
    .with_partial_failure_handling(PartialFailureMode::ContinueOnError);

assert_eq!(batch.job_count(), 3);

let batch_id = queue.enqueue_batch(batch).await?;
# Ok(())
# }

Building Batches Incrementally

# #[allow(unused_imports)] use hammerwork::queue::DatabaseQueue;
# #[allow(unused_variables, unused_mut, dead_code, unreachable_code)]
# async fn doc(queue: std::sync::Arc<hammerwork::JobQueue<sqlx::Postgres>>, handler: hammerwork::worker::JobHandler, jobs: Vec<hammerwork::Job>, worker: hammerwork::Worker<sqlx::Postgres>, batch_id: hammerwork::BatchId) -> std::result::Result<(), Box<dyn std::error::Error>> {
use hammerwork::{batch::{JobBatch, PartialFailureMode}, Job};
use serde_json::json;

let files = vec![("a.csv", 10u64), ("b.csv", 20)];

let mut batch = JobBatch::new("data_processing");

for (path, size) in files {
    let job = Job::new("process_queue".to_string(), json!({
        "file": path,
        "size": size
    }));
    batch = batch.add_job(job);
}

batch = batch
    .with_partial_failure_handling(PartialFailureMode::CollectErrors)
    .with_metadata("department", "data_science")
    .with_metadata("priority", "high");
# Ok(())
# }

Batch Configuration

Batch Size

with_batch_size records a chunk size on the batch. enqueue_batch itself inserts every job of the batch in one transaction; the chunk size is used by JobBatch::into_chunks, which splits a batch into several smaller batches (named <name>_chunk_<n>, default chunk size 1000) that you can then enqueue one by one:

# #[allow(unused_imports)] use hammerwork::queue::DatabaseQueue;
# #[allow(unused_variables, unused_mut, dead_code, unreachable_code)]
# async fn doc(queue: std::sync::Arc<hammerwork::JobQueue<sqlx::Postgres>>, handler: hammerwork::worker::JobHandler, jobs: Vec<hammerwork::Job>, worker: hammerwork::Worker<sqlx::Postgres>, batch_id: hammerwork::BatchId) -> std::result::Result<(), Box<dyn std::error::Error>> {
use hammerwork::{batch::JobBatch, queue::DatabaseQueue};

let batch = JobBatch::new("large_batch")
    .with_jobs(jobs)
    .with_batch_size(100);

for chunk in batch.into_chunks() {
    queue.enqueue_batch(chunk).await?;
}
# Ok(())
# }

Metadata

Attach string metadata to batches for tracking:

# #[allow(unused_imports)] use hammerwork::queue::DatabaseQueue;
# #[allow(unused_variables, unused_mut, dead_code, unreachable_code)]
# async fn doc(queue: std::sync::Arc<hammerwork::JobQueue<sqlx::Postgres>>, handler: hammerwork::worker::JobHandler, jobs: Vec<hammerwork::Job>, worker: hammerwork::Worker<sqlx::Postgres>, batch_id: hammerwork::BatchId) -> std::result::Result<(), Box<dyn std::error::Error>> {
use hammerwork::batch::JobBatch;

let batch = JobBatch::new("campaign_emails")
    .with_jobs(jobs)
    .with_metadata("campaign_id", "summer_2024")
    .with_metadata("department", "marketing")
    .with_metadata("priority", "high");

assert_eq!(batch.metadata.get("campaign_id").map(String::as_str), Some("summer_2024"));
# Ok(())
# }

Job Priorities

Jobs within batches keep their individual priorities:

# #[allow(unused_imports)] use hammerwork::queue::DatabaseQueue;
# #[allow(unused_variables, unused_mut, dead_code, unreachable_code)]
# async fn doc(queue: std::sync::Arc<hammerwork::JobQueue<sqlx::Postgres>>, handler: hammerwork::worker::JobHandler, jobs: Vec<hammerwork::Job>, worker: hammerwork::Worker<sqlx::Postgres>, batch_id: hammerwork::BatchId) -> std::result::Result<(), Box<dyn std::error::Error>> {
use hammerwork::{batch::JobBatch, Job, JobPriority};
use serde_json::json;

let jobs = vec![
    Job::new("queue".to_string(), json!({"critical": true})).as_critical(),
    Job::new("queue".to_string(), json!({"important": true})).as_high_priority(),
    Job::new("queue".to_string(), json!({"regular": true})).with_priority(JobPriority::Normal),
];

let batch = JobBatch::new("mixed_priority_batch").with_jobs(jobs);
# Ok(())
# }

Validation

JobBatch::validate (called by enqueue_batch) rejects empty batches, batches over 10,000 jobs and batches whose jobs are not all on the same queue:

# #[allow(unused_imports)] use hammerwork::queue::DatabaseQueue;
# #[allow(unused_variables, unused_mut, dead_code, unreachable_code)]
# async fn doc(queue: std::sync::Arc<hammerwork::JobQueue<sqlx::Postgres>>, handler: hammerwork::worker::JobHandler, jobs: Vec<hammerwork::Job>, worker: hammerwork::Worker<sqlx::Postgres>, batch_id: hammerwork::BatchId) -> std::result::Result<(), Box<dyn std::error::Error>> {
use hammerwork::{batch::JobBatch, Job};
use serde_json::json;

assert!(JobBatch::new("empty").validate().is_err());

let mixed = JobBatch::new("mixed").with_jobs(vec![
    Job::new("queue1".to_string(), json!({})),
    Job::new("queue2".to_string(), json!({})),
]);
assert!(mixed.validate().is_err());
# Ok(())
# }

Failure Handling Modes

Inserting a batch is all-or-nothing: every job is stored with the same fields as enqueue (result storage, dependencies, workflow, tracing and retry strategy) in one transaction, whatever the failure mode.

get_batch_status tallies progress from the batch's jobs. The hammerwork_batches row moves to Processing when its first job finishes, and its counters, final status and completed_at are written when the last job finishes (or when it fails fast).

Batches support three failure handling modes:

ContinueOnError

Continue processing even if some jobs fail:

# #[allow(unused_imports)] use hammerwork::queue::DatabaseQueue;
# #[allow(unused_variables, unused_mut, dead_code, unreachable_code)]
# async fn doc(queue: std::sync::Arc<hammerwork::JobQueue<sqlx::Postgres>>, handler: hammerwork::worker::JobHandler, jobs: Vec<hammerwork::Job>, worker: hammerwork::Worker<sqlx::Postgres>, batch_id: hammerwork::BatchId) -> std::result::Result<(), Box<dyn std::error::Error>> {
use hammerwork::batch::{JobBatch, PartialFailureMode};

let batch = JobBatch::new("resilient_batch")
    .with_jobs(jobs)
    .with_partial_failure_handling(PartialFailureMode::ContinueOnError);
# Ok(())
# }

FailFast

Stop processing on the first failure. When a job of the batch fails terminally (its last attempt failed or timed out, or it was failed manually), the batch's jobs that have not started yet are marked Failed ("Batch failed: job ... failed") in the same transaction, so no worker picks them up. Jobs already running finish normally. Retried attempts do not trigger it.

# #[allow(unused_imports)] use hammerwork::queue::DatabaseQueue;
# #[allow(unused_variables, unused_mut, dead_code, unreachable_code)]
# async fn doc(queue: std::sync::Arc<hammerwork::JobQueue<sqlx::Postgres>>, handler: hammerwork::worker::JobHandler, jobs: Vec<hammerwork::Job>, worker: hammerwork::Worker<sqlx::Postgres>, batch_id: hammerwork::BatchId) -> std::result::Result<(), Box<dyn std::error::Error>> {
use hammerwork::batch::{JobBatch, PartialFailureMode};

let batch = JobBatch::new("critical_batch")
    .with_jobs(jobs)
    .with_partial_failure_handling(PartialFailureMode::FailFast);
# Ok(())
# }

CollectErrors

Keep processing and collect the error of every failed job. The errors are available from BatchResult::job_errors (keyed by job ID) when you query the batch status:

# #[allow(unused_imports)] use hammerwork::queue::DatabaseQueue;
# #[allow(unused_variables, unused_mut, dead_code, unreachable_code)]
# async fn doc(queue: std::sync::Arc<hammerwork::JobQueue<sqlx::Postgres>>, handler: hammerwork::worker::JobHandler, jobs: Vec<hammerwork::Job>, worker: hammerwork::Worker<sqlx::Postgres>, batch_id: hammerwork::BatchId) -> std::result::Result<(), Box<dyn std::error::Error>> {
use hammerwork::{batch::{JobBatch, PartialFailureMode}, queue::DatabaseQueue};

let batch = JobBatch::new("analytics_batch")
    .with_jobs(jobs)
    .with_partial_failure_handling(PartialFailureMode::CollectErrors);

let batch_id = queue.enqueue_batch(batch).await?;

// Later, once jobs have run
let result = queue.get_batch_status(batch_id).await?;
for (job_id, error) in &result.job_errors {
    eprintln!("Job {} failed: {}", job_id, error);
}
# Ok(())
# }

Worker Batch Processing

Batch Claims

A worker in batch mode claims up to batch_size jobs per poll in one round trip instead of one job per round trip, then runs them batch_concurrency at a time (one at a time by default). with_batch_size(n) enables batch mode; with_batch_size(1) claims one job per poll as without batch mode and only adds batch statistics. The [worker] settings batch_size and batch_concurrency (WorkerConfig) set the same from configuration:

# #[allow(unused_imports)] use hammerwork::queue::DatabaseQueue;
# #[allow(unused_variables, unused_mut, dead_code, unreachable_code)]
# async fn doc(queue: std::sync::Arc<hammerwork::JobQueue<sqlx::Postgres>>, handler: hammerwork::worker::JobHandler, jobs: Vec<hammerwork::Job>, worker: hammerwork::Worker<sqlx::Postgres>, batch_id: hammerwork::BatchId) -> std::result::Result<(), Box<dyn std::error::Error>> {
use hammerwork::{Job, Worker, worker::JobHandler};
use std::{sync::Arc, time::Duration};

let handler: JobHandler = Arc::new(|job: Job| {
    Box::pin(async move {
        // Process job
        Ok(())
    })
});

// Claim up to 50 jobs per poll and run up to 8 of them at once.
let worker = Worker::new(queue, "batch_queue".to_string(), handler)
    .with_batch_size(50)
    .with_batch_concurrency(8)
    .with_poll_interval(Duration::from_millis(100));
assert!(worker.is_batch_processing_enabled());
# Ok(())
# }

Every claimed job gets the same guarantees as a job claimed on its own:

• One statement per claim. PostgreSQL claims the batch with a single UPDATE ... WHERE id IN (SELECT ... ORDER BY priority DESC, scheduled_at ASC LIMIT n FOR UPDATE SKIP LOCKED) RETURNING ...; MySQL locks the rows with SELECT ... FOR UPDATE SKIP LOCKED LIMIT n and claims them with one UPDATE in the same READ COMMITTED transaction (retried on deadlock). Concurrent workers never claim the same job.
• Same eligibility. Paused queues, unfinished dependencies and disabled recurring jobs are honoured, and each job's attempts is incremented.
• A lease per job, from the claim. Each job's lease is written by the claiming statement. While a job waits for its turn in the batch, the worker renews its lease with heartbeats every third of the lease duration, as it does for running jobs, so the stale job reaper never reclaims a held job while its worker is alive, however long the jobs ahead of it take. (Sizing the lease to cover the whole batch would not work: the worker cannot know how long the jobs ahead will run.)
• Per-job processing. Timeouts, retries, hooks, events, spawning, result storage, encrypted payloads and statistics work exactly as for single jobs.
• Throttle and rate limit. With ThrottleConfig::max_concurrent, every claimed job holds a permit until it finishes, so a worker claims no more jobs than there are free permits. With a rate limit, a batch claims no more jobs than there are tokens, and tokens for jobs it did not get are handed back.
• Pause. Before starting a claimed job, the worker checks whether the queue was paused (at most once per poll interval). If it was, the claimed jobs that have not started are released back to Pending as on shutdown; running jobs finish.
• Shutdown. Claimed jobs that have not started are released back to Pending at once (DatabaseQueue::release_job_run, which also takes back the claim's attempt), so other workers can run them right away. Running jobs get the shutdown grace period.

Priorities

Without priority weights, or with PriorityWeights::strict(), a batch is filled in strict priority order: highest priority first, then oldest. With weighted priorities, a priority level is picked by weight (as a single weighted claim picks one) and the batch is filled with that level's oldest jobs; if the level runs out first, another level is picked by weight among the remaining ones to fill the rest. Over many batches each level gets batches in proportion to its weight, and no batch comes back short while other levels have runnable jobs.

Choosing a batch size

Batch claims save database round trips, so they help most when jobs are short. A worker holds the jobs it claimed until it gets to them, so idle workers cannot take them: keep batch_size small relative to the queue depth when jobs are slow, or raise batch_concurrency so held jobs start sooner. In a local benchmark (PostgreSQL 16 and MySQL 8 on the same machine, no-op handler, 2,000 jobs) claiming in batches of 50 was about 5-6x faster than claiming one job at a time, and a single batch-mode worker with batch size 50 and concurrency 10 processed 3-4x more jobs per second than a single-job worker. Run HAMMERWORK_BENCH=1 cargo test --all-features --test batch_dequeue_tests -- --include-ignored --nocapture throughput to measure your own setup.

Claiming Batches Directly

The queue API claims batches too, for custom consumers:

# #[allow(unused_imports)] use hammerwork::queue::DatabaseQueue;
# #[allow(unused_variables, unused_mut, dead_code, unreachable_code)]
# async fn doc(queue: std::sync::Arc<hammerwork::JobQueue<sqlx::Postgres>>, handler: hammerwork::worker::JobHandler, jobs: Vec<hammerwork::Job>, worker: hammerwork::Worker<sqlx::Postgres>, batch_id: hammerwork::BatchId) -> std::result::Result<(), Box<dyn std::error::Error>> {
use hammerwork::{JobOutcome, queue::{DatabaseQueue, DequeueOptions}};
use std::time::Duration;

// Up to 20 jobs, strict priority order, each leased for 2 minutes.
let options = DequeueOptions::batch(20).with_lease_duration(Duration::from_secs(120));
let jobs = queue.dequeue("emails", &options).await?;
for job in jobs {
    let run = job.clone();
    let job = queue.decrypt_job(job).await?; // a no-op for unencrypted jobs
    // ... run it, renewing its lease with `heartbeat_job` while it runs or waits ...
    queue.finish_job_run(&run, JobOutcome::Completed).await?;
}
# Ok(())
# }

Jobs you claimed but will not run go back with queue.release_job_run(&job).

Access Batch Statistics

Workers with batch processing enabled also keep statistics for jobs that belong to a JobBatch:

# #[allow(unused_imports)] use hammerwork::queue::DatabaseQueue;
# #[allow(unused_variables, unused_mut, dead_code, unreachable_code)]
# async fn doc(queue: std::sync::Arc<hammerwork::JobQueue<sqlx::Postgres>>, handler: hammerwork::worker::JobHandler, jobs: Vec<hammerwork::Job>, worker: hammerwork::Worker<sqlx::Postgres>, batch_id: hammerwork::BatchId) -> std::result::Result<(), Box<dyn std::error::Error>> {
let stats = worker.get_batch_stats();
println!("Batch jobs processed: {}", stats.jobs_processed);
println!("Batch success rate: {:.1}%", stats.success_rate() * 100.0);
println!("Average processing time: {:.1}ms", stats.average_processing_time_ms);
# Ok(())
# }

Monitoring Batch Progress

Check Batch Status

get_batch_status returns a BatchResult:

# #[allow(unused_imports)] use hammerwork::queue::DatabaseQueue;
# #[allow(unused_variables, unused_mut, dead_code, unreachable_code)]
# async fn doc(queue: std::sync::Arc<hammerwork::JobQueue<sqlx::Postgres>>, handler: hammerwork::worker::JobHandler, jobs: Vec<hammerwork::Job>, worker: hammerwork::Worker<sqlx::Postgres>, batch_id: hammerwork::BatchId) -> std::result::Result<(), Box<dyn std::error::Error>> {
use hammerwork::queue::DatabaseQueue;

let batch_result = queue.get_batch_status(batch_id).await?;

println!("Status: {:?}", batch_result.status);
println!("Total jobs: {}", batch_result.total_jobs);
println!("Pending: {}", batch_result.pending_jobs);
println!("Completed: {}", batch_result.completed_jobs);
println!("Failed: {}", batch_result.failed_jobs);
println!("Success rate: {:.1}%", batch_result.success_rate() * 100.0);
# Ok(())
# }

List Batch Jobs

# #[allow(unused_imports)] use hammerwork::queue::DatabaseQueue;
# #[allow(unused_variables, unused_mut, dead_code, unreachable_code)]
# async fn doc(queue: std::sync::Arc<hammerwork::JobQueue<sqlx::Postgres>>, handler: hammerwork::worker::JobHandler, jobs: Vec<hammerwork::Job>, worker: hammerwork::Worker<sqlx::Postgres>, batch_id: hammerwork::BatchId) -> std::result::Result<(), Box<dyn std::error::Error>> {
use hammerwork::queue::DatabaseQueue;

let batch_jobs = queue.get_batch_jobs(batch_id).await?;

for job in batch_jobs {
    println!("Job {}: Status={:?}, Priority={:?}", job.id, job.status, job.priority);
}
# Ok(())
# }

Monitor Completion

# #[allow(unused_imports)] use hammerwork::queue::DatabaseQueue;
# #[allow(unused_variables, unused_mut, dead_code, unreachable_code)]
# async fn doc(queue: std::sync::Arc<hammerwork::JobQueue<sqlx::Postgres>>, handler: hammerwork::worker::JobHandler, jobs: Vec<hammerwork::Job>, worker: hammerwork::Worker<sqlx::Postgres>, batch_id: hammerwork::BatchId) -> std::result::Result<(), Box<dyn std::error::Error>> {
use hammerwork::queue::DatabaseQueue;
use std::time::Duration;

loop {
    let batch_result = queue.get_batch_status(batch_id).await?;

    if batch_result.pending_jobs == 0 {
        println!("Batch finished with {:.1}% success", batch_result.success_rate() * 100.0);
        break;
    }

    tokio::time::sleep(Duration::from_secs(1)).await;
}
# Ok(())
# }

CLI Batch Commands

cargo hammerwork batch works on jobs in bulk (it does not create JobBatch records):

# Enqueue jobs from a JSON-lines file
cargo hammerwork batch enqueue --file jobs.jsonl --queue emails --progress-every 500 --continue-on-error

# Retry failed or dead jobs matching criteria (preview first with --dry-run)
cargo hammerwork batch retry --queue emails --status failed --failed-since-hours 24 --dry-run
cargo hammerwork batch retry --queue emails --status failed --failed-since-hours 24 --confirm

# Cancel pending jobs older than a day
cargo hammerwork batch cancel --queue emails --status pending --older-than-hours 24 --confirm

# Export jobs to a file (csv, json or jsonl)
cargo hammerwork batch export --output jobs.csv --queue emails --format csv

Best Practices

1. Choose Appropriate Batch Sizes

Smaller chunks give faster feedback and shorter transactions; larger ones mean fewer round trips. Chunk with into_chunks:

# #[allow(unused_imports)] use hammerwork::queue::DatabaseQueue;
# #[allow(unused_variables, unused_mut, dead_code, unreachable_code)]
# async fn doc(queue: std::sync::Arc<hammerwork::JobQueue<sqlx::Postgres>>, handler: hammerwork::worker::JobHandler, jobs: Vec<hammerwork::Job>, worker: hammerwork::Worker<sqlx::Postgres>, batch_id: hammerwork::BatchId) -> std::result::Result<(), Box<dyn std::error::Error>> {
use hammerwork::batch::JobBatch;

let small = JobBatch::new("quick_tasks").with_jobs(jobs.clone()).with_batch_size(10);
let large = JobBatch::new("bulk_import").with_jobs(jobs).with_batch_size(1000);
# Ok(())
# }

2. Use Metadata for Tracking

# #[allow(unused_imports)] use hammerwork::queue::DatabaseQueue;
# #[allow(unused_variables, unused_mut, dead_code, unreachable_code)]
# async fn doc(queue: std::sync::Arc<hammerwork::JobQueue<sqlx::Postgres>>, handler: hammerwork::worker::JobHandler, jobs: Vec<hammerwork::Job>, worker: hammerwork::Worker<sqlx::Postgres>, batch_id: hammerwork::BatchId) -> std::result::Result<(), Box<dyn std::error::Error>> {
use hammerwork::batch::JobBatch;

let batch = JobBatch::new("user_notifications")
    .with_metadata("run_id", uuid::Uuid::new_v4().to_string())
    .with_metadata("triggered_by", "automated_system")
    .with_metadata("timestamp", chrono::Utc::now().to_rfc3339());
# Ok(())
# }

3. Handle Large Batches

A batch is limited to 10,000 jobs. For more, split at the application level:

# #[allow(unused_imports)] use hammerwork::queue::DatabaseQueue;
# #[allow(unused_variables, unused_mut, dead_code, unreachable_code)]
# async fn doc(queue: std::sync::Arc<hammerwork::JobQueue<sqlx::Postgres>>, handler: hammerwork::worker::JobHandler, jobs: Vec<hammerwork::Job>, worker: hammerwork::Worker<sqlx::Postgres>, batch_id: hammerwork::BatchId) -> std::result::Result<(), Box<dyn std::error::Error>> {
use hammerwork::{batch::JobBatch, queue::DatabaseQueue, Job};
use serde_json::json;

let all_jobs: Vec<Job> = (0..25_000)
    .map(|i| Job::new("bulk".to_string(), json!({"n": i})))
    .collect();

for (i, chunk) in all_jobs.chunks(5000).enumerate() {
    let batch = JobBatch::new(format!("large_batch_part_{}", i))
        .with_jobs(chunk.to_vec())
        .with_metadata("parent_batch", "large_batch")
        .with_metadata("part", i.to_string());

    queue.enqueue_batch(batch).await?;
}
# Ok(())
# }

4. Monitor Batch Performance

# #[allow(unused_imports)] use hammerwork::queue::DatabaseQueue;
# #[allow(unused_variables, unused_mut, dead_code, unreachable_code)]
# async fn doc(queue: std::sync::Arc<hammerwork::JobQueue<sqlx::Postgres>>, handler: hammerwork::worker::JobHandler, jobs: Vec<hammerwork::Job>, worker: hammerwork::Worker<sqlx::Postgres>, batch_id: hammerwork::BatchId) -> std::result::Result<(), Box<dyn std::error::Error>> {
use hammerwork::stats::{InMemoryStatsCollector, StatisticsCollector};
use std::{sync::Arc, time::Duration};

let stats_collector = Arc::new(InMemoryStatsCollector::new_default());
// ... pass it to workers with `with_stats_collector` ...

let stats = stats_collector
    .get_queue_statistics("batch_queue", Duration::from_secs(3600))
    .await?;

println!("Jobs processed: {}", stats.total_processed);
println!("Error rate: {:.1}%", stats.error_rate * 100.0);
# Ok(())
# }

5. Clean Up Finished Batches

delete_batch removes the batch row and all of its jobs:

# #[allow(unused_imports)] use hammerwork::queue::DatabaseQueue;
# #[allow(unused_variables, unused_mut, dead_code, unreachable_code)]
# async fn doc(queue: std::sync::Arc<hammerwork::JobQueue<sqlx::Postgres>>, handler: hammerwork::worker::JobHandler, jobs: Vec<hammerwork::Job>, worker: hammerwork::Worker<sqlx::Postgres>, batch_id: hammerwork::BatchId) -> std::result::Result<(), Box<dyn std::error::Error>> {
use hammerwork::queue::DatabaseQueue;

let batch_result = queue.get_batch_status(batch_id).await?;
if batch_result.pending_jobs == 0 {
    queue.delete_batch(batch_id).await?;
}
# Ok(())
# }

Example: Complete Batch Processing Workflow

use hammerwork::{
    batch::{JobBatch, PartialFailureMode},
    queue::DatabaseQueue,
    worker::JobHandler,
    Job, JobQueue, Worker, WorkerPool,
};
use serde_json::json;
use std::{sync::Arc, time::Duration};

#[tokio::main]
async fn main() -> Result<(), Box<dyn std::error::Error>> {
    // Setup (run the migrations first, e.g. with `cargo hammerwork migration run`)
    let pool = sqlx::PgPool::connect("postgresql://localhost/hammerwork").await?;
    let queue = Arc::new(JobQueue::new(pool));

    // Create a batch of data processing jobs
    let jobs: Vec<Job> = (0..100)
        .map(|i| {
            Job::new(
                "data_processing".to_string(),
                json!({
                    "file_id": format!("file_{}", i),
                    "operation": "transform"
                }),
            )
        })
        .collect();

    let batch = JobBatch::new("daily_data_processing")
        .with_jobs(jobs)
        .with_partial_failure_handling(PartialFailureMode::ContinueOnError)
        .with_metadata("date", "2024-01-15")
        .with_metadata("source", "automated_pipeline");

    let batch_id = queue.enqueue_batch(batch).await?;
    println!("Enqueued batch: {}", batch_id);

    // Worker in batch mode: claims up to 20 jobs per poll
    let handler: JobHandler = Arc::new(|_job: Job| {
        Box::pin(async move {
            tokio::time::sleep(Duration::from_millis(100)).await;
            Ok(())
        })
    });

    let worker = Worker::new(queue.clone(), "data_processing".to_string(), handler)
        .with_batch_size(20);

    let mut worker_pool = WorkerPool::new();
    worker_pool.add_worker(worker);

    tokio::spawn(async move {
        if let Err(e) = worker_pool.start().await {
            eprintln!("Worker pool stopped: {}", e);
        }
    });

    // Monitor progress
    loop {
        tokio::time::sleep(Duration::from_secs(2)).await;

        let status = queue.get_batch_status(batch_id).await?;
        println!(
            "Progress: {}/{} jobs completed",
            status.completed_jobs, status.total_jobs
        );

        if status.pending_jobs == 0 {
            println!(
                "Batch finished with {:.1}% success rate",
                status.success_rate() * 100.0
            );
            break;
        }
    }

    // Cleanup (removes the batch and its jobs)
    queue.delete_batch(batch_id).await?;

    Ok(())
}

Performance Considerations

Database-Specific Optimizations

PostgreSQL inserts a batch's jobs inside one transaction using bulk statements, and MySQL uses multi-row INSERT ... VALUES statements. Either way a batch costs a handful of statements instead of one per job.

Network Overhead

• Individual enqueue: 1000 jobs = 1000 round trips
• Batch enqueue: 1000 jobs = a few statements in one transaction

Memory Usage

The whole batch is held in memory while it is built and inserted. Keep batches moderate (a few thousand jobs) and use into_chunks or application-level chunking for more.

Troubleshooting

Batch Validation Errors

enqueue_batch returns HammerworkError::Queue when validation fails: an empty batch, more than 10,000 jobs, or jobs with different queue names. Call batch.validate() yourself to check ahead of time.

Performance Issues

If batch processing is slow:

1. Check the size of the batches you enqueue
2. Monitor database performance
3. Make sure workers poll the right queue and have enough capacity
4. Review job handler efficiency
5. Consider more workers in a WorkerPool