hammerwork 2.3.0

//! End-to-end tests for workers and worker pools built from configuration.
//!
//! Each scenario builds a worker (or pool) from a `HammerworkConfig` or the worker
//! builder methods, runs it against a real database and checks that it behaves as
//! configured: pool size, job timeout, retry strategy, priority weights, throttles,
//! autoscaling, pausing, error backoff and job hooks. Each scenario is written once and
//! run against both backends.

#![cfg(any(feature = "postgres", feature = "mysql"))]

mod test_utils;

use chrono::Utc;
use hammerwork::{
    HammerworkConfig, HammerworkError, Job, JobId, JobQueue, JobStatus, Worker, WorkerPool,
    priority::{JobPriority, PriorityWeights},
    queue::DatabaseQueue,
    rate_limit::ThrottleConfig,
    retry::RetryStrategy,
    worker::{JobHandler, JobHookEvent},
};
use serde_json::json;
use std::{
    future::Future,
    sync::{
        Arc, Mutex,
        atomic::{AtomicUsize, Ordering},
    },
    time::{Duration, Instant},
};
use tokio::sync::mpsc;

/// Wait until every job in `ids` matches `predicate`.
async fn wait_for_jobs<DB, P>(
    queue: &Arc<JobQueue<DB>>,
    ids: &[JobId],
    timeout: Duration,
    predicate: P,
) -> bool
where
    DB: sqlx::Database,
    JobQueue<DB>: DatabaseQueue<Database = DB> + Send + Sync + 'static,
    P: Fn(&Job) -> bool,
{
    test_utils::eventually(timeout, || async {
        for id in ids {
            match queue.get_job(*id).await.unwrap() {
                Some(job) if predicate(&job) => {}
                _ => return false,
            }
        }
        true
    })
    .await
}

fn completed(job: &Job) -> bool {
    job.status == JobStatus::Completed
}

/// Run `pool` until `until` finishes, then shut it down gracefully.
async fn run_pool_until<DB, Fut>(pool: &mut WorkerPool<DB>, until: Fut)
where
    DB: sqlx::Database + Send + Sync + 'static,
    JobQueue<DB>: DatabaseQueue<Database = DB> + Send + Sync + 'static,
    Fut: Future<Output = ()>,
{
    tokio::select! {
        result = pool.start() => panic!("pool stopped early: {result:?}"),
        () = until => {}
    }
    pool.shutdown().await.unwrap();
}

/// Counts handlers running at once and remembers the highest count.
#[derive(Clone, Default)]
struct Concurrency {
    running: Arc<AtomicUsize>,
    peak: Arc<AtomicUsize>,
}

impl Concurrency {
    /// A handler that sleeps for `work` while counted as running.
    fn handler(&self, work: Duration) -> JobHandler {
        let this = self.clone();
        Arc::new(move |_job: Job| {
            let this = this.clone();
            Box::pin(async move {
                let now = this.running.fetch_add(1, Ordering::SeqCst) + 1;
                this.peak.fetch_max(now, Ordering::SeqCst);
                tokio::time::sleep(work).await;
                this.running.fetch_sub(1, Ordering::SeqCst);
                Ok(())
            })
        })
    }

    fn peak(&self) -> usize {
        self.peak.load(Ordering::SeqCst)
    }

    fn running(&self) -> usize {
        self.running.load(Ordering::SeqCst)
    }

    fn reset_peak(&self) {
        self.peak.store(self.running(), Ordering::SeqCst);
    }
}

async fn enqueue_many<DB>(queue: &Arc<JobQueue<DB>>, queue_name: &str, count: usize) -> Vec<JobId>
where
    DB: sqlx::Database,
    JobQueue<DB>: DatabaseQueue<Database = DB> + Send + Sync + 'static,
{
    let mut ids = Vec::new();
    for i in 0..count {
        ids.push(
            queue
                .enqueue(Job::new(queue_name.to_string(), json!({ "i": i })))
                .await
                .unwrap(),
        );
    }
    ids
}

async fn delete_jobs<DB>(queue: &Arc<JobQueue<DB>>, ids: &[JobId])
where
    DB: sqlx::Database,
    JobQueue<DB>: DatabaseQueue<Database = DB> + Send + Sync + 'static,
{
    for id in ids {
        queue.delete_job(*id).await.unwrap();
    }
}

/// A pool built with `WorkerPool::from_hammerwork_config` runs `worker.pool_size`
/// workers at once and no more.
async fn pool_runs_pool_size_workers<DB>(queue: Arc<JobQueue<DB>>)
where
    DB: sqlx::Database + Send + Sync + 'static,
    JobQueue<DB>: DatabaseQueue<Database = DB> + Send + Sync + 'static,
{
    let queue_name = test_utils::unique_queue("cfg_pool_size");
    let ids = enqueue_many(&queue, &queue_name, 9).await;

    let mut config = HammerworkConfig::new();
    config.worker.pool_size = 3;
    config.worker.polling_interval = Duration::from_millis(20);
    let concurrency = Concurrency::default();
    let worker = Worker::new(
        Arc::clone(&queue),
        queue_name.clone(),
        concurrency.handler(Duration::from_millis(300)),
    )
    .with_hammerwork_config(&config);
    let mut pool = WorkerPool::from_hammerwork_config(worker, &config)
        .unwrap()
        .without_stale_job_reaper();

    let started = Instant::now();
    run_pool_until(&mut pool, async {
        assert!(wait_for_jobs(&queue, &ids, Duration::from_secs(30), completed).await);
    })
    .await;
    let elapsed = started.elapsed();
    assert_eq!(concurrency.peak(), 3, "pool_size workers run at once");
    // 9 jobs of 300ms on 3 workers take at least 3 rounds.
    assert!(elapsed >= Duration::from_millis(850), "took {elapsed:?}");
    delete_jobs(&queue, &ids).await;
}

/// `worker.job_timeout` and `worker.retry_strategy` from the configuration apply to
/// jobs without their own timeout or strategy; a job's own settings win.
async fn config_timeout_and_retry_strategy_apply<DB>(queue: Arc<JobQueue<DB>>)
where
    DB: sqlx::Database + Send + Sync + 'static,
    JobQueue<DB>: DatabaseQueue<Database = DB> + Send + Sync + 'static,
{
    let queue_name = test_utils::unique_queue("cfg_timeout_retry");
    let slow = queue
        .enqueue(Job::new(queue_name.clone(), json!({ "action": "sleep" })).with_max_attempts(3))
        .await
        .unwrap();
    let failing = queue
        .enqueue(Job::new(queue_name.clone(), json!({ "action": "fail" })).with_max_attempts(3))
        .await
        .unwrap();
    // Its own strategy (a 2 minute delay) overrides the worker default (1 hour).
    let own_strategy = queue
        .enqueue(
            Job::new(queue_name.clone(), json!({ "action": "fail" }))
                .with_max_attempts(3)
                .with_retry_strategy(RetryStrategy::fixed(Duration::from_secs(120))),
        )
        .await
        .unwrap();

    let mut config = HammerworkConfig::new();
    config.worker.pool_size = 1;
    config.worker.polling_interval = Duration::from_millis(20);
    config.worker.job_timeout = Duration::from_millis(200);
    config.worker.retry_strategy = RetryStrategy::fixed(Duration::from_secs(3600));

    let handler: JobHandler = Arc::new(|job: Job| {
        Box::pin(async move {
            match job.payload["action"].as_str() {
                Some("sleep") => {
                    tokio::time::sleep(Duration::from_secs(10)).await;
                    Ok(())
                }
                _ => Err(HammerworkError::Worker {
                    message: "configured failure".to_string(),
                }),
            }
        })
    });
    let worker = Worker::new(Arc::clone(&queue), queue_name.clone(), handler)
        .with_hammerwork_config(&config);
    let mut pool = WorkerPool::from_hammerwork_config(worker, &config)
        .unwrap()
        .without_stale_job_reaper();

    let before = Utc::now();
    let ids = [slow, failing, own_strategy];
    run_pool_until(&mut pool, async {
        assert!(
            wait_for_jobs(&queue, &ids, Duration::from_secs(20), |job| job.attempts
                == 1
                && job.status != JobStatus::Running)
            .await
        );
    })
    .await;

    let slow = queue.get_job(slow).await.unwrap().unwrap();
    assert_eq!(slow.status, JobStatus::Pending, "rescheduled for a retry");
    let error = slow.error_message.clone().unwrap_or_default();
    assert!(error.contains("timed out after 200ms"), "error: {error}");
    let delay = slow.scheduled_at - before;
    assert!(
        delay > chrono::Duration::minutes(59) && delay < chrono::Duration::minutes(61),
        "a timed-out job is retried with the configured strategy, got {delay}"
    );

    let failing = queue.get_job(failing).await.unwrap().unwrap();
    assert_eq!(
        failing.status,
        JobStatus::Pending,
        "rescheduled for a retry"
    );
    let delay = failing.scheduled_at - before;
    assert!(
        delay > chrono::Duration::minutes(59) && delay < chrono::Duration::minutes(61),
        "a failed job is retried with the configured strategy, got {delay}"
    );

    let own = queue.get_job(own_strategy).await.unwrap().unwrap();
    let delay = own.scheduled_at - before;
    assert!(
        delay > chrono::Duration::seconds(110) && delay < chrono::Duration::seconds(130),
        "the job's own strategy wins, got {delay}"
    );
    delete_jobs(&queue, &ids).await;
}

/// `worker.priority_weights = strict` makes the worker always take the highest
/// priority job first.
async fn config_strict_priority_orders_jobs<DB>(queue: Arc<JobQueue<DB>>)
where
    DB: sqlx::Database + Send + Sync + 'static,
    JobQueue<DB>: DatabaseQueue<Database = DB> + Send + Sync + 'static,
{
    let queue_name = test_utils::unique_queue("cfg_strict_priority");
    let mut ids = Vec::new();
    for priority in [
        JobPriority::Background,
        JobPriority::Low,
        JobPriority::Normal,
        JobPriority::High,
        JobPriority::Critical,
    ] {
        ids.push(
            queue
                .enqueue(
                    Job::new(
                        queue_name.clone(),
                        json!({ "priority": priority.to_string() }),
                    )
                    .with_priority(priority),
                )
                .await
                .unwrap(),
        );
    }

    let mut config = HammerworkConfig::new();
    config.worker.pool_size = 1;
    config.worker.polling_interval = Duration::from_millis(20);
    config.worker.priority_weights = PriorityWeights::strict();

    let order = Arc::new(Mutex::new(Vec::new()));
    let seen = Arc::clone(&order);
    let handler: JobHandler = Arc::new(move |job: Job| {
        seen.lock().unwrap().push(job.priority);
        Box::pin(async { Ok(()) })
    });
    let worker = Worker::new(Arc::clone(&queue), queue_name.clone(), handler)
        .with_hammerwork_config(&config);
    let mut pool = WorkerPool::from_hammerwork_config(worker, &config)
        .unwrap()
        .without_stale_job_reaper();
    run_pool_until(&mut pool, async {
        assert!(wait_for_jobs(&queue, &ids, Duration::from_secs(20), completed).await);
    })
    .await;

    assert_eq!(
        *order.lock().unwrap(),
        vec![
            JobPriority::Critical,
            JobPriority::High,
            JobPriority::Normal,
            JobPriority::Low,
            JobPriority::Background,
        ]
    );
    delete_jobs(&queue, &ids).await;
}

/// A queue throttle's `max_concurrent` caps how many jobs a whole pool runs at once.
async fn throttle_max_concurrent_caps_the_pool<DB>(queue: Arc<JobQueue<DB>>)
where
    DB: sqlx::Database + Send + Sync + 'static,
    JobQueue<DB>: DatabaseQueue<Database = DB> + Send + Sync + 'static,
{
    let queue_name = test_utils::unique_queue("cfg_max_concurrent");
    let ids = enqueue_many(&queue, &queue_name, 8).await;

    let mut config = HammerworkConfig::new();
    config.worker.pool_size = 4;
    config.worker.polling_interval = Duration::from_millis(20);
    config.rate_limiting.enabled = true;
    config
        .rate_limiting
        .queue_throttles
        .insert(queue_name.clone(), ThrottleConfig::new().max_concurrent(2));

    let concurrency = Concurrency::default();
    let worker = Worker::new(
        Arc::clone(&queue),
        queue_name.clone(),
        concurrency.handler(Duration::from_millis(200)),
    )
    .with_hammerwork_config(&config);
    let mut pool = WorkerPool::from_hammerwork_config(worker, &config)
        .unwrap()
        .without_stale_job_reaper();
    run_pool_until(&mut pool, async {
        assert!(wait_for_jobs(&queue, &ids, Duration::from_secs(30), completed).await);
    })
    .await;
    assert_eq!(
        concurrency.peak(),
        2,
        "4 workers, but the throttle allows 2 jobs at once"
    );
    delete_jobs(&queue, &ids).await;
}

/// A queue throttle's `rate_per_minute` limits how fast a pool takes jobs. The bucket
/// holds a minute's worth of tokens, so the first `rate` jobs run at once and the rest
/// at the configured rate.
async fn throttle_rate_limits_the_pool<DB>(queue: Arc<JobQueue<DB>>)
where
    DB: sqlx::Database + Send + Sync + 'static,
    JobQueue<DB>: DatabaseQueue<Database = DB> + Send + Sync + 'static,
{
    let queue_name = test_utils::unique_queue("cfg_rate_limit");
    // 60 per minute: 60 tokens up front, then one per second.
    let ids = enqueue_many(&queue, &queue_name, 62).await;

    let mut config = HammerworkConfig::new();
    config.worker.pool_size = 2;
    config.worker.polling_interval = Duration::from_millis(20);
    config.rate_limiting.enabled = true;
    config.rate_limiting.default_throttle = ThrottleConfig::new().rate_per_minute(60);

    let concurrency = Concurrency::default();
    let worker = Worker::new(
        Arc::clone(&queue),
        queue_name.clone(),
        concurrency.handler(Duration::ZERO),
    )
    .with_hammerwork_config(&config);
    let mut pool = WorkerPool::from_hammerwork_config(worker, &config)
        .unwrap()
        .without_stale_job_reaper();

    let started = Instant::now();
    run_pool_until(&mut pool, async {
        assert!(wait_for_jobs(&queue, &ids, Duration::from_secs(60), completed).await);
    })
    .await;
    let elapsed = started.elapsed();
    // Jobs 61 and 62 each wait for a new token (one per second).
    assert!(
        elapsed >= Duration::from_millis(1500),
        "62 jobs at 60/minute finished in {elapsed:?}"
    );
    delete_jobs(&queue, &ids).await;
}

/// #64 M6: polls that claim no job (empty or paused queue) do not spend the rate limit,
/// so a burst is available when jobs arrive.
async fn rate_limit_is_not_spent_on_empty_polls<DB>(queue: Arc<JobQueue<DB>>)
where
    DB: sqlx::Database + Send + Sync + 'static,
    JobQueue<DB>: DatabaseQueue<Database = DB> + Send + Sync + 'static,
{
    let queue_name = test_utils::unique_queue("cfg_rate_idle");
    let concurrency = Concurrency::default();
    // Three jobs an hour: after the burst, the next token takes 20 minutes.
    let worker = Worker::new(
        Arc::clone(&queue),
        queue_name.clone(),
        concurrency.handler(Duration::ZERO),
    )
    .with_poll_interval(Duration::from_millis(10))
    .with_rate_limit(hammerwork::RateLimit::per_hour(3));
    let mut pool = WorkerPool::new().without_stale_job_reaper();
    pool.add_worker(worker);

    run_pool_until(&mut pool, async {
        // Dozens of empty polls, then a paused stretch.
        tokio::time::sleep(Duration::from_millis(300)).await;
        queue.pause_queue(&queue_name, Some("test")).await.unwrap();
        tokio::time::sleep(Duration::from_millis(200)).await;
        queue.resume_queue(&queue_name, Some("test")).await.unwrap();

        // The whole burst is still there.
        let ids = enqueue_many(&queue, &queue_name, 3).await;
        assert!(
            wait_for_jobs(&queue, &ids, Duration::from_secs(10), completed).await,
            "idle polls spent the rate limit"
        );
        delete_jobs(&queue, &ids).await;
    })
    .await;
}

/// An autoscaling pool starts workers from its template while the queue is deep (up
/// to `max_workers`) and retires them down to `min_workers` once it drains.
async fn autoscaling_adds_and_retires_workers<DB>(queue: Arc<JobQueue<DB>>)
where
    DB: sqlx::Database + Send + Sync + 'static,
    JobQueue<DB>: DatabaseQueue<Database = DB> + Send + Sync + 'static,
{
    let queue_name = test_utils::unique_queue("cfg_autoscale");
    let ids = enqueue_many(&queue, &queue_name, 24).await;

    let concurrency = Concurrency::default();
    let worker = Worker::new(
        Arc::clone(&queue),
        queue_name.clone(),
        concurrency.handler(Duration::from_millis(150)),
    )
    .with_poll_interval(Duration::from_millis(20));
    let config = hammerwork::config::WorkerConfig {
        pool_size: 1,
        autoscaling_enabled: true,
        min_workers: 1,
        max_workers: 3,
        ..Default::default()
    };
    let mut pool = WorkerPool::from_config(worker, &config)
        .with_autoscaling(
            config
                .autoscale_config()
                .with_scale_up_threshold(2)
                .with_scale_down_threshold(1)
                .with_cooldown_period(Duration::ZERO)
                .with_evaluation_window(Duration::from_millis(100)),
        )
        .without_stale_job_reaper();

    tokio::select! {
        result = pool.start() => panic!("pool stopped early: {result:?}"),
        () = async {
            assert!(wait_for_jobs(&queue, &ids, Duration::from_secs(30), completed).await);
        } => {}
    }
    assert_eq!(
        concurrency.peak(),
        3,
        "the pool grows from 1 to max_workers (3) and no further"
    );

    // With the queue empty it scales back down to min_workers.
    let scaled_down = test_utils::eventually(Duration::from_secs(10), || async {
        pool.get_autoscale_metrics().active_workers == 1
    })
    .await;
    assert!(scaled_down, "metrics: {:?}", pool.get_autoscale_metrics());
    let metrics = pool.get_autoscale_metrics();
    assert_eq!(metrics.current_queue_depth, 0);
    assert!(metrics.last_scale_time.is_some());

    // The retired workers stopped: two jobs (2 per worker, not above the scale-up
    // threshold) now run one at a time.
    concurrency.reset_peak();
    let more = enqueue_many(&queue, &queue_name, 2).await;
    assert!(wait_for_jobs(&queue, &more, Duration::from_secs(10), completed).await);
    assert_eq!(concurrency.peak(), 1, "only min_workers remain");

    // A new backlog scales the pool up again (reusing the retired workers' slots).
    concurrency.reset_peak();
    let again = enqueue_many(&queue, &queue_name, 24).await;
    assert!(wait_for_jobs(&queue, &again, Duration::from_secs(30), completed).await);
    assert_eq!(concurrency.peak(), 3, "scaled up to max_workers again");

    pool.shutdown().await.unwrap();
    delete_jobs(&queue, &ids).await;
    delete_jobs(&queue, &more).await;
    delete_jobs(&queue, &again).await;
}

/// #64 H9: autoscaling scales only the template's queue. Workers of other queues are
/// never retired (even when the scaled queue is idle) and never cloned.
async fn autoscaling_only_scales_the_template_queue<DB>(queue: Arc<JobQueue<DB>>)
where
    DB: sqlx::Database + Send + Sync + 'static,
    JobQueue<DB>: DatabaseQueue<Database = DB> + Send + Sync + 'static,
{
    let scaled = test_utils::unique_queue("cfg_scaled");
    let others = [
        test_utils::unique_queue("cfg_other_a"),
        test_utils::unique_queue("cfg_other_b"),
    ];
    let scaled_load = Concurrency::default();
    let other_load = Concurrency::default();
    let worker = |name: &str, load: &Concurrency| {
        Worker::new(
            Arc::clone(&queue),
            name.to_string(),
            load.handler(Duration::from_millis(150)),
        )
        .with_poll_interval(Duration::from_millis(20))
    };
    let mut pool = WorkerPool::new()
        .with_autoscaling(
            hammerwork::worker::AutoscaleConfig::new()
                .with_min_workers(1)
                .with_max_workers(3)
                .with_scale_up_threshold(2)
                .with_scale_down_threshold(1)
                .with_cooldown_period(Duration::ZERO)
                .with_evaluation_window(Duration::from_millis(100)),
        )
        .without_stale_job_reaper();
    // The first worker is the template; the scaled queue starts with two workers.
    pool.add_worker(worker(&scaled, &scaled_load));
    pool.add_worker(worker(&others[0], &other_load));
    pool.add_worker(worker(&scaled, &scaled_load));
    pool.add_worker(worker(&others[1], &other_load));

    // Run the pool for many evaluations while the scaled queue is idle, then drop the
    // `start` future: the workers keep running in their own tasks.
    tokio::select! {
        result = pool.start() => panic!("pool stopped early: {result:?}"),
        () = tokio::time::sleep(Duration::from_millis(1500)) => {}
    }
    // The idle scaled queue shrank to min_workers...
    assert_eq!(
        pool.get_autoscale_metrics().active_workers,
        1,
        "metrics: {:?}",
        pool.get_autoscale_metrics()
    );
    // ...and both other queues are still served.
    let mut ids = Vec::new();
    for name in &others {
        ids.extend(enqueue_many(&queue, name, 2).await);
    }
    assert!(
        wait_for_jobs(&queue, &ids, Duration::from_secs(10), completed).await,
        "a worker of another queue was retired"
    );
    delete_jobs(&queue, &ids).await;

    // A backlog on the scaled queue grows it to max_workers with clones of the template.
    let backlog = enqueue_many(&queue, &scaled, 24).await;
    assert!(wait_for_jobs(&queue, &backlog, Duration::from_secs(30), completed).await);
    assert_eq!(scaled_load.peak(), 3, "scaled up to max_workers");
    assert!(other_load.peak() <= 2, "the other queues were not scaled");
    delete_jobs(&queue, &backlog).await;

    pool.shutdown().await.unwrap();
}

/// A worker leaves a paused queue alone and picks its jobs up once it is resumed.
async fn worker_waits_while_queue_is_paused<DB>(queue: Arc<JobQueue<DB>>)
where
    DB: sqlx::Database + Send + Sync + 'static,
    JobQueue<DB>: DatabaseQueue<Database = DB> + Send + Sync + 'static,
{
    let queue_name = test_utils::unique_queue("cfg_paused");
    queue.pause_queue(&queue_name, Some("test")).await.unwrap();
    let ids = enqueue_many(&queue, &queue_name, 2).await;

    let handled = Arc::new(AtomicUsize::new(0));
    let count = Arc::clone(&handled);
    let handler: JobHandler = Arc::new(move |_job: Job| {
        count.fetch_add(1, Ordering::SeqCst);
        Box::pin(async { Ok(()) })
    });
    let worker = Worker::new(Arc::clone(&queue), queue_name.clone(), handler)
        .with_poll_interval(Duration::from_millis(20));
    let (shutdown_tx, shutdown_rx) = mpsc::channel(1);
    let task = tokio::spawn(async move { worker.run(shutdown_rx).await });

    tokio::time::sleep(Duration::from_millis(400)).await;
    assert_eq!(
        handled.load(Ordering::SeqCst),
        0,
        "paused queue is not processed"
    );
    for id in &ids {
        assert_eq!(
            queue.get_job(*id).await.unwrap().unwrap().status,
            JobStatus::Pending
        );
    }

    queue.resume_queue(&queue_name, Some("test")).await.unwrap();
    assert!(wait_for_jobs(&queue, &ids, Duration::from_secs(10), completed).await);
    assert_eq!(handled.load(Ordering::SeqCst), 2);

    shutdown_tx.send(()).await.unwrap();
    task.await.unwrap().unwrap();
    delete_jobs(&queue, &ids).await;
}

/// The job hooks registered with `on_job_*` fire for each outcome.
async fn job_hooks_fire_for_each_outcome<DB>(queue: Arc<JobQueue<DB>>)
where
    DB: sqlx::Database + Send + Sync + 'static,
    JobQueue<DB>: DatabaseQueue<Database = DB> + Send + Sync + 'static,
{
    let queue_name = test_utils::unique_queue("cfg_hooks");
    let ok = queue
        .enqueue(Job::new(queue_name.clone(), json!({ "action": "ok" })))
        .await
        .unwrap();
    let retried = queue
        .enqueue(Job::new(queue_name.clone(), json!({ "action": "fail" })).with_max_attempts(2))
        .await
        .unwrap();
    let failed = queue
        .enqueue(Job::new(queue_name.clone(), json!({ "action": "fail" })).with_max_attempts(1))
        .await
        .unwrap();
    let timed_out = queue
        .enqueue(
            Job::new(queue_name.clone(), json!({ "action": "sleep" }))
                .with_max_attempts(1)
                .with_timeout(Duration::from_millis(100)),
        )
        .await
        .unwrap();

    type Log = Arc<Mutex<Vec<(&'static str, JobId, Option<String>)>>>;
    let log: Log = Arc::default();
    let record = |name: &'static str, log: &Log| {
        let log = Arc::clone(log);
        move |event: JobHookEvent| {
            log.lock().unwrap().push((name, event.job.id, event.error));
        }
    };
    let handler: JobHandler = Arc::new(|job: Job| {
        Box::pin(async move {
            match job.payload["action"].as_str() {
                Some("ok") => Ok(()),
                Some("sleep") => {
                    tokio::time::sleep(Duration::from_secs(5)).await;
                    Ok(())
                }
                _ => Err(HammerworkError::Worker {
                    message: "hook failure".to_string(),
                }),
            }
        })
    });
    let worker = Worker::new(Arc::clone(&queue), queue_name.clone(), handler)
        .with_poll_interval(Duration::from_millis(20))
        .with_retry_delay(Duration::from_secs(3600))
        .on_job_start(record("start", &log))
        .on_job_complete(record("complete", &log))
        .on_job_fail(record("fail", &log))
        .on_job_retry(record("retry", &log))
        .on_job_timeout(record("timeout", &log));
    let (shutdown_tx, shutdown_rx) = mpsc::channel(1);
    let task = tokio::spawn(async move { worker.run(shutdown_rx).await });

    let ids = [ok, retried, failed, timed_out];
    assert!(
        wait_for_jobs(&queue, &ids, Duration::from_secs(20), |job| {
            job.attempts >= 1 && job.status != JobStatus::Running
        })
        .await
    );
    shutdown_tx.send(()).await.unwrap();
    task.await.unwrap().unwrap();

    let log = log.lock().unwrap().clone();
    let fired = |name: &str, id: JobId| log.iter().any(|(n, i, _)| *n == name && *i == id);
    for id in ids {
        assert!(fired("start", id), "start hook for every job: {log:?}");
    }
    assert!(fired("complete", ok));
    assert!(fired("retry", retried));
    assert!(
        !fired("fail", retried),
        "a retried job has not failed for good"
    );
    assert!(fired("fail", failed));
    assert!(fired("timeout", timed_out));
    assert!(!fired("complete", failed) && !fired("complete", timed_out));
    let retry_error = log
        .iter()
        .find(|(n, i, _)| *n == "retry" && *i == retried)
        .and_then(|(_, _, error)| error.clone())
        .unwrap_or_default();
    assert!(retry_error.contains("hook failure"), "error: {retry_error}");
    delete_jobs(&queue, &ids).await;
}

/// A worker whose database is unreachable keeps running (backing off between polls)
/// and still stops promptly when shut down during a long backoff.
async fn worker_backs_off_on_errors_and_stops_promptly<DB>(queue: Arc<JobQueue<DB>>)
where
    DB: sqlx::Database + Send + Sync + 'static,
    JobQueue<DB>: DatabaseQueue<Database = DB> + Send + Sync + 'static,
{
    let handled = Arc::new(AtomicUsize::new(0));
    let count = Arc::clone(&handled);
    let handler: JobHandler = Arc::new(move |_job: Job| {
        count.fetch_add(1, Ordering::SeqCst);
        Box::pin(async { Ok(()) })
    });
    let worker = Worker::new(
        Arc::clone(&queue),
        test_utils::unique_queue("cfg_backoff"),
        handler,
    )
    .with_poll_interval(Duration::from_millis(10))
    .with_throttle_config(ThrottleConfig::new().backoff_on_error(Duration::from_secs(60)));

    let (shutdown_tx, shutdown_rx) = mpsc::channel(1);
    let task = tokio::spawn(async move { worker.run(shutdown_rx).await });
    tokio::time::sleep(Duration::from_millis(300)).await;
    assert!(!task.is_finished(), "dequeue errors do not stop the worker");

    let stopping = Instant::now();
    shutdown_tx.send(()).await.unwrap();
    tokio::time::timeout(Duration::from_secs(5), task)
        .await
        .expect("shutdown interrupts the error backoff")
        .unwrap()
        .unwrap();
    assert!(stopping.elapsed() < Duration::from_secs(5));
    assert_eq!(handled.load(Ordering::SeqCst), 0);
}

/// A local HTTP endpoint that records the JSON body of every request it receives.
#[cfg(feature = "alerting")]
async fn capture_server() -> (
    String,
    tokio::sync::mpsc::UnboundedReceiver<serde_json::Value>,
) {
    use tokio::io::{AsyncReadExt, AsyncWriteExt};

    let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
    let url = format!("http://{}/alerts", listener.local_addr().unwrap());
    let (tx, rx) = tokio::sync::mpsc::unbounded_channel();
    tokio::spawn(async move {
        while let Ok((mut socket, _)) = listener.accept().await {
            let tx = tx.clone();
            tokio::spawn(async move {
                let mut buffer = Vec::new();
                let mut chunk = [0_u8; 4096];
                let body = loop {
                    let Ok(read) = socket.read(&mut chunk).await else {
                        return;
                    };
                    if read == 0 {
                        return;
                    }
                    buffer.extend_from_slice(&chunk[..read]);
                    let Some(end) = buffer.windows(4).position(|w| w == b"\r\n\r\n") else {
                        continue;
                    };
                    let headers = String::from_utf8_lossy(&buffer[..end]).to_lowercase();
                    let length: usize = headers
                        .lines()
                        .find_map(|line| line.strip_prefix("content-length:"))
                        .and_then(|value| value.trim().parse().ok())
                        .unwrap_or(0);
                    if buffer.len() >= end + 4 + length {
                        break buffer[end + 4..end + 4 + length].to_vec();
                    }
                };
                let _ = socket
                    .write_all(b"HTTP/1.1 200 OK\r\ncontent-length: 0\r\nconnection: close\r\n\r\n")
                    .await;
                if let Ok(json) = serde_json::from_slice(&body) {
                    let _ = tx.send(json);
                }
            });
        }
    });
    (url, rx)
}

/// Wait for an alert of `alert_type` on `rx`.
#[cfg(feature = "alerting")]
async fn next_alert(
    rx: &mut tokio::sync::mpsc::UnboundedReceiver<serde_json::Value>,
    alert_type: &str,
) -> serde_json::Value {
    tokio::time::timeout(Duration::from_secs(10), async {
        loop {
            let alert = rx.recv().await.expect("capture server stopped");
            if alert["alert_type"] == alert_type {
                return alert;
            }
        }
    })
    .await
    .unwrap_or_else(|_| panic!("no {alert_type} alert"))
}

/// A worker configured with alerting sends queue depth and worker starvation alerts
/// (from its monitoring task) to the webhook target.
#[cfg(feature = "alerting")]
async fn worker_sends_queue_depth_and_starvation_alerts<DB>(queue: Arc<JobQueue<DB>>)
where
    DB: sqlx::Database + Send + Sync + 'static,
    JobQueue<DB>: DatabaseQueue<Database = DB> + Send + Sync + 'static,
{
    use hammerwork::alerting::AlertingConfig;

    let (url, mut alerts) = capture_server().await;
    let noop: JobHandler = Arc::new(|_job: Job| Box::pin(async { Ok(()) }));

    // Queue depth: a paused queue with 3 jobs, threshold 2.
    let deep = test_utils::unique_queue("cfg_alert_depth");
    queue.pause_queue(&deep, None).await.unwrap();
    let ids = enqueue_many(&queue, &deep, 3).await;
    let mut config = HammerworkConfig::new();
    config.alerting = AlertingConfig::new().alert_on_queue_depth(2).webhook(&url);
    let worker = Worker::new(Arc::clone(&queue), deep.clone(), Arc::clone(&noop))
        .with_poll_interval(Duration::from_millis(20))
        .with_monitoring_interval(Duration::from_millis(50))
        .with_hammerwork_config(&config);
    let (shutdown_tx, shutdown_rx) = mpsc::channel(1);
    let task = tokio::spawn(async move { worker.run(shutdown_rx).await });
    let alert = next_alert(&mut alerts, "QueueDepthExceeded").await;
    assert_eq!(alert["queue_name"], deep.as_str());
    assert_eq!(alert["current_value"], 3.0);
    assert_eq!(alert["threshold"], 2.0);
    shutdown_tx.send(()).await.unwrap();
    task.await.unwrap().unwrap();
    queue.resume_queue(&deep, None).await.unwrap();
    delete_jobs(&queue, &ids).await;

    // Starvation: an empty queue and a 50ms threshold.
    let idle = test_utils::unique_queue("cfg_alert_starved");
    let worker = Worker::new(Arc::clone(&queue), idle.clone(), noop)
        .with_poll_interval(Duration::from_millis(20))
        .with_monitoring_interval(Duration::from_millis(50))
        .with_alerting_config(
            AlertingConfig::new()
                .alert_on_worker_starvation(Duration::from_millis(50))
                .webhook(&url),
        );
    let (shutdown_tx, shutdown_rx) = mpsc::channel(1);
    let task = tokio::spawn(async move { worker.run(shutdown_rx).await });
    let alert = next_alert(&mut alerts, "WorkerStarvation").await;
    assert_eq!(alert["queue_name"], idle.as_str());
    shutdown_tx.send(()).await.unwrap();
    task.await.unwrap().unwrap();
}

/// #64 H10: an alert target that never answers does not hold up job processing; alerts
/// are checked and delivered by the monitoring task, not the polling loop.
#[cfg(feature = "alerting")]
async fn hanging_alert_target_does_not_block_jobs<DB>(queue: Arc<JobQueue<DB>>)
where
    DB: sqlx::Database + Send + Sync + 'static,
    JobQueue<DB>: DatabaseQueue<Database = DB> + Send + Sync + 'static,
{
    use hammerwork::{alerting::AlertingConfig, stats::InMemoryStatsCollector};

    // Accepts connections and never answers.
    let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
    let url = format!("http://{}/alert", listener.local_addr().unwrap());
    let accepted = Arc::new(AtomicUsize::new(0));
    let server = {
        let accepted = Arc::clone(&accepted);
        tokio::spawn(async move {
            let mut open = Vec::new();
            while let Ok((socket, _)) = listener.accept().await {
                accepted.fetch_add(1, Ordering::SeqCst);
                open.push(socket);
            }
        })
    };

    let queue_name = test_utils::unique_queue("cfg_alert_hang");
    let noop: JobHandler = Arc::new(|_job: Job| Box::pin(async { Ok(()) }));
    let worker = Worker::new(Arc::clone(&queue), queue_name.clone(), noop)
        .with_poll_interval(Duration::from_millis(20))
        .with_monitoring_interval(Duration::from_millis(20))
        .with_stats_collector(Arc::new(InMemoryStatsCollector::new_default()))
        .with_alerting_config(
            AlertingConfig::new()
                .alert_on_worker_starvation(Duration::from_millis(1))
                .alert_on_queue_depth(0)
                .alert_on_high_error_rate(0.0)
                .webhook(&url),
        );
    let (shutdown_tx, shutdown_rx) = mpsc::channel(1);
    let task = tokio::spawn(async move { worker.run(shutdown_rx).await });

    // Let alerts fire (and hang) while the queue is idle, then give it work.
    let fired = test_utils::eventually(Duration::from_secs(5), || async {
        accepted.load(Ordering::SeqCst) > 0
    })
    .await;
    assert!(fired, "no alert was sent");
    let ids = enqueue_many(&queue, &queue_name, 3).await;
    assert!(
        wait_for_jobs(&queue, &ids, Duration::from_secs(5), completed).await,
        "a hanging alert target blocked job processing"
    );

    // Shutdown does not wait for the hanging alert either.
    tokio::time::timeout(Duration::from_secs(5), async {
        shutdown_tx.send(()).await.unwrap();
        task.await.unwrap().unwrap();
    })
    .await
    .expect("shutdown waited for a hanging alert");
    server.abort();
    delete_jobs(&queue, &ids).await;
}

/// A worker with an alert manager and a statistics collector raises a high error rate
/// alert once its failed jobs push the error rate over the threshold.
#[cfg(feature = "alerting")]
async fn worker_sends_error_rate_alerts<DB>(queue: Arc<JobQueue<DB>>)
where
    DB: sqlx::Database + Send + Sync + 'static,
    JobQueue<DB>: DatabaseQueue<Database = DB> + Send + Sync + 'static,
{
    use hammerwork::{
        alerting::{AlertManager, AlertingConfig},
        stats::InMemoryStatsCollector,
    };

    let (url, mut alerts) = capture_server().await;
    let queue_name = test_utils::unique_queue("cfg_alert_errors");
    let mut ids = Vec::new();
    for _ in 0..3 {
        ids.push(
            queue
                .enqueue(Job::new(queue_name.clone(), json!({})).with_max_attempts(1))
                .await
                .unwrap(),
        );
    }
    let failing: JobHandler = Arc::new(|_job: Job| {
        Box::pin(async {
            Err(HammerworkError::Worker {
                message: "always fails".to_string(),
            })
        })
    });
    let manager = Arc::new(AlertManager::new(
        AlertingConfig::new()
            .alert_on_high_error_rate(0.5)
            .webhook(&url),
    ));
    let worker = Worker::new(Arc::clone(&queue), queue_name.clone(), failing)
        .with_poll_interval(Duration::from_millis(20))
        .with_monitoring_interval(Duration::from_millis(50))
        .with_stats_collector(Arc::new(InMemoryStatsCollector::new_default()))
        .with_alert_manager(manager);
    let (shutdown_tx, shutdown_rx) = mpsc::channel(1);
    let task = tokio::spawn(async move { worker.run(shutdown_rx).await });

    let alert = next_alert(&mut alerts, "HighErrorRate").await;
    assert_eq!(alert["queue_name"], queue_name.as_str());
    assert!(alert["current_value"].as_f64().unwrap() > 0.5);
    shutdown_tx.send(()).await.unwrap();
    task.await.unwrap().unwrap();
    delete_jobs(&queue, &ids).await;
}

/// A worker with a Prometheus collector records job outcomes and the queue depth.
#[cfg(feature = "metrics")]
async fn worker_records_prometheus_metrics<DB>(queue: Arc<JobQueue<DB>>)
where
    DB: sqlx::Database + Send + Sync + 'static,
    JobQueue<DB>: DatabaseQueue<Database = DB> + Send + Sync + 'static,
{
    use hammerwork::metrics::{MetricsConfig, PrometheusMetricsCollector};

    let queue_name = test_utils::unique_queue("cfg_metrics");
    let ok = queue
        .enqueue(Job::new(queue_name.clone(), json!({ "fail": false })))
        .await
        .unwrap();
    let dead = queue
        .enqueue(Job::new(queue_name.clone(), json!({ "fail": true })).with_max_attempts(1))
        .await
        .unwrap();
    let handler: JobHandler = Arc::new(|job: Job| {
        Box::pin(async move {
            if job.payload["fail"] == true {
                Err(HammerworkError::Worker {
                    message: "connection refused".to_string(),
                })
            } else {
                Ok(())
            }
        })
    });
    let collector = Arc::new(PrometheusMetricsCollector::new(MetricsConfig::new()).unwrap());
    let worker = Worker::new(Arc::clone(&queue), queue_name.clone(), handler)
        .with_poll_interval(Duration::from_millis(20))
        .with_metrics_collector(Arc::clone(&collector));
    let (shutdown_tx, shutdown_rx) = mpsc::channel(1);
    let task = tokio::spawn(async move { worker.run(shutdown_rx).await });
    assert!(
        wait_for_jobs(&queue, &[ok, dead], Duration::from_secs(10), |job| {
            matches!(job.status, JobStatus::Completed | JobStatus::Dead)
        })
        .await
    );
    shutdown_tx.send(()).await.unwrap();
    task.await.unwrap().unwrap();

    let text = collector.get_metrics_text().unwrap();
    let line = |metric: &str, labels: &[&str]| {
        text.lines()
            .find(|line| {
                line.starts_with(metric)
                    && line.contains(&format!("queue=\"{queue_name}\""))
                    && labels.iter().all(|label| line.contains(label))
            })
            .unwrap_or_else(|| panic!("no {metric} {labels:?} line in:\n{text}"))
            .to_string()
    };
    assert!(line("hammerwork_jobs_total", &["status=\"completed\""]).ends_with(" 1"));
    assert!(line("hammerwork_jobs_total", &["status=\"dead\""]).ends_with(" 1"));
    assert!(
        line(
            "hammerwork_jobs_failed_total",
            &["error_type=\"exhausted\""]
        )
        .ends_with(" 1")
    );
    line("hammerwork_job_duration_seconds_count", &[]);
    line("hammerwork_queue_depth", &[]);
    delete_jobs(&queue, &[ok, dead]).await;
}

/// A worker with an event manager publishes lifecycle events for its jobs.
#[cfg(feature = "webhooks")]
async fn worker_publishes_lifecycle_events<DB>(queue: Arc<JobQueue<DB>>)
where
    DB: sqlx::Database + Send + Sync + 'static,
    JobQueue<DB>: DatabaseQueue<Database = DB> + Send + Sync + 'static,
{
    use hammerwork::events::{EventFilter, EventManager, JobLifecycleEventType};

    let queue_name = test_utils::unique_queue("cfg_events");
    let events = Arc::new(EventManager::new_default());
    let mut subscription = events
        .subscribe(EventFilter::new().with_queue_names(vec![queue_name.clone()]))
        .await
        .unwrap();
    let id = queue
        .enqueue(Job::new(queue_name.clone(), json!({})))
        .await
        .unwrap();
    let noop: JobHandler = Arc::new(|_job: Job| Box::pin(async { Ok(()) }));
    let worker = Worker::new(Arc::clone(&queue), queue_name.clone(), noop)
        .with_poll_interval(Duration::from_millis(20))
        .with_event_manager(Arc::clone(&events));
    let (shutdown_tx, shutdown_rx) = mpsc::channel(1);
    let task = tokio::spawn(async move { worker.run(shutdown_rx).await });

    let mut seen = Vec::new();
    tokio::time::timeout(Duration::from_secs(10), async {
        while !seen.contains(&JobLifecycleEventType::Completed) {
            let event = subscription.receiver.recv().await.unwrap();
            if event.job_id == id {
                assert_eq!(event.queue_name, queue_name);
                assert_eq!(event.metadata["worker_queue"], queue_name);
                seen.push(event.event_type);
            }
        }
    })
    .await
    .expect("lifecycle events for the job");
    assert_eq!(
        seen,
        vec![
            JobLifecycleEventType::Started,
            JobLifecycleEventType::Completed
        ]
    );
    shutdown_tx.send(()).await.unwrap();
    task.await.unwrap().unwrap();
    delete_jobs(&queue, &[id]).await;
}

#[cfg(feature = "postgres")]
mod postgres_tests {
    use super::*;

    async fn queue() -> Arc<JobQueue<sqlx::Postgres>> {
        test_utils::setup_postgres_queue().await
    }

    /// A queue whose pool is closed: every query fails.
    async fn broken_queue() -> Arc<JobQueue<sqlx::Postgres>> {
        let pool = sqlx::PgPool::connect(&test_utils::postgres_url())
            .await
            .unwrap();
        pool.close().await;
        Arc::new(JobQueue::new(pool))
    }

    crate::test_utils::scenario_tests!(
        queue;
        test_postgres_pool_runs_pool_size_workers => pool_runs_pool_size_workers,
        test_postgres_config_timeout_and_retry_strategy_apply => config_timeout_and_retry_strategy_apply,
        test_postgres_config_strict_priority_orders_jobs => config_strict_priority_orders_jobs,
        test_postgres_throttle_max_concurrent_caps_the_pool => throttle_max_concurrent_caps_the_pool,
        test_postgres_throttle_rate_limits_the_pool => throttle_rate_limits_the_pool,
        test_postgres_autoscaling_adds_and_retires_workers => autoscaling_adds_and_retires_workers,
        test_postgres_autoscaling_only_scales_the_template_queue => autoscaling_only_scales_the_template_queue,
        test_postgres_rate_limit_is_not_spent_on_empty_polls => rate_limit_is_not_spent_on_empty_polls,
        test_postgres_worker_waits_while_queue_is_paused => worker_waits_while_queue_is_paused,
        test_postgres_job_hooks_fire_for_each_outcome => job_hooks_fire_for_each_outcome,
        #[cfg(feature = "alerting")]
        test_postgres_worker_sends_queue_depth_and_starvation_alerts => worker_sends_queue_depth_and_starvation_alerts,
        #[cfg(feature = "alerting")]
        test_postgres_worker_sends_error_rate_alerts => worker_sends_error_rate_alerts,
        #[cfg(feature = "alerting")]
        test_postgres_hanging_alert_target_does_not_block_jobs => hanging_alert_target_does_not_block_jobs,
        #[cfg(feature = "metrics")]
        test_postgres_worker_records_prometheus_metrics => worker_records_prometheus_metrics,
        #[cfg(feature = "webhooks")]
        test_postgres_worker_publishes_lifecycle_events => worker_publishes_lifecycle_events,
    );

    crate::test_utils::scenario_tests!(
        broken_queue;
        test_postgres_worker_backs_off_on_errors_and_stops_promptly => worker_backs_off_on_errors_and_stops_promptly,
    );
}

#[cfg(feature = "mysql")]
mod mysql_tests {
    use super::*;

    async fn queue() -> Arc<JobQueue<sqlx::MySql>> {
        test_utils::setup_mysql_queue().await
    }

    /// A queue whose pool is closed: every query fails.
    async fn broken_queue() -> Arc<JobQueue<sqlx::MySql>> {
        let pool = sqlx::MySqlPool::connect(&test_utils::mysql_url())
            .await
            .unwrap();
        pool.close().await;
        Arc::new(JobQueue::new(pool))
    }

    crate::test_utils::scenario_tests!(
        queue;
        test_mysql_pool_runs_pool_size_workers => pool_runs_pool_size_workers,
        test_mysql_config_timeout_and_retry_strategy_apply => config_timeout_and_retry_strategy_apply,
        test_mysql_config_strict_priority_orders_jobs => config_strict_priority_orders_jobs,
        test_mysql_throttle_max_concurrent_caps_the_pool => throttle_max_concurrent_caps_the_pool,
        test_mysql_throttle_rate_limits_the_pool => throttle_rate_limits_the_pool,
        test_mysql_autoscaling_adds_and_retires_workers => autoscaling_adds_and_retires_workers,
        test_mysql_autoscaling_only_scales_the_template_queue => autoscaling_only_scales_the_template_queue,
        test_mysql_rate_limit_is_not_spent_on_empty_polls => rate_limit_is_not_spent_on_empty_polls,
        test_mysql_worker_waits_while_queue_is_paused => worker_waits_while_queue_is_paused,
        test_mysql_job_hooks_fire_for_each_outcome => job_hooks_fire_for_each_outcome,
        #[cfg(feature = "alerting")]
        test_mysql_worker_sends_queue_depth_and_starvation_alerts => worker_sends_queue_depth_and_starvation_alerts,
        #[cfg(feature = "alerting")]
        test_mysql_worker_sends_error_rate_alerts => worker_sends_error_rate_alerts,
        #[cfg(feature = "alerting")]
        test_mysql_hanging_alert_target_does_not_block_jobs => hanging_alert_target_does_not_block_jobs,
        #[cfg(feature = "metrics")]
        test_mysql_worker_records_prometheus_metrics => worker_records_prometheus_metrics,
        #[cfg(feature = "webhooks")]
        test_mysql_worker_publishes_lifecycle_events => worker_publishes_lifecycle_events,
    );

    crate::test_utils::scenario_tests!(
        broken_queue;
        test_mysql_worker_backs_off_on_errors_and_stops_promptly => worker_backs_off_on_errors_and_stops_promptly,
    );
}