hammerwork 2.4.0

//! Worker types for processing jobs from the job queue.
//!
//! This module provides the [`Worker`] and [`WorkerPool`] types that are responsible
//! for polling the job queue for work and executing job handlers. Workers support
//! extensive configuration including priority scheduling, rate limiting, timeouts,
//! statistics collection, and monitoring.

use crate::{
    Result,
    batch::BatchId,
    error::HammerworkError,
    job::Job,
    priority::PriorityWeights,
    queue::{DatabaseQueue, JobOutcome, JobQueue, RecordedOutcome},
    rate_limit::{RateLimit, RateLimiter, ThrottleConfig},
    retry::RetryStrategy,
    stats::{JobEvent, JobEventType, StatisticsCollector},
};

#[cfg(feature = "metrics")]
use crate::metrics::PrometheusMetricsCollector;

#[cfg(feature = "alerting")]
use crate::alerting::{AlertManager, AlertingConfig};

#[cfg(feature = "webhooks")]
use crate::events::{EventManager, JobError, JobLifecycleEvent, JobLifecycleEventType};

use chrono::{DateTime, Utc};
use sqlx::Database;
use std::{
    future::Future,
    panic::AssertUnwindSafe,
    pin::Pin,
    sync::Arc,
    task::{Context, Poll},
    time::Duration,
};
use tokio::{
    sync::{mpsc, watch},
    task::JoinSet,
    time::sleep,
};
use tracing::{debug, error, info, warn};

mod autoscale;
mod hooks;
mod pool;

pub use autoscale::{AutoscaleConfig, AutoscaleMetrics};
pub use hooks::{JobEventHooks, JobHookEvent, JobHookHandler};
pub use pool::{DEFAULT_RESULT_CLEANUP_INTERVAL, WorkerPool};

/// Default time a worker waits for an in-flight job to finish after shutdown is requested.
pub const DEFAULT_SHUTDOWN_GRACE_PERIOD: Duration = Duration::from_secs(30);

/// Default lease a worker holds on a running job. The lease is taken when the job is
/// claimed and renewed every third of this period; if the worker dies, the job becomes
/// eligible for [`DatabaseQueue::requeue_stale_jobs`] once the lease expires.
pub const DEFAULT_LEASE_DURATION: Duration = crate::queue::DEFAULT_LEASE_DURATION;

/// Default interval at which a worker's monitoring task updates its queue depth metric
/// and checks its alert thresholds ([`Worker::with_monitoring_interval`]).
pub const DEFAULT_MONITORING_INTERVAL: Duration = Duration::from_secs(30);

/// Floor on the monitoring interval.
#[cfg(any(feature = "metrics", feature = "alerting"))]
const MIN_MONITORING_INTERVAL: Duration = Duration::from_millis(10);

/// Default interval at which a [`WorkerPool`] runs the stale job reaper.
pub const DEFAULT_STALE_JOB_REAPER_INTERVAL: Duration = Duration::from_secs(60);

/// Upper bound on a computed retry delay. Larger values (from a custom strategy or an
/// unbounded backoff) are clamped so the retry timestamp stays representable in both
/// databases.
pub const MAX_RETRY_DELAY: Duration = Duration::from_secs(365 * 24 * 60 * 60);

/// Cap on the exponential backoff applied after consecutive worker loop errors.
const MAX_ERROR_BACKOFF: Duration = Duration::from_secs(60);

/// Floor on the error backoff so a zero poll interval can never cause a hot loop.
const MIN_ERROR_BACKOFF: Duration = Duration::from_millis(100);

/// Floor on the wait between polls of an idle worker, so a zero poll interval can never
/// cause a hot loop.
pub const MIN_POLL_INTERVAL: Duration = Duration::from_millis(10);

/// Floor on the heartbeat interval derived from the lease duration.
const MIN_HEARTBEAT_INTERVAL: Duration = Duration::from_millis(100);

/// Default number of jobs a worker in batch mode claims per poll: one, so
/// `with_batch_processing_enabled(true)` alone keeps claiming one job at a time (as before
/// 2.1) and only adds batch statistics. Set a larger size with [`Worker::with_batch_size`].
pub const DEFAULT_BATCH_SIZE: usize = 1;

/// Default number of claimed jobs a worker in batch mode runs at once
/// ([`Worker::with_batch_concurrency`]): one, so batch mode changes how jobs are claimed
/// but not how many handlers a worker runs at a time.
pub const DEFAULT_BATCH_CONCURRENCY: usize = 1;

/// Delay before a [`WorkerPool`] restarts a worker that died unexpectedly.
const WORKER_RESTART_DELAY: Duration = Duration::from_secs(1);

type HandlerFuture = Pin<Box<dyn Future<Output = Result<JobResult>> + Send>>;

/// Future adapter that turns a panic during `poll` into an `Err` carrying the payload.
struct CatchUnwind<F> {
    inner: F,
}

impl<F: Future + Unpin> Future for CatchUnwind<F> {
    type Output = std::thread::Result<F::Output>;

    fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
        let inner = &mut self.inner;
        match std::panic::catch_unwind(AssertUnwindSafe(|| Pin::new(&mut *inner).poll(cx))) {
            Ok(Poll::Pending) => Poll::Pending,
            Ok(Poll::Ready(output)) => Poll::Ready(Ok(output)),
            Err(payload) => Poll::Ready(Err(payload)),
        }
    }
}

/// Describe a panic payload for an error message.
fn panic_message(payload: &(dyn std::any::Any + Send)) -> String {
    if let Some(message) = payload.downcast_ref::<&str>() {
        (*message).to_string()
    } else if let Some(message) = payload.downcast_ref::<String>() {
        message.clone()
    } else {
        "non-string panic payload".to_string()
    }
}

/// Await `future`, which runs user-supplied code (a statistics collector or a spawn
/// handler), containing a panic: it is logged at error level and returned as an error,
/// so the worker still records the job's outcome and keeps polling.
///
/// Event hooks are synchronous and are guarded the same way by `JobEventHooks`.
async fn contain_panic<T>(what: &str, future: impl Future<Output = Result<T>>) -> Result<T> {
    let guarded = CatchUnwind {
        inner: Box::pin(future),
    };
    match guarded.await {
        Ok(result) => result,
        Err(payload) => {
            let message = format!("{what} panicked: {}", panic_message(payload.as_ref()));
            error!("{}", message);
            Err(HammerworkError::Worker { message })
        }
    }
}

/// Build the error used when a job handler panics.
fn handler_panic_error(payload: &(dyn std::any::Any + Send)) -> HammerworkError {
    HammerworkError::Worker {
        message: format!("Job handler panicked: {}", panic_message(payload)),
    }
}

/// Convert a retry delay into the timestamp at which the job should run again.
///
/// The delay is clamped to [`MAX_RETRY_DELAY`] and the conversion never panics.
fn retry_at_from_delay(now: DateTime<Utc>, retry_delay: Duration) -> DateTime<Utc> {
    let delay = chrono::Duration::from_std(retry_delay.min(MAX_RETRY_DELAY))
        .unwrap_or(chrono::Duration::MAX);
    now.checked_add_signed(delay)
        .unwrap_or(DateTime::<Utc>::MAX_UTC)
}

/// Whether a finished batch counts as successful in [`BatchProcessingStats`]: at least
/// 95% of its jobs completed. [`BatchResult::success_rate`](crate::batch::BatchResult::success_rate)
/// is a percentage (0 to 100).
fn batch_counts_as_successful(result: &crate::batch::BatchResult) -> bool {
    result.success_rate() >= 95.0
}

/// How many attempts a job gets: its own `max_attempts`, lowered by the worker's cap
/// (`Worker::with_max_retries`) when one is set.
fn effective_attempt_limit(job_max_attempts: i32, worker_cap: Option<i32>) -> i32 {
    match worker_cap {
        Some(cap) => job_max_attempts.min(cap),
        None => job_max_attempts,
    }
}

/// Aborts a background task when dropped, so it never outlives its owner.
#[cfg(any(feature = "metrics", feature = "alerting"))]
struct AbortOnDrop(tokio::task::JoinHandle<()>);

#[cfg(any(feature = "metrics", feature = "alerting"))]
impl Drop for AbortOnDrop {
    fn drop(&mut self) {
        self.0.abort();
    }
}

/// Result of one attempt to acquire work in the worker loop: a job (boxed, as a `Job`
/// is much larger than the other variants) or, in batch mode, the claimed jobs.
enum Acquired<T> {
    Work(T),
    Idle,
    Shutdown,
}

/// A job claimed in a batch, with the throttle permit it holds until it finishes.
struct ClaimedJob {
    job: Job,
    permit: Option<tokio::sync::OwnedSemaphorePermit>,
}

/// One job of a batch being processed; resolves to its id and processing outcome.
type BatchRun<'a> = Pin<Box<dyn Future<Output = (crate::job::JobId, Result<()>)> + Send + 'a>>;

/// Wait for the next of `runs` to finish and remove it, or `None` when there are none.
fn next_finished<'r, 'a>(
    runs: &'r mut Vec<BatchRun<'a>>,
) -> impl Future<Output = Option<(crate::job::JobId, Result<()>)>> + Send + 'r {
    std::future::poll_fn(move |cx| {
        if runs.is_empty() {
            return Poll::Ready(None);
        }
        for index in 0..runs.len() {
            if let Poll::Ready(finished) = runs[index].as_mut().poll(cx) {
                drop(runs.swap_remove(index));
                return Poll::Ready(Some(finished));
            }
        }
        Poll::Pending
    })
}

/// The claimed jobs of a batch that have not started yet, whose leases the worker
/// renews while they wait, and those whose lease was found lost.
#[derive(Default)]
struct HeldJobs {
    waiting: Vec<Job>,
    lost: std::collections::HashSet<crate::job::JobId>,
    /// When each waiting job's lease was last written (by the claim or a heartbeat).
    renewed: std::collections::HashMap<crate::job::JobId, std::time::Instant>,
}

/// What to do with a held job when its turn comes.
#[derive(Debug, PartialEq, Eq)]
enum HeldStart {
    /// Its lease was lost while it waited: do not run it.
    Lost,
    /// Run it; its lease was renewed recently enough for the running heartbeat to take over.
    Run,
    /// Renew its lease first: the running heartbeat only fires after a full interval, and
    /// together with the time since the last renewal that could outlast the lease.
    RenewThenRun,
}

/// A [`JobEvent`] of `event_type` for `job`.
fn job_event(
    job: &Job,
    event_type: JobEventType,
    processing_time_ms: Option<u64>,
    error_message: Option<String>,
    timestamp: DateTime<Utc>,
) -> JobEvent {
    JobEvent {
        job_id: job.id,
        queue_name: job.queue_name.clone(),
        event_type,
        priority: job.priority,
        processing_time_ms,
        error_message,
        timestamp,
    }
}

/// The payload to put in a lifecycle event of `job`: `None` for a job with payload
/// encryption, so that an encrypted job's payload (sealed or redacted as stored, or its
/// plaintext) never reaches events, webhooks or streams.
#[cfg(feature = "webhooks")]
fn event_payload(job: &Job) -> Option<serde_json::Value> {
    if job.is_encrypted || job.has_encryption() {
        None
    } else {
        Some(job.payload.clone())
    }
}

/// How many finished batches a worker remembers, to count each of them once.
const FINISHED_BATCHES_REMEMBERED: usize = 10_000;

/// The batches a worker has counted as finished, most recent last; bounded by
/// [`FINISHED_BATCHES_REMEMBERED`].
#[derive(Default)]
struct FinishedBatches {
    ids: std::collections::HashSet<BatchId>,
    order: std::collections::VecDeque<BatchId>,
}

impl FinishedBatches {
    /// Remember `batch_id`; `false` if it already was.
    fn insert(&mut self, batch_id: BatchId) -> bool {
        if !self.ids.insert(batch_id) {
            return false;
        }
        self.order.push_back(batch_id);
        if self.order.len() > FINISHED_BATCHES_REMEMBERED
            && let Some(oldest) = self.order.pop_front()
        {
            self.ids.remove(&oldest);
        }
        true
    }
}

/// Whether shutdown was requested (or the shutdown sender was dropped), without waiting.
fn shutdown_requested(shutdown_rx: &mut mpsc::Receiver<()>) -> bool {
    !matches!(
        shutdown_rx.try_recv(),
        Err(mpsc::error::TryRecvError::Empty)
    )
}

/// Exponential backoff after `consecutive_errors` failed loop iterations.
fn error_backoff(base: Duration, consecutive_errors: u32) -> Duration {
    let base = base.max(MIN_ERROR_BACKOFF);
    let exponent = consecutive_errors.saturating_sub(1).min(16);
    base.checked_mul(1u32 << exponent)
        .unwrap_or(MAX_ERROR_BACKOFF)
        .min(MAX_ERROR_BACKOFF.max(base))
}

/// Statistics for batch job processing by a worker.
#[derive(Debug, Clone, Default)]
pub struct BatchProcessingStats {
    /// Total number of batch jobs processed
    pub jobs_processed: u64,
    /// Total number of batch jobs completed successfully
    pub jobs_completed: u64,
    /// Total number of batch jobs that failed
    pub jobs_failed: u64,
    /// Total number of batches completed
    pub batches_completed: u64,
    /// Total number of batches completed successfully (>95% success rate)
    pub batches_successful: u64,
    /// Total processing time for all batch jobs in milliseconds
    pub total_processing_time_ms: u64,
    /// Average processing time per job in milliseconds
    pub average_processing_time_ms: f64,
    /// Timestamp of the last processed batch job
    pub last_processed_job: Option<DateTime<Utc>>,
}

impl BatchProcessingStats {
    /// Calculate the success rate for batch jobs
    pub fn success_rate(&self) -> f64 {
        if self.jobs_processed == 0 {
            0.0
        } else {
            self.jobs_completed as f64 / self.jobs_processed as f64
        }
    }

    /// Calculate the batch success rate
    pub fn batch_success_rate(&self) -> f64 {
        if self.batches_completed == 0 {
            0.0
        } else {
            self.batches_successful as f64 / self.batches_completed as f64
        }
    }

    /// Update the average processing time
    pub fn update_average_processing_time(&mut self) {
        if self.jobs_completed > 0 {
            self.average_processing_time_ms =
                self.total_processing_time_ms as f64 / self.jobs_completed as f64;
        }
    }
}

/// Type alias for job handler functions.
///
/// Job handlers are async functions that take a [`Job`] and return a [`Result`].
/// The handler is responsible for processing the job's payload and performing
/// the actual work. Handlers should return `Ok(())` on success or an error
/// if the job should be retried or marked as failed.
///
/// # Examples
///
/// ```rust
/// use hammerwork::{Job, Result, worker::JobHandler};
/// use std::sync::Arc;
///
/// let handler: JobHandler = Arc::new(|job: Job| {
///     Box::pin(async move {
///         // Process the job
///         println!("Processing job: {:?}", job.payload);
///         
///         // Return Ok(()) on success, Err(_) to trigger retry
///         Ok(())
///     })
/// });
/// ```
pub type JobHandler = Arc<
    dyn Fn(Job) -> std::pin::Pin<Box<dyn std::future::Future<Output = Result<()>> + Send>>
        + Send
        + Sync,
>;

/// Type alias for job handler functions that can return result data.
///
/// Enhanced job handlers are async functions that take a [`Job`] and return a [`Result<JobResult>`].
/// They support returning optional result data that can be stored and retrieved later.
///
/// # Examples
///
/// ```rust
/// use hammerwork::{Job, Result, worker::{JobHandlerWithResult, JobResult}};
/// use std::sync::Arc;
/// use serde_json::json;
///
/// let handler: JobHandlerWithResult = Arc::new(|job: Job| {
///     Box::pin(async move {
///         // Process the job
///         let result_data = json!({
///             "processed_items": 42,
///             "status": "completed"
///         });
///         
///         // Return result with data
///         Ok(JobResult::with_data(result_data))
///     })
/// });
/// ```
pub type JobHandlerWithResult = Arc<
    dyn Fn(Job) -> std::pin::Pin<Box<dyn std::future::Future<Output = Result<JobResult>> + Send>>
        + Send
        + Sync,
>;

/// Result returned by job handlers, optionally containing result data.
#[derive(Debug, Clone)]
pub struct JobResult {
    /// Optional result data to store for retrieval
    pub data: Option<serde_json::Value>,
}

impl JobResult {
    /// Create a JobResult with no data (equivalent to the old Ok(()) pattern)
    pub fn success() -> Self {
        Self { data: None }
    }

    /// Create a JobResult with result data
    pub fn with_data(data: serde_json::Value) -> Self {
        Self { data: Some(data) }
    }
}

impl Default for JobResult {
    fn default() -> Self {
        Self::success()
    }
}

/// Enum to support both original and enhanced job handlers
#[derive(Clone)]
pub enum JobHandlerType {
    /// Original handler type that returns Result<()>
    Legacy(JobHandler),
    /// Enhanced handler type that returns Result<JobResult> with optional data
    WithResult(JobHandlerWithResult),
}

/// A worker that processes jobs from a specific queue.
///
/// Workers continuously poll their assigned queue for pending jobs and execute
/// them using the provided handler function. They support extensive configuration
/// for controlling job processing behavior.
///
/// # Features
///
/// - **Priority-aware job selection**: Configurable weighted or strict priority scheduling
/// - **Rate limiting**: Token bucket rate limiting with configurable burst limits
/// - **Automatic retries**: Configurable retry attempts with exponential backoff
/// - **Timeout handling**: Per-job and worker-level timeout configuration
/// - **Statistics collection**: Integration with statistics collectors for monitoring
/// - **Metrics and alerting**: Optional Prometheus metrics and alerting support
///
/// # Examples
///
/// ## Basic Worker
///
#[cfg_attr(feature = "postgres", doc = "```rust,no_run")]
#[cfg_attr(not(feature = "postgres"), doc = "```rust,ignore")]
/// use hammerwork::{Worker, JobQueue, Job};
/// use std::sync::Arc;
///
/// # async fn example() -> Result<(), Box<dyn std::error::Error>> {
/// # let pool = sqlx::PgPool::connect("postgresql://localhost/test").await?;
/// let queue = Arc::new(JobQueue::new(pool));
///
/// let handler: hammerwork::worker::JobHandler = Arc::new(|job: Job| {
///     Box::pin(async move {
///         println!("Processing: {:?}", job.payload);
///         Ok(())
///     })
/// });
///
/// let worker = Worker::new(queue, "email_queue".to_string(), handler)
///     .with_poll_interval(std::time::Duration::from_millis(500))
///     .with_max_retries(5);
///
/// // Start processing jobs
/// let mut pool = hammerwork::WorkerPool::new();
/// pool.add_worker(worker);
/// pool.start().await?;
/// # Ok(())
/// # }
/// ```
///
/// ## Worker with Priority and Rate Limiting
///
#[cfg_attr(feature = "postgres", doc = "```rust,no_run")]
#[cfg_attr(not(feature = "postgres"), doc = "```rust,ignore")]
/// use hammerwork::{Worker, JobQueue, PriorityWeights, RateLimit};
/// use std::sync::Arc;
/// use std::time::Duration;
///
/// # async fn example() -> Result<(), Box<dyn std::error::Error>> {
/// # let pool = sqlx::PgPool::connect("postgresql://localhost/test").await?;
/// # let queue = Arc::new(JobQueue::new(pool));
/// # let handler: hammerwork::worker::JobHandler = Arc::new(|job| Box::pin(async move { Ok(()) }));
///
/// let priority_weights = PriorityWeights::new()
///     .with_weight(hammerwork::JobPriority::Critical, 50)
///     .with_weight(hammerwork::JobPriority::High, 20);
///
/// let rate_limit = RateLimit::per_second(10).with_burst_limit(20);
///
/// let worker = Worker::new(queue, "api_queue".to_string(), handler)
///     .with_priority_weights(priority_weights)
///     .with_rate_limit(rate_limit)
///     .with_default_timeout(Duration::from_secs(300));
///
/// let mut pool = hammerwork::WorkerPool::new();
/// pool.add_worker(worker);
/// pool.start().await?;
/// # Ok(())
/// # }
/// ```
pub struct Worker<DB: Database> {
    /// The job queue to poll for work
    queue: Arc<JobQueue<DB>>,
    /// Name of the queue this worker processes
    queue_name: String,
    /// Function to handle job processing
    handler: JobHandlerType,
    /// How often to poll for new jobs
    poll_interval: Duration,
    /// Maximum number of retry attempts for failed jobs
    max_retries: Option<i32>,
    /// Delay between retry attempts
    retry_delay: Duration,
    /// Default retry strategy for failed jobs (overrides retry_delay if specified)
    default_retry_strategy: Option<RetryStrategy>,
    /// Default timeout for jobs (overridden by job-specific timeouts)
    default_timeout: Option<Duration>,
    /// Priority weights for job selection
    priority_weights: Option<PriorityWeights>,
    /// Statistics collector for monitoring
    stats_collector: Option<Arc<dyn StatisticsCollector>>,
    /// Rate limiter for controlling job processing rate
    rate_limiter: Option<RateLimiter>,
    /// Throttling configuration
    throttle_config: Option<ThrottleConfig>,
    /// Permits for the throttle's `max_concurrent`, shared by clones of this worker
    concurrency_limit: Option<Arc<tokio::sync::Semaphore>>,
    /// Prometheus metrics collector (when metrics feature is enabled)
    #[cfg(feature = "metrics")]
    metrics_collector: Option<Arc<PrometheusMetricsCollector>>,
    /// Alert manager for notifications (when alerting feature is enabled)
    #[cfg(feature = "alerting")]
    alert_manager: Option<Arc<AlertManager>>,
    /// Timestamp of the last processed job (for starvation detection)
    last_job_time: Arc<std::sync::RwLock<DateTime<Utc>>>,
    /// Claim jobs in batches (and track statistics of jobs that belong to a `JobBatch`)
    batch_processing_enabled: bool,
    /// Most jobs claimed per poll in batch mode
    batch_size: usize,
    /// Most claimed jobs run at once in batch mode
    batch_concurrency: usize,
    /// Track batch processing statistics
    batch_stats: Arc<std::sync::RwLock<BatchProcessingStats>>,
    /// Batches already counted in `batch_stats` as finished
    finished_batches: Arc<std::sync::Mutex<FinishedBatches>>,
    /// Job lifecycle event hooks
    event_hooks: JobEventHooks,
    /// Spawn manager for dynamic job spawning
    spawn_manager: Option<Arc<crate::spawn::SpawnManager<DB>>>,
    /// Event manager for publishing job lifecycle events
    #[cfg(feature = "webhooks")]
    event_manager: Option<Arc<crate::events::EventManager>>,
    /// How long to wait for an in-flight job to finish after shutdown is requested
    shutdown_grace_period: Duration,
    /// Lease held on a running job, renewed by heartbeats
    lease_duration: Duration,
    /// How often the monitoring task updates metrics and checks alerts, when set
    /// explicitly (otherwise the metrics collector's update interval, or 30 seconds)
    monitoring_interval: Option<Duration>,
}

impl<DB: Database + Send + Sync + 'static> Clone for Worker<DB>
where
    JobQueue<DB>: DatabaseQueue<Database = DB> + Send + Sync,
{
    fn clone(&self) -> Self {
        Self {
            queue: Arc::clone(&self.queue),
            queue_name: self.queue_name.clone(),
            handler: self.handler.clone(),
            poll_interval: self.poll_interval,
            max_retries: self.max_retries,
            retry_delay: self.retry_delay,
            default_retry_strategy: self.default_retry_strategy.clone(),
            default_timeout: self.default_timeout,
            priority_weights: self.priority_weights.clone(),
            stats_collector: self.stats_collector.clone(),
            rate_limiter: self.rate_limiter.clone(),
            throttle_config: self.throttle_config.clone(),
            concurrency_limit: self.concurrency_limit.clone(),
            #[cfg(feature = "metrics")]
            metrics_collector: self.metrics_collector.clone(),
            #[cfg(feature = "alerting")]
            alert_manager: self.alert_manager.clone(),
            // Create new instances for per-worker state
            last_job_time: Arc::new(std::sync::RwLock::new(Utc::now())),
            batch_processing_enabled: self.batch_processing_enabled,
            batch_size: self.batch_size,
            batch_concurrency: self.batch_concurrency,
            batch_stats: Arc::new(std::sync::RwLock::new(BatchProcessingStats::default())),
            finished_batches: Default::default(),
            event_hooks: self.event_hooks.clone(),
            spawn_manager: self.spawn_manager.clone(),
            #[cfg(feature = "webhooks")]
            event_manager: self.event_manager.clone(),
            shutdown_grace_period: self.shutdown_grace_period,
            lease_duration: self.lease_duration,
            monitoring_interval: self.monitoring_interval,
        }
    }
}

impl<DB: Database + Send + Sync + 'static> Worker<DB>
where
    JobQueue<DB>: DatabaseQueue<Database = DB> + Send + Sync,
{
    /// Creates a new worker with default configuration.
    ///
    /// The worker will be created with:
    /// - 1 second polling interval
    /// - no cap on attempts: each job runs up to its own
    ///   [`max_attempts`](crate::Job::with_max_attempts) (3 unless set); see
    ///   [`with_max_retries`](Self::with_max_retries)
    /// - 30 second retry delay
    /// - a [lease](Self::with_lease_duration) of [`DEFAULT_LEASE_DURATION`] (5 minutes)
    /// - No timeout, rate limiting, or priority configuration
    ///
    /// # Arguments
    ///
    /// * `queue` - The job queue to poll for work
    /// * `queue_name` - Name of the queue this worker should process
    /// * `handler` - Function to handle job processing
    ///
    /// # Examples
    ///
    #[cfg_attr(feature = "postgres", doc = "```rust,no_run")]
    #[cfg_attr(not(feature = "postgres"), doc = "```rust,ignore")]
    /// use hammerwork::{Worker, JobQueue, Job};
    /// use std::sync::Arc;
    ///
    /// # async fn example() -> Result<(), Box<dyn std::error::Error>> {
    /// # let pool = sqlx::PgPool::connect("postgresql://localhost/test").await?;
    /// let queue = Arc::new(JobQueue::new(pool));
    ///
    /// let handler: hammerwork::worker::JobHandler = Arc::new(|job: Job| {
    ///     Box::pin(async move {
    ///         match job.payload.get("action").and_then(|v| v.as_str()) {
    ///             Some("send_email") => {
    ///                 // Send email logic
    ///                 println!("Sending email to: {:?}", job.payload.get("to"));
    ///                 Ok(())
    ///             },
    ///             Some("process_data") => {
    ///                 // Data processing logic
    ///                 println!("Processing data: {:?}", job.payload.get("data"));
    ///                 Ok(())
    ///             },
    ///             _ => Err(hammerwork::HammerworkError::Processing("Unknown action".to_string())),
    ///         }
    ///     })
    /// });
    ///
    /// let worker = Worker::new(queue, "default".to_string(), handler);
    /// # Ok(())
    /// # }
    /// ```
    pub fn new(queue: Arc<JobQueue<DB>>, queue_name: String, handler: JobHandler) -> Self {
        Self {
            queue,
            queue_name,
            handler: JobHandlerType::Legacy(handler),
            poll_interval: Duration::from_secs(1),
            max_retries: None,
            retry_delay: Duration::from_secs(30),
            default_retry_strategy: None,
            default_timeout: None,
            priority_weights: None,
            stats_collector: None,
            rate_limiter: None,
            throttle_config: None,
            concurrency_limit: None,
            #[cfg(feature = "metrics")]
            metrics_collector: None,
            #[cfg(feature = "alerting")]
            alert_manager: None,
            last_job_time: Arc::new(std::sync::RwLock::new(Utc::now())),
            batch_processing_enabled: false,
            batch_size: DEFAULT_BATCH_SIZE,
            batch_concurrency: DEFAULT_BATCH_CONCURRENCY,
            batch_stats: Arc::new(std::sync::RwLock::new(BatchProcessingStats::default())),
            finished_batches: Default::default(),
            event_hooks: JobEventHooks::default(),
            spawn_manager: None,
            #[cfg(feature = "webhooks")]
            event_manager: None,
            shutdown_grace_period: DEFAULT_SHUTDOWN_GRACE_PERIOD,
            lease_duration: DEFAULT_LEASE_DURATION,
            monitoring_interval: None,
        }
    }

    /// Creates a new worker with an enhanced handler that can return result data.
    ///
    /// This method creates a worker that uses the enhanced job handler type,
    /// which can return result data that will be automatically stored when
    /// the job completes successfully.
    ///
    /// # Arguments
    ///
    /// * `queue` - The job queue to poll for work
    /// * `queue_name` - Name of the queue this worker will process
    /// * `handler` - Enhanced handler function that can return result data
    ///
    /// # Examples
    ///
    #[cfg_attr(feature = "postgres", doc = "```rust,no_run")]
    #[cfg_attr(not(feature = "postgres"), doc = "```rust,ignore")]
    /// use hammerwork::{Worker, JobQueue, Job, worker::{JobHandlerWithResult, JobResult}};
    /// use std::sync::Arc;
    /// use serde_json::json;
    ///
    /// # async fn example() -> Result<(), Box<dyn std::error::Error>> {
    /// # let pool = sqlx::PgPool::connect("postgresql://localhost/test").await?;
    /// let queue = Arc::new(JobQueue::new(pool));
    ///
    /// let handler: JobHandlerWithResult = Arc::new(|job: Job| {
    ///     Box::pin(async move {
    ///         // Process the job and generate results
    ///         let result_data = json!({
    ///             "processed_items": 42,
    ///             "status": "completed"
    ///         });
    ///         
    ///         Ok(JobResult::with_data(result_data))
    ///     })
    /// });
    ///
    /// let worker = Worker::new_with_result_handler(queue, "default".to_string(), handler);
    /// # Ok(())
    /// # }
    /// ```
    pub fn new_with_result_handler(
        queue: Arc<JobQueue<DB>>,
        queue_name: String,
        handler: JobHandlerWithResult,
    ) -> Self {
        Self {
            queue,
            queue_name,
            handler: JobHandlerType::WithResult(handler),
            poll_interval: Duration::from_secs(1),
            max_retries: None,
            retry_delay: Duration::from_secs(30),
            default_retry_strategy: None,
            default_timeout: None,
            priority_weights: None,
            stats_collector: None,
            rate_limiter: None,
            throttle_config: None,
            concurrency_limit: None,
            #[cfg(feature = "metrics")]
            metrics_collector: None,
            #[cfg(feature = "alerting")]
            alert_manager: None,
            last_job_time: Arc::new(std::sync::RwLock::new(Utc::now())),
            batch_processing_enabled: false,
            batch_size: DEFAULT_BATCH_SIZE,
            batch_concurrency: DEFAULT_BATCH_CONCURRENCY,
            batch_stats: Arc::new(std::sync::RwLock::new(BatchProcessingStats::default())),
            finished_batches: Default::default(),
            event_hooks: JobEventHooks::default(),
            spawn_manager: None,
            #[cfg(feature = "webhooks")]
            event_manager: None,
            shutdown_grace_period: DEFAULT_SHUTDOWN_GRACE_PERIOD,
            lease_duration: DEFAULT_LEASE_DURATION,
            monitoring_interval: None,
        }
    }

    /// Adds a statistics collector for monitoring job processing.
    ///
    /// The statistics collector will receive events for job start, completion,
    /// failure, and timeout events, allowing for comprehensive monitoring
    /// of worker performance.
    ///
    /// # Arguments
    ///
    /// * `stats_collector` - The statistics collector to use
    ///
    /// # Examples
    ///
    #[cfg_attr(feature = "postgres", doc = "```rust,no_run")]
    #[cfg_attr(not(feature = "postgres"), doc = "```rust,ignore")]
    /// use hammerwork::{Worker, JobQueue, InMemoryStatsCollector};
    /// use std::sync::Arc;
    ///
    /// # async fn example() -> Result<(), Box<dyn std::error::Error>> {
    /// # let pool = sqlx::PgPool::connect("postgresql://localhost/test").await?;
    /// # let queue = Arc::new(JobQueue::new(pool));
    /// # let handler: hammerwork::worker::JobHandler = Arc::new(|job| Box::pin(async move { Ok(()) }));
    ///
    /// let stats = Arc::new(InMemoryStatsCollector::new_default());
    /// let worker = Worker::new(queue, "monitored".to_string(), handler)
    ///     .with_stats_collector(stats.clone());
    ///
    /// // Later, check statistics  
    /// use std::time::Duration;
    /// use hammerwork::StatisticsCollector;
    /// let queue_stats = stats.get_queue_statistics("monitored", Duration::from_secs(3600)).await?;
    /// println!("Processed: {}", queue_stats.total_processed);
    /// # Ok(())
    /// # }
    /// ```
    pub fn with_stats_collector(mut self, stats_collector: Arc<dyn StatisticsCollector>) -> Self {
        self.stats_collector = Some(stats_collector);
        self
    }

    /// Sets how often the worker polls for new jobs.
    ///
    /// Shorter intervals result in lower latency but higher database load.
    /// Longer intervals reduce database load but increase job processing latency.
    /// Intervals below [`MIN_POLL_INTERVAL`] (10ms), including zero, wait
    /// [`MIN_POLL_INTERVAL`], so an idle worker never polls in a tight loop.
    ///
    /// # Arguments
    ///
    /// * `interval` - Time between polling attempts
    ///
    /// # Examples
    ///
    #[cfg_attr(feature = "postgres", doc = "```rust,no_run")]
    #[cfg_attr(not(feature = "postgres"), doc = "```rust,ignore")]
    /// use hammerwork::Worker;
    /// use std::time::Duration;
    /// # use std::sync::Arc;
    /// # async fn example() -> Result<(), Box<dyn std::error::Error>> {
    /// # let pool = sqlx::PgPool::connect("postgresql://localhost/test").await?;
    /// # let queue = Arc::new(hammerwork::JobQueue::new(pool));
    /// # let handler: hammerwork::worker::JobHandler = Arc::new(|job| Box::pin(async move { Ok(()) }));
    ///
    /// // High-frequency polling for low latency
    /// let fast_worker = Worker::new(queue.clone(), "fast".to_string(), handler.clone())
    ///     .with_poll_interval(Duration::from_millis(100));
    ///
    /// // Lower frequency polling for reduced load
    /// let slow_worker = Worker::new(queue, "slow".to_string(), handler)
    ///     .with_poll_interval(Duration::from_secs(5));
    /// # Ok(())
    /// # }
    /// ```
    pub fn with_poll_interval(mut self, interval: Duration) -> Self {
        self.poll_interval = interval;
        self
    }

    /// Caps the number of attempts this worker gives any job.
    ///
    /// A job's own [`max_attempts`](crate::Job::with_max_attempts) is the source of
    /// truth: a failing (or timing-out) job runs up to `max_attempts` times and then
    /// becomes `Dead` (or `TimedOut`). This setting only lowers that limit: the
    /// effective limit is `min(job.max_attempts, max_retries)`. By default there is no
    /// cap.
    ///
    /// Previously this value replaced the job's `max_attempts` (default 3), so jobs
    /// with `with_max_attempts(5)` stopped after 3 runs in a non-retried `Failed`
    /// state, and `with_max_attempts(1)` still ran 3 times.
    ///
    /// # Arguments
    ///
    /// * `max_retries` - Maximum number of attempts per job on this worker
    ///
    /// # Examples
    ///
    #[cfg_attr(feature = "postgres", doc = "```rust,no_run")]
    #[cfg_attr(not(feature = "postgres"), doc = "```rust,ignore")]
    /// use hammerwork::Worker;
    /// # use std::sync::Arc;
    /// # async fn example() -> Result<(), Box<dyn std::error::Error>> {
    /// # let pool = sqlx::PgPool::connect("postgresql://localhost/test").await?;
    /// # let queue = Arc::new(hammerwork::JobQueue::new(pool));
    /// # let handler: hammerwork::worker::JobHandler = Arc::new(|job| Box::pin(async move { Ok(()) }));
    ///
    /// // Expensive calls: never run a job more than twice on this worker, even if the
    /// // job itself allows more attempts. To give jobs *more* attempts, raise their own
    /// // limit with `Job::with_max_attempts`; this setting can only lower it.
    /// let expensive_worker = Worker::new(queue, "expensive".to_string(), handler)
    ///     .with_max_retries(2);
    /// # Ok(())
    /// # }
    /// ```
    pub fn with_max_retries(mut self, max_retries: i32) -> Self {
        self.max_retries = Some(max_retries);
        self
    }

    /// Sets the delay between retry attempts.
    ///
    /// When a job fails and is scheduled for retry, it will wait this
    /// duration before becoming eligible for processing again.
    ///
    /// # Arguments
    ///
    /// * `delay` - Time to wait between retry attempts
    ///
    /// # Examples
    ///
    #[cfg_attr(feature = "postgres", doc = "```rust,no_run")]
    #[cfg_attr(not(feature = "postgres"), doc = "```rust,ignore")]
    /// use hammerwork::Worker;
    /// use std::time::Duration;
    /// # use std::sync::Arc;
    /// # async fn example() -> Result<(), Box<dyn std::error::Error>> {
    /// # let pool = sqlx::PgPool::connect("postgresql://localhost/test").await?;
    /// # let queue = Arc::new(hammerwork::JobQueue::new(pool));
    /// # let handler: hammerwork::worker::JobHandler = Arc::new(|job| Box::pin(async move { Ok(()) }));
    ///
    /// // Longer delay for API rate limit recovery
    /// let api_worker = Worker::new(queue, "api".to_string(), handler)
    ///     .with_retry_delay(Duration::from_secs(300)); // 5 minutes
    /// # Ok(())
    /// # }
    /// ```
    pub fn with_retry_delay(mut self, delay: Duration) -> Self {
        self.retry_delay = delay;
        self
    }

    /// Sets a default retry strategy for all jobs processed by this worker.
    ///
    /// This retry strategy will be used for jobs that don't have their own
    /// retry strategy configured. Jobs with their own retry strategy will
    /// use that instead of this default.
    ///
    /// If no default retry strategy is set, the worker will fall back to
    /// using the fixed `retry_delay` for all retries.
    ///
    /// # Arguments
    ///
    /// * `strategy` - The default retry strategy to use for failed jobs
    ///
    /// # Examples
    ///
    #[cfg_attr(feature = "postgres", doc = "```rust,no_run")]
    #[cfg_attr(not(feature = "postgres"), doc = "```rust,ignore")]
    /// use hammerwork::{Worker, retry::RetryStrategy};
    /// use std::time::Duration;
    /// # use std::sync::Arc;
    /// # async fn example() -> Result<(), Box<dyn std::error::Error>> {
    /// # let pool = sqlx::PgPool::connect("postgresql://localhost/test").await?;
    /// # let queue = Arc::new(hammerwork::JobQueue::new(pool));
    /// # let handler: hammerwork::worker::JobHandler = Arc::new(|job| Box::pin(async move { Ok(()) }));
    ///
    /// // Use exponential backoff as default for all jobs
    /// let worker = Worker::new(queue, "api_calls".to_string(), handler)
    ///     .with_default_retry_strategy(RetryStrategy::exponential(
    ///         Duration::from_secs(1),
    ///         2.0,
    ///         Some(Duration::from_secs(10 * 60))
    ///     ));
    /// # Ok(())
    /// # }
    /// ```
    pub fn with_default_retry_strategy(mut self, strategy: RetryStrategy) -> Self {
        self.default_retry_strategy = Some(strategy);
        self
    }

    /// Sets a default timeout for all jobs processed by this worker.
    ///
    /// Jobs that don't have their own timeout setting will use this default.
    /// Job-specific timeouts always take precedence over worker defaults.
    /// `Duration::ZERO` means no default timeout (it would otherwise time every job
    /// out before its handler ran).
    ///
    /// # Arguments
    ///
    /// * `timeout` - Maximum time a job can run before being terminated
    ///
    /// # Examples
    ///
    #[cfg_attr(feature = "postgres", doc = "```rust,no_run")]
    #[cfg_attr(not(feature = "postgres"), doc = "```rust,ignore")]
    /// use hammerwork::Worker;
    /// use std::time::Duration;
    /// # use std::sync::Arc;
    /// # async fn example() -> Result<(), Box<dyn std::error::Error>> {
    /// # let pool = sqlx::PgPool::connect("postgresql://localhost/test").await?;
    /// # let queue = Arc::new(hammerwork::JobQueue::new(pool));
    /// # let handler: hammerwork::worker::JobHandler = Arc::new(|job| Box::pin(async move { Ok(()) }));
    ///
    /// // Set 5 minute default timeout for all jobs
    /// let worker = Worker::new(queue, "processing".to_string(), handler)
    ///     .with_default_timeout(Duration::from_secs(300));
    /// # Ok(())
    /// # }
    /// ```
    pub fn with_default_timeout(mut self, timeout: Duration) -> Self {
        self.default_timeout = Some(timeout);
        self
    }

    /// Set how long the worker waits for an in-flight job after shutdown is requested.
    ///
    /// When [`WorkerPool::shutdown`] (or the worker's shutdown channel) fires while a
    /// job is running, the worker stops polling but lets the job finish for up to this
    /// long. Only after the grace period is the handler cancelled; the job then stays
    /// `Running` until its lease expires and the stale job reaper reclaims it.
    ///
    /// Defaults to [`DEFAULT_SHUTDOWN_GRACE_PERIOD`] (30 seconds).
    ///
    /// # Examples
    ///
    #[cfg_attr(feature = "postgres", doc = "```rust,no_run")]
    #[cfg_attr(not(feature = "postgres"), doc = "```rust,ignore")]
    /// use hammerwork::{Worker, JobQueue};
    /// use std::{sync::Arc, time::Duration};
    ///
    /// # async fn example() -> Result<(), Box<dyn std::error::Error>> {
    /// # let pool = sqlx::PgPool::connect("postgresql://localhost/test").await?;
    /// # let queue = Arc::new(JobQueue::new(pool));
    /// # let handler: hammerwork::worker::JobHandler = Arc::new(|_job| Box::pin(async { Ok(()) }));
    /// let worker = Worker::new(queue, "emails".to_string(), handler)
    ///     .with_shutdown_grace_period(Duration::from_secs(120));
    /// # Ok(())
    /// # }
    /// ```
    pub fn with_shutdown_grace_period(mut self, grace_period: Duration) -> Self {
        self.shutdown_grace_period = grace_period;
        self
    }

    /// Set the lease the worker holds on a running job.
    ///
    /// The lease is taken in the same statement that claims the job
    /// ([`DatabaseQueue::dequeue_leased`]). While the handler runs, the worker records a
    /// heartbeat and extends the lease every third of this duration (via
    /// [`DatabaseQueue::heartbeat_job`]). If the worker process dies, the lease stops
    /// being renewed and [`DatabaseQueue::requeue_stale_jobs`] reclaims the job once it
    /// expires; a reaper never reclaims a job whose lease is still valid, whatever its
    /// own staleness window. Shorter leases recover crashed jobs faster at the cost of
    /// more heartbeat writes for long-running jobs; jobs that finish within a third of
    /// the lease never write one.
    ///
    /// `Duration::ZERO` takes a lease that never expires and sends no heartbeats: the
    /// reaper never reclaims this worker's jobs, so if the worker dies an operator has
    /// to re-run them (for example with `cargo hammerwork job retry`).
    ///
    /// Defaults to [`DEFAULT_LEASE_DURATION`] (5 minutes). Requires migration
    /// `015_add_job_leases`.
    pub fn with_lease_duration(mut self, lease: Duration) -> Self {
        self.lease_duration = lease;
        self
    }

    /// Set how often the worker's monitoring task runs. Without this, it runs at the
    /// [metrics collector](Self::with_metrics_collector)'s update interval
    /// (`MetricsConfig::update_interval`), or every [`DEFAULT_MONITORING_INTERVAL`]
    /// (30 seconds) without one.
    ///
    /// With a [metrics collector](Self::with_metrics_collector) or alerting configured,
    /// a background task updates the queue depth metric and checks the alert thresholds
    /// (queue depth, worker starvation, and with a statistics collector the error rate
    /// and processing time) at this interval. These checks never run in the job polling
    /// loop, so a slow alert target or the queue depth `COUNT(*)` never delays
    /// dequeuing. Values below 10ms are raised to 10ms.
    pub fn with_monitoring_interval(mut self, interval: Duration) -> Self {
        self.monitoring_interval = Some(interval);
        self
    }

    /// The lease this worker holds on running jobs. See [`Worker::with_lease_duration`].
    pub fn lease_duration(&self) -> Duration {
        self.lease_duration
    }

    /// Apply a [`WorkerConfig`](crate::config::WorkerConfig).
    ///
    /// Sets the poll interval, default job timeout, priority weights and default retry
    /// strategy. `pool_size` and the autoscaling settings apply to a [`WorkerPool`]; see
    /// [`WorkerPool::from_config`].
    pub fn with_config(self, config: &crate::config::WorkerConfig) -> Self {
        self.with_poll_interval(config.polling_interval)
            .with_default_timeout(config.job_timeout)
            .with_priority_weights(config.priority_weights.clone())
            .with_default_retry_strategy(config.retry_strategy.clone())
    }

    /// Apply the worker-related sections of a [`HammerworkConfig`](crate::HammerworkConfig).
    ///
    /// Applies `worker` (see [`with_config`](Self::with_config)), the throttle from
    /// `rate_limiting` that matches this worker's queue, and `alerting` (when the
    /// `alerting` feature is enabled). Events are not applied here because the
    /// [`EventManager`](crate::events::EventManager) must be shared with any webhook or
    /// stream managers: build it once with `EventManager::new(config.events.clone())`
    /// and pass it to [`with_event_manager`](Self::with_event_manager).
    ///
    /// # Examples
    ///
    #[cfg_attr(feature = "postgres", doc = "```rust,no_run")]
    #[cfg_attr(not(feature = "postgres"), doc = "```rust,ignore")]
    /// use hammerwork::{HammerworkConfig, JobQueue, Worker};
    /// use std::sync::Arc;
    ///
    /// # async fn example() -> hammerwork::Result<()> {
    /// let config = HammerworkConfig::from_file("hammerwork.toml")?;
    /// let queue = Arc::new(JobQueue::<sqlx::Postgres>::from_config(&config).await?);
    /// let handler: hammerwork::worker::JobHandler = Arc::new(|_job| Box::pin(async { Ok(()) }));
    ///
    /// let worker = Worker::new(queue, "email".to_string(), handler)
    ///     .with_hammerwork_config(&config);
    /// # Ok(())
    /// # }
    /// ```
    pub fn with_hammerwork_config(mut self, config: &crate::HammerworkConfig) -> Self {
        self = self.with_config(&config.worker);

        if let Some(throttle) = config.rate_limiting.throttle_for(&self.queue_name) {
            self = self.with_throttle_config(throttle);
        }

        #[cfg(feature = "alerting")]
        {
            self = self.with_alerting_config(config.alerting.clone());
        }

        self
    }

    /// Configure priority weights for job selection
    pub fn with_priority_weights(mut self, weights: PriorityWeights) -> Self {
        self.priority_weights = Some(weights);
        self
    }

    /// Enable strict priority mode (always process highest priority first)
    pub fn with_strict_priority(mut self) -> Self {
        self.priority_weights = Some(PriorityWeights::strict());
        self
    }

    /// Use default weighted priority selection
    pub fn with_weighted_priority(mut self) -> Self {
        self.priority_weights = Some(PriorityWeights::new());
        self
    }

    /// Configure rate limiting for this worker
    pub fn with_rate_limit(mut self, rate_limit: RateLimit) -> Self {
        self.rate_limiter = Some(RateLimiter::new(rate_limit));
        self
    }

    /// Configure throttling for this worker.
    ///
    /// - `rate_per_minute` limits how often jobs are dequeued.
    /// - `max_concurrent` limits how many jobs run at once (at least one).
    /// - `backoff_on_error` is the base delay after a polling error.
    ///
    /// The limits are shared by every clone of this worker, so they apply to a whole
    /// [`WorkerPool`] built from it. A throttle with `enabled == false` is ignored.
    pub fn with_throttle_config(mut self, throttle_config: ThrottleConfig) -> Self {
        if !throttle_config.enabled {
            return self;
        }
        // If the throttle config has a rate limit, create a rate limiter
        if let Some(rate_limit) = throttle_config.to_rate_limit() {
            self.rate_limiter = Some(RateLimiter::new(rate_limit));
        }
        self.concurrency_limit = throttle_config.max_concurrent.map(|max| {
            let permits = usize::try_from(max)
                .unwrap_or(usize::MAX)
                .clamp(1, tokio::sync::Semaphore::MAX_PERMITS);
            Arc::new(tokio::sync::Semaphore::new(permits))
        });
        self.throttle_config = Some(throttle_config);
        self
    }

    /// Configure Prometheus metrics collection for this worker
    #[cfg(feature = "metrics")]
    pub fn with_metrics_collector(
        mut self,
        metrics_collector: Arc<PrometheusMetricsCollector>,
    ) -> Self {
        self.metrics_collector = Some(metrics_collector);
        self
    }

    /// Configure alerting for this worker
    #[cfg(feature = "alerting")]
    pub fn with_alerting_config(mut self, alerting_config: AlertingConfig) -> Self {
        self.alert_manager = Some(Arc::new(AlertManager::new(alerting_config)));
        self
    }

    /// Configure alert manager for this worker
    #[cfg(feature = "alerting")]
    pub fn with_alert_manager(mut self, alert_manager: Arc<AlertManager>) -> Self {
        self.alert_manager = Some(alert_manager);
        self
    }

    /// Add an event manager for publishing job lifecycle events.
    ///
    /// The event manager will receive job lifecycle events which can be delivered
    /// to webhooks, streaming systems, and other external integrations.
    ///
    /// # Arguments
    ///
    /// * `event_manager` - The event manager to use for publishing events
    ///
    /// # Examples
    ///
    #[cfg_attr(feature = "postgres", doc = "```rust,no_run")]
    #[cfg_attr(not(feature = "postgres"), doc = "```rust,ignore")]
    /// use hammerwork::{Worker, JobQueue, events::EventManager};
    /// use std::sync::Arc;
    ///
    /// # async fn example() -> Result<(), Box<dyn std::error::Error>> {
    /// # let pool = sqlx::PgPool::connect("postgresql://localhost/test").await?;
    /// let queue = Arc::new(JobQueue::new(pool));
    /// let event_manager = Arc::new(EventManager::new_default());
    ///
    /// let handler: Arc<dyn Fn(hammerwork::Job) -> std::pin::Pin<Box<dyn std::future::Future<Output = Result<(), hammerwork::HammerworkError>> + Send>> + Send + Sync> = Arc::new(|_job| Box::pin(async move { Ok(()) }));
    /// let worker = Worker::new(queue, "default".to_string(), handler)
    ///     .with_event_manager(event_manager);
    /// # Ok(())
    /// # }
    /// ```
    #[cfg(feature = "webhooks")]
    pub fn with_event_manager(mut self, event_manager: Arc<EventManager>) -> Self {
        self.event_manager = Some(event_manager);
        self
    }

    /// Enable batch processing: claim jobs in batches, and track statistics of jobs that
    /// belong to a [`JobBatch`](crate::batch::JobBatch) ([`get_batch_stats`](Self::get_batch_stats)).
    ///
    /// In batch mode the worker claims up to [`batch_size`](Self::with_batch_size) jobs
    /// per poll in one round trip ([`DatabaseQueue::dequeue_batch_leased`]) and runs them
    /// [`batch_concurrency`](Self::with_batch_concurrency) at a time, one at a time by
    /// default. Every claimed job goes through the same path as a single claim: its own
    /// timeout, retries, hooks, events, spawning, result storage and statistics.
    ///
    /// - **Leases.** Each claimed job holds its own lease from the moment it is claimed.
    ///   While a job waits for its turn, the worker renews its lease with heartbeats
    ///   every third of the [lease duration](Self::with_lease_duration), as it does for
    ///   a running job, so a waiting job is never reclaimed by the stale job reaper
    ///   while the worker is alive, however long the jobs ahead of it take. If a
    ///   waiting job's lease is lost anyway (an operator changed it, or heartbeats
    ///   failed for longer than the lease), the worker skips it.
    /// - **Start time.** A claim stamps the jobs' `started_at` with the claim time. A
    ///   job that waited for its turn records its actual start when it starts
    ///   ([`DatabaseQueue::start_claimed_run`]), so the time it waited does not count as
    ///   processing time in statistics.
    /// - **Throttle and rate limit.** With a throttle's `max_concurrent`, every claimed
    ///   job holds one of its permits until it finishes, so the worker never claims more
    ///   jobs than there are free permits (it waits for at least one). With a rate limit,
    ///   it claims no more jobs than it can take tokens for and hands back the tokens of
    ///   jobs it did not get, or that it did not run (released, skipped because their
    ///   lease was lost, or failed with their batch).
    /// - **Pause.** Before starting a claimed job, the worker checks whether the queue
    ///   was paused (at most once per [poll interval](Self::with_poll_interval)). If it
    ///   was, the claimed jobs that have not started are handed back to the queue as on
    ///   shutdown, and the running ones finish.
    /// - **Fail-fast batches.** At the same checks, the worker checks whether the
    ///   [`JobBatch`](crate::batch::JobBatch) of a claimed job that has not started has
    ///   failed. A [`FailFast`](crate::batch::PartialFailureMode::FailFast) batch fails
    ///   its `Pending` jobs on its first failure; the jobs this worker holds are
    ///   `Running`, so the worker fails them the same way (`Failed`, not retried)
    ///   instead of starting them, including when it hands jobs back on a pause or
    ///   shutdown. Jobs of the batch that are already running finish.
    /// - **Shutdown.** Claimed jobs that have not started are handed back to the queue
    ///   at once ([`DatabaseQueue::release_job_run`]), without counting an attempt, so
    ///   other workers can run them instead of waiting for their leases to expire. Jobs
    ///   already running get the [shutdown grace period](Self::with_shutdown_grace_period),
    ///   as in single-job mode.
    ///
    /// Batch claims cut the database round trips per job; holding jobs a worker has not
    /// started yet keeps them from idle workers, so keep batches small relative to the
    /// queue when jobs are slow. With batch processing off (the default), the worker
    /// claims and runs one job at a time.
    ///
    /// # Examples
    ///
    #[cfg_attr(feature = "postgres", doc = "```rust,no_run")]
    #[cfg_attr(not(feature = "postgres"), doc = "```rust,ignore")]
    /// use hammerwork::Worker;
    /// # use std::sync::Arc;
    /// # async fn example() -> Result<(), Box<dyn std::error::Error>> {
    /// # let pool = sqlx::PgPool::connect("postgresql://localhost/test").await?;
    /// # let queue = Arc::new(hammerwork::JobQueue::new(pool));
    /// # let handler: hammerwork::worker::JobHandler = Arc::new(|job| Box::pin(async move { Ok(()) }));
    ///
    /// // Keep statistics for job batches; jobs are still claimed one at a time. Use
    /// // `with_batch_size` to claim several jobs per poll.
    /// let worker = Worker::new(queue, "batch_queue".to_string(), handler)
    ///     .with_batch_processing_enabled(true);
    /// # Ok(())
    /// # }
    /// ```
    pub fn with_batch_processing_enabled(mut self, enabled: bool) -> Self {
        self.batch_processing_enabled = enabled;
        self
    }

    /// Claim up to `size` jobs per poll, and enable batch processing (see
    /// [`with_batch_processing_enabled`](Self::with_batch_processing_enabled)).
    ///
    /// Defaults to [`DEFAULT_BATCH_SIZE`] (1). Values below 1 are raised to 1, and the
    /// built-in backends claim at most
    /// [`MAX_DEQUEUE_BATCH_SIZE`](crate::queue::MAX_DEQUEUE_BATCH_SIZE) jobs at once.
    ///
    /// # Examples
    ///
    #[cfg_attr(feature = "postgres", doc = "```rust,no_run")]
    #[cfg_attr(not(feature = "postgres"), doc = "```rust,ignore")]
    /// use hammerwork::Worker;
    /// # use std::sync::Arc;
    /// # async fn example() -> Result<(), Box<dyn std::error::Error>> {
    /// # let pool = sqlx::PgPool::connect("postgresql://localhost/test").await?;
    /// # let queue = Arc::new(hammerwork::JobQueue::new(pool));
    /// # let handler: hammerwork::worker::JobHandler = Arc::new(|job| Box::pin(async move { Ok(()) }));
    ///
    /// // Claim up to 50 jobs per poll and run up to 8 of them at once.
    /// let worker = Worker::new(queue, "emails".to_string(), handler)
    ///     .with_batch_size(50)
    ///     .with_batch_concurrency(8);
    /// assert!(worker.is_batch_processing_enabled());
    /// # Ok(())
    /// # }
    /// ```
    pub fn with_batch_size(mut self, size: usize) -> Self {
        self.batch_size = size.max(1);
        self.batch_processing_enabled = true;
        self
    }

    /// Run up to `concurrency` of a claimed batch's jobs at once (batch mode only).
    ///
    /// Defaults to [`DEFAULT_BATCH_CONCURRENCY`] (1: the jobs of a batch run one after
    /// another). The jobs run concurrently on the worker's task, so a handler that blocks
    /// its thread holds up the others; use `tokio::task::spawn_blocking` for blocking
    /// work. Values below 1 are raised to 1. A throttle's `max_concurrent` still applies.
    pub fn with_batch_concurrency(mut self, concurrency: usize) -> Self {
        self.batch_concurrency = concurrency.max(1);
        self
    }

    /// Whether batch processing is enabled (see
    /// [`with_batch_processing_enabled`](Self::with_batch_processing_enabled)).
    pub fn is_batch_processing_enabled(&self) -> bool {
        self.batch_processing_enabled
    }

    /// The most jobs this worker claims per poll in batch mode. See
    /// [`with_batch_size`](Self::with_batch_size).
    pub fn batch_size(&self) -> usize {
        self.batch_size
    }

    /// The most claimed jobs this worker runs at once in batch mode. See
    /// [`with_batch_concurrency`](Self::with_batch_concurrency).
    pub fn batch_concurrency(&self) -> usize {
        self.batch_concurrency
    }

    /// Set the job lifecycle event hooks for this worker.
    ///
    /// Event hooks allow you to register callbacks that will be called at various
    /// points in the job lifecycle, such as when jobs start, complete, fail, timeout,
    /// or are retried. This is useful for distributed tracing, logging, metrics,
    /// and debugging.
    ///
    /// # Arguments
    ///
    /// * `hooks` - The event hooks to register
    ///
    /// # Examples
    ///
    #[cfg_attr(feature = "postgres", doc = "```rust,no_run")]
    #[cfg_attr(not(feature = "postgres"), doc = "```rust,ignore")]
    /// # use hammerwork::{Worker, worker::{JobEventHooks, JobHookEvent}};
    /// # use std::sync::Arc;
    /// # async fn example() -> Result<(), Box<dyn std::error::Error>> {
    /// # let queue = Arc::new(hammerwork::JobQueue::new(sqlx::PgPool::connect("").await?));
    /// # let handler: hammerwork::worker::JobHandler = Arc::new(|job| Box::pin(async move { Ok(()) }));
    ///
    /// let hooks = JobEventHooks::new()
    ///     .on_start(|event: JobHookEvent| {
    ///         println!("Job {} started at {}", event.job.id, event.timestamp);
    ///     })
    ///     .on_complete(|event: JobHookEvent| {
    ///         if let Some(duration) = event.duration {
    ///             println!("Job {} completed in {:?}", event.job.id, duration);
    ///         }
    ///     })
    ///     .on_fail(|event: JobHookEvent| {
    ///         if let Some(error) = &event.error {
    ///             println!("Job {} failed: {}", event.job.id, error);
    ///         }
    ///     });
    ///
    /// let worker = Worker::new(queue, "traced_queue".to_string(), handler)
    ///     .with_event_hooks(hooks);
    /// # Ok(())
    /// # }
    /// ```
    pub fn with_event_hooks(mut self, hooks: JobEventHooks) -> Self {
        self.event_hooks = hooks;
        self
    }

    /// Set a spawn manager for dynamic job spawning.
    ///
    /// The spawn manager handles creating child jobs when parent jobs complete successfully.
    /// Jobs with registered spawn handlers will automatically spawn child jobs based on
    /// their payload and configuration.
    ///
    /// # Arguments
    ///
    /// * `spawn_manager` - The spawn manager to use for this worker
    ///
    /// # Examples
    ///
    /// ```rust,no_run
    /// use hammerwork::{Worker, spawn::SpawnManager};
    /// use std::sync::Arc;
    ///
    /// # #[cfg(feature = "postgres")]
    /// # async fn example() -> Result<(), Box<dyn std::error::Error>> {
    /// // In real usage, you'd create a database connection pool
    /// # let pool = sqlx::PgPool::connect("postgresql://localhost/test").await?;
    /// # let queue = Arc::new(hammerwork::JobQueue::new(pool));
    /// let handler: hammerwork::worker::JobHandler = Arc::new(|job| Box::pin(async move { Ok(()) }));
    /// let mut spawn_manager: SpawnManager<sqlx::Postgres> = SpawnManager::new();
    /// // Register spawn handlers...
    ///
    /// let worker = Worker::new(queue, "queue".to_string(), handler)
    ///     .with_spawn_manager(Arc::new(spawn_manager));
    /// # Ok(())
    /// # }
    /// ```
    pub fn with_spawn_manager(
        mut self,
        spawn_manager: Arc<crate::spawn::SpawnManager<DB>>,
    ) -> Self {
        self.spawn_manager = Some(spawn_manager);
        self
    }

    /// Set a job start event handler for this worker.
    ///
    /// This is a convenience method for setting just the start event handler.
    /// For multiple event handlers, use `with_event_hooks()`.
    ///
    /// # Arguments
    ///
    /// * `handler` - Function to call when a job starts processing
    ///
    /// # Examples
    ///
    #[cfg_attr(feature = "postgres", doc = "```rust,no_run")]
    #[cfg_attr(not(feature = "postgres"), doc = "```rust,ignore")]
    /// # use hammerwork::Worker;
    /// # use std::sync::Arc;
    /// # async fn example() -> Result<(), Box<dyn std::error::Error>> {
    /// # let queue = Arc::new(hammerwork::JobQueue::new(sqlx::PgPool::connect("").await?));
    /// # let handler: hammerwork::worker::JobHandler = Arc::new(|job| Box::pin(async move { Ok(()) }));
    ///
    /// let worker = Worker::new(queue, "queue".to_string(), handler)
    ///     .on_job_start(|event| {
    ///         println!("Starting job: {}", event.job.id);
    ///     });
    /// # Ok(())
    /// # }
    /// ```
    pub fn on_job_start<F>(mut self, handler: F) -> Self
    where
        F: Fn(JobHookEvent) + Send + Sync + 'static,
    {
        self.event_hooks.on_job_start = Some(Arc::new(handler));
        self
    }

    /// Set a job completion event handler for this worker.
    ///
    /// # Arguments
    ///
    /// * `handler` - Function to call when a job completes successfully
    ///
    /// # Examples
    ///
    #[cfg_attr(feature = "postgres", doc = "```rust,no_run")]
    #[cfg_attr(not(feature = "postgres"), doc = "```rust,ignore")]
    /// # use hammerwork::Worker;
    /// # use std::sync::Arc;
    /// # async fn example() -> Result<(), Box<dyn std::error::Error>> {
    /// # let queue = Arc::new(hammerwork::JobQueue::new(sqlx::PgPool::connect("").await?));
    /// # let handler: hammerwork::worker::JobHandler = Arc::new(|job| Box::pin(async move { Ok(()) }));
    ///
    /// let worker = Worker::new(queue, "queue".to_string(), handler)
    ///     .on_job_complete(|event| {
    ///         if let Some(duration) = event.duration {
    ///             println!("Job {} completed in {:?}", event.job.id, duration);
    ///         }
    ///     });
    /// # Ok(())
    /// # }
    /// ```
    pub fn on_job_complete<F>(mut self, handler: F) -> Self
    where
        F: Fn(JobHookEvent) + Send + Sync + 'static,
    {
        self.event_hooks.on_job_complete = Some(Arc::new(handler));
        self
    }

    /// Set a job failure event handler for this worker.
    ///
    /// # Arguments
    ///
    /// * `handler` - Function to call when a job fails
    ///
    /// # Examples
    ///
    #[cfg_attr(feature = "postgres", doc = "```rust,no_run")]
    #[cfg_attr(not(feature = "postgres"), doc = "```rust,ignore")]
    /// # use hammerwork::Worker;
    /// # use std::sync::Arc;
    /// # async fn example() -> Result<(), Box<dyn std::error::Error>> {
    /// # let queue = Arc::new(hammerwork::JobQueue::new(sqlx::PgPool::connect("").await?));
    /// # let handler: hammerwork::worker::JobHandler = Arc::new(|job| Box::pin(async move { Ok(()) }));
    ///
    /// let worker = Worker::new(queue, "queue".to_string(), handler)
    ///     .on_job_fail(|event| {
    ///         if let Some(error) = &event.error {
    ///             eprintln!("Job {} failed: {}", event.job.id, error);
    ///         }
    ///     });
    /// # Ok(())
    /// # }
    /// ```
    pub fn on_job_fail<F>(mut self, handler: F) -> Self
    where
        F: Fn(JobHookEvent) + Send + Sync + 'static,
    {
        self.event_hooks.on_job_fail = Some(Arc::new(handler));
        self
    }

    /// Set a job timeout event handler for this worker.
    ///
    /// # Arguments
    ///
    /// * `handler` - Function to call when a job times out
    ///
    /// # Examples
    ///
    #[cfg_attr(feature = "postgres", doc = "```rust,no_run")]
    #[cfg_attr(not(feature = "postgres"), doc = "```rust,ignore")]
    /// # use hammerwork::Worker;
    /// # use std::sync::Arc;
    /// # async fn example() -> Result<(), Box<dyn std::error::Error>> {
    /// # let queue = Arc::new(hammerwork::JobQueue::new(sqlx::PgPool::connect("").await?));
    /// # let handler: hammerwork::worker::JobHandler = Arc::new(|job| Box::pin(async move { Ok(()) }));
    ///
    /// let worker = Worker::new(queue, "queue".to_string(), handler)
    ///     .on_job_timeout(|event| {
    ///         println!("Job {} timed out after {:?}", event.job.id, event.duration);
    ///     });
    /// # Ok(())
    /// # }
    /// ```
    pub fn on_job_timeout<F>(mut self, handler: F) -> Self
    where
        F: Fn(JobHookEvent) + Send + Sync + 'static,
    {
        self.event_hooks.on_job_timeout = Some(Arc::new(handler));
        self
    }

    /// Set a job retry event handler for this worker.
    ///
    /// # Arguments
    ///
    /// * `handler` - Function to call when a job is retried
    ///
    /// # Examples
    ///
    #[cfg_attr(feature = "postgres", doc = "```rust,no_run")]
    #[cfg_attr(not(feature = "postgres"), doc = "```rust,ignore")]
    /// # use hammerwork::Worker;
    /// # use std::sync::Arc;
    /// # async fn example() -> Result<(), Box<dyn std::error::Error>> {
    /// # let queue = Arc::new(hammerwork::JobQueue::new(sqlx::PgPool::connect("").await?));
    /// # let handler: hammerwork::worker::JobHandler = Arc::new(|job| Box::pin(async move { Ok(()) }));
    ///
    /// let worker = Worker::new(queue, "queue".to_string(), handler)
    ///     .on_job_retry(|event| {
    ///         println!("Retrying job {} due to: {:?}", event.job.id, event.error);
    ///     });
    /// # Ok(())
    /// # }
    /// ```
    pub fn on_job_retry<F>(mut self, handler: F) -> Self
    where
        F: Fn(JobHookEvent) + Send + Sync + 'static,
    {
        self.event_hooks.on_job_retry = Some(Arc::new(handler));
        self
    }

    /// Run the worker until a shutdown signal arrives on `shutdown_rx` (or its sender
    /// is dropped).
    ///
    /// Shutdown is graceful: while idle the worker stops immediately, but a job that is
    /// already running gets up to the configured
    /// [shutdown grace period](Worker::with_shutdown_grace_period) to finish and record
    /// its outcome before it is cancelled.
    ///
    /// Errors inside the loop (database failures while dequeuing or recording a job's
    /// outcome) are logged and followed by an exponential backoff starting at the
    /// throttle's `backoff_on_error` (or the poll interval), so a failing database never
    /// causes a hot loop. A panicking job handler fails the job instead of killing the
    /// worker.
    pub async fn run(&self, mut shutdown_rx: mpsc::Receiver<()>) -> Result<()> {
        info!("Worker started for queue: {}", self.queue_name);

        // Start background monitoring task for metrics and alerting. The guard aborts it
        // however `run` ends: a normal return, an error, a panic, or this future being
        // dropped (for example by a pool restarting the worker).
        #[cfg(any(feature = "metrics", feature = "alerting"))]
        let _monitoring_task = AbortOnDrop(self.start_monitoring_task());

        let mut consecutive_errors: u32 = 0;

        loop {
            // Idle phase. Only the waits (poll interval, rate limiter) are interrupted by
            // shutdown; the dequeue itself always runs to completion. Dropping a dequeue
            // future mid-transaction can return a pooled connection with an open
            // transaction and row locks still held.
            if shutdown_requested(&mut shutdown_rx) {
                info!("Worker shutting down for queue: {}", self.queue_name);
                break;
            }

            // Single-job mode: a throttle permit held until this iteration is over.
            let _permit;
            let (outcome, shutting_down) = if self.batch_processing_enabled {
                // Each claimed job holds its own throttle permit (see `acquire_batch`).
                match self.acquire_batch(&mut shutdown_rx).await {
                    Ok(Acquired::Shutdown) => {
                        info!("Worker shutting down for queue: {}", self.queue_name);
                        break;
                    }
                    Ok(Acquired::Work(batch)) => {
                        self.run_batch_until_done(batch, &mut shutdown_rx).await
                    }
                    Ok(Acquired::Idle) => (Ok(()), false),
                    Err(e) => {
                        error!("Error dequeuing jobs from queue {}: {}", self.queue_name, e);
                        (Err(e), false)
                    }
                }
            } else {
                // Held until this iteration's job is done, so at most `max_concurrent`
                // workers sharing this throttle dequeue or run jobs at once.
                _permit = match &self.concurrency_limit {
                    Some(limit) => tokio::select! {
                        biased;
                        _ = shutdown_rx.recv() => {
                            info!("Worker shutting down for queue: {}", self.queue_name);
                            break;
                        }
                        // The semaphore is never closed, so this is always a permit.
                        permit = Arc::clone(limit).acquire_owned() => permit.ok(),
                    },
                    None => None,
                };
                let acquired = self.acquire_job(&mut shutdown_rx).await;

                match acquired {
                    Ok(Acquired::Shutdown) => {
                        info!("Worker shutting down for queue: {}", self.queue_name);
                        break;
                    }
                    Ok(Acquired::Work(job)) => {
                        self.run_job_until_done(*job, &mut shutdown_rx).await
                    }
                    Ok(Acquired::Idle) => (Ok(()), false),
                    Err(e) => {
                        error!("Error dequeuing job from queue {}: {}", self.queue_name, e);
                        (Err(e), false)
                    }
                }
            };

            match outcome {
                Ok(()) => consecutive_errors = 0,
                Err(e) => {
                    consecutive_errors = consecutive_errors.saturating_add(1);
                    if !shutting_down {
                        let backoff = error_backoff(self.error_backoff_base(), consecutive_errors);
                        warn!(
                            "Worker for queue {} hit an error ({} in a row), backing off for {:?}: {}",
                            self.queue_name, consecutive_errors, backoff, e
                        );
                        tokio::select! {
                            biased;
                            _ = shutdown_rx.recv() => {
                                info!("Worker shutting down for queue: {}", self.queue_name);
                                break;
                            }
                            _ = sleep(backoff) => {}
                        }
                    }
                }
            }

            if shutting_down {
                info!("Worker shutting down for queue: {}", self.queue_name);
                break;
            }
        }

        Ok(())
    }

    /// Process `job` to completion. If shutdown is requested meanwhile, keep going for
    /// up to the shutdown grace period, then cancel.
    ///
    /// Returns the processing outcome and whether shutdown was requested.
    async fn run_job_until_done(
        &self,
        job: Job,
        shutdown_rx: &mut mpsc::Receiver<()>,
    ) -> (Result<()>, bool) {
        let job_id = job.id;
        let processing = self.process_job(job);
        tokio::pin!(processing);

        let outcome = tokio::select! {
            biased;
            outcome = &mut processing => return (self.log_process_outcome(job_id, outcome), false),
            _ = shutdown_rx.recv() => {
                info!(
                    "Shutdown requested while job {} is running; waiting up to {:?} for it to finish",
                    job_id, self.shutdown_grace_period
                );
                tokio::time::timeout(self.shutdown_grace_period, &mut processing).await
            }
        };

        match outcome {
            Ok(outcome) => (self.log_process_outcome(job_id, outcome), true),
            Err(_) => {
                warn!(
                    "Job {} did not finish within the {:?} shutdown grace period and was \
                     cancelled; it stays Running until its lease expires and the stale job \
                     reaper reclaims it",
                    job_id, self.shutdown_grace_period
                );
                (Ok(()), true)
            }
        }
    }

    fn log_process_outcome(&self, job_id: crate::job::JobId, outcome: Result<()>) -> Result<()> {
        if let Err(ref e) = outcome {
            error!(
                "Failed to record the outcome of job {} on queue {}: {}",
                job_id, self.queue_name, e
            );
        }
        outcome
    }

    /// Base delay for the error backoff: the throttle's `backoff_on_error` if set,
    /// otherwise the poll interval.
    fn error_backoff_base(&self) -> Duration {
        self.throttle_config
            .as_ref()
            .and_then(|throttle| throttle.backoff_on_error)
            .unwrap_or(self.poll_interval)
    }

    /// Wait for capacity and try to dequeue a job.
    ///
    /// Returns `Idle` after waiting for the poll interval when the queue is empty or
    /// paused, `Shutdown` when shutdown is requested during a wait, and `Err` when the
    /// dequeue itself fails. Database calls are never interrupted by shutdown.
    async fn acquire_job(
        &self,
        shutdown_rx: &mut mpsc::Receiver<()>,
    ) -> Result<Acquired<Box<Job>>> {
        // Alert checks and queue depth metrics run in the monitoring task
        // (`start_monitoring_task`), never here: a slow alert endpoint or a COUNT(*) must
        // not hold up dequeuing.

        // Take a rate limit token before dequeuing. It is handed back when no job is
        // claimed, so empty or paused polls never spend the budget.
        let rate_token = match &self.rate_limiter {
            Some(rate_limiter) => {
                let permit = tokio::select! {
                    biased;
                    _ = shutdown_rx.recv() => return Ok(Acquired::Shutdown),
                    permit = rate_limiter.acquire() => permit,
                };
                if let Err(e) = permit {
                    warn!("Rate limiter error: {}", e);
                    return Ok(self.idle_wait(shutdown_rx).await);
                }
                Some(rate_limiter)
            }
            None => None,
        };
        let job = self.claim_job().await;
        if !matches!(job, Ok(Some(_)))
            && let Some(rate_limiter) = rate_token
        {
            rate_limiter.refund();
        }

        match job? {
            Some(job) => {
                debug!(
                    "Processing job: {} with priority: {:?}",
                    job.id, job.priority
                );
                Ok(Acquired::Work(Box::new(job)))
            }
            // No job available (or the queue is paused): wait before polling again.
            None => Ok(self.idle_wait(shutdown_rx).await),
        }
    }

    /// Claim the next job of this worker's queue, with this worker's lease.
    async fn claim_job(&self) -> Result<Option<Job>> {
        if self.queue_is_paused().await {
            return Ok(None);
        }

        self.queue
            .dequeue_leased(
                &self.queue_name,
                self.priority_weights.as_ref(),
                self.lease_duration,
            )
            .await
    }

    /// Whether this worker's queue is paused, so the dequeue can be skipped. A failed
    /// check counts as not paused: the dequeue checks the pause itself.
    async fn queue_is_paused(&self) -> bool {
        match self.queue.is_queue_paused(&self.queue_name).await {
            Ok(true) => {
                debug!(
                    "Queue '{}' is paused, skipping job dequeue",
                    self.queue_name
                );
                true
            }
            Ok(false) => false,
            Err(e) => {
                warn!("Failed to check queue pause status: {}", e);
                false
            }
        }
    }

    /// Batch mode: wait for capacity, then claim up to `batch_size` jobs in one call.
    ///
    /// Each claimed job needs a throttle permit (with `max_concurrent`) and a rate limit
    /// token (with a rate limit): the worker waits for the first of each, takes as many
    /// more as are free right now, up to the batch size, and claims no more jobs than
    /// that. Tokens for jobs it did not get are handed back; permits it did not need are
    /// dropped.
    async fn acquire_batch(
        &self,
        shutdown_rx: &mut mpsc::Receiver<()>,
    ) -> Result<Acquired<Vec<ClaimedJob>>> {
        let mut capacity = self.batch_size.max(1);

        let mut permits = Vec::new();
        if let Some(limit) = &self.concurrency_limit {
            let first = tokio::select! {
                biased;
                _ = shutdown_rx.recv() => return Ok(Acquired::Shutdown),
                // The semaphore is never closed, so this is always a permit.
                permit = Arc::clone(limit).acquire_owned() => permit.ok(),
            };
            permits.extend(first);
            while permits.len() < capacity {
                match Arc::clone(limit).try_acquire_owned() {
                    Ok(permit) => permits.push(permit),
                    Err(_) => break,
                }
            }
            capacity = permits.len();
            if capacity == 0 {
                return Ok(self.idle_wait(shutdown_rx).await);
            }
        }

        let mut tokens = 0;
        if let Some(rate_limiter) = &self.rate_limiter {
            let permit = tokio::select! {
                biased;
                _ = shutdown_rx.recv() => return Ok(Acquired::Shutdown),
                permit = rate_limiter.acquire() => permit,
            };
            if let Err(e) = permit {
                warn!("Rate limiter error: {}", e);
                return Ok(self.idle_wait(shutdown_rx).await);
            }
            tokens = 1;
            while tokens < capacity && rate_limiter.try_acquire() {
                tokens += 1;
            }
            capacity = tokens;
        }

        let claimed = if self.queue_is_paused().await {
            Ok(Vec::new())
        } else {
            self.queue
                .dequeue_batch_leased(
                    &self.queue_name,
                    self.priority_weights.as_ref(),
                    self.lease_duration,
                    capacity,
                )
                .await
        };
        if let Some(rate_limiter) = &self.rate_limiter {
            let used = claimed.as_ref().map_or(0, Vec::len);
            for _ in used..tokens {
                rate_limiter.refund();
            }
        }

        let jobs = claimed?;
        if jobs.is_empty() {
            // No job available (or the queue is paused): wait before polling again.
            return Ok(self.idle_wait(shutdown_rx).await);
        }
        debug!(
            "Claimed {} job(s) from queue {}",
            jobs.len(),
            self.queue_name
        );
        let mut permits = permits.into_iter();
        Ok(Acquired::Work(
            jobs.into_iter()
                .map(|job| ClaimedJob {
                    job,
                    permit: permits.next(),
                })
                .collect(),
        ))
    }

    /// Process the jobs of a claimed batch, up to `batch_concurrency` at a time, while
    /// renewing the leases of the ones still waiting for their turn.
    ///
    /// If shutdown is requested meanwhile, the jobs that have not started are released
    /// back to the queue at once, and the running ones get up to the shutdown grace
    /// period to finish before they are cancelled (as in
    /// [`run_job_until_done`](Self::run_job_until_done)).
    ///
    /// Returns the first error recording a job's outcome (the other jobs are still
    /// processed) and whether shutdown was requested.
    async fn run_batch_until_done(
        &self,
        batch: Vec<ClaimedJob>,
        shutdown_rx: &mut mpsc::Receiver<()>,
    ) -> (Result<()>, bool) {
        // The runs whose leases are renewed while they wait: the jobs as claimed (same
        // id, attempts and started_at), without their payloads.
        let held = std::sync::Mutex::new(HeldJobs {
            waiting: batch
                .iter()
                .map(|claimed| Job {
                    payload: serde_json::Value::Null,
                    ..claimed.job.clone()
                })
                .collect(),
            lost: Default::default(),
            renewed: {
                let claimed_at = std::time::Instant::now();
                batch
                    .iter()
                    .map(|claimed| (claimed.job.id, claimed_at))
                    .collect()
            },
        });
        let renew_after = self.held_renewal_interval();
        let lease_keeper = self.maintain_held_leases(&held);
        tokio::pin!(lease_keeper);

        let concurrency = self.batch_concurrency.max(1);
        let mut queued: std::collections::VecDeque<ClaimedJob> = batch.into();
        let mut running: Vec<BatchRun<'_>> = Vec::new();
        let mut first_error = None;
        let mut shutting_down = false;
        // The claim checked the pause and found the jobs runnable; check again before
        // starting held jobs, at most once per poll interval (as often as a single-job
        // worker polls), whether the queue was paused or a job's fail-fast batch failed.
        let check_interval = self.poll_interval.max(MIN_POLL_INTERVAL);
        let mut checked_at = std::time::Instant::now();
        let mut failed_batches = std::collections::HashSet::new();
        // Jobs started right after the claim keep the claim time as their start time;
        // the ones that waited record when they actually start.
        let mut initial_fill = true;

        loop {
            if !queued.is_empty()
                && running.len() < concurrency
                && checked_at.elapsed() >= check_interval
            {
                checked_at = std::time::Instant::now();
                self.find_failed_batches(&queued, &mut failed_batches).await;
                if self.queue_is_paused().await {
                    let unstarted: Vec<Job> = queued.drain(..).map(|claimed| claimed.job).collect();
                    {
                        let mut held = held.lock().unwrap_or_else(|poisoned| poisoned.into_inner());
                        held.waiting.clear();
                        held.renewed.clear();
                    }
                    info!(
                        "Queue {} was paused; handing back {} claimed job(s) that have not \
                         started",
                        self.queue_name,
                        unstarted.len()
                    );
                    self.hand_back_unstarted(unstarted, &failed_batches).await;
                }
            }

            while running.len() < concurrency
                && let Some(ClaimedJob { job, permit }) = queued.pop_front()
            {
                let start = Self::start_held(&held, job.id, renew_after);
                if start == HeldStart::Lost {
                    warn!(
                        "Skipping job {} of a claimed batch: it lost its lease while waiting \
                         (it is no longer Running this run)",
                        job.id
                    );
                    self.refund_rate_token();
                    continue;
                }
                let batch_failed = job.batch_id.is_some_and(|id| failed_batches.contains(&id));
                // Recording the start also renews the lease.
                let record_start = !initial_fill || start == HeldStart::RenewThenRun;
                running.push(Box::pin(async move {
                    // The job's throttle permit is held until it is done.
                    let _permit = permit;
                    let mut job = job;
                    let job_id = job.id;
                    if batch_failed {
                        return (job_id, self.fail_held_job(&job).await);
                    }
                    if record_start {
                        match self
                            .queue
                            .start_claimed_run(&mut job, self.lease_duration)
                            .await
                        {
                            Ok(true) => {}
                            Ok(false) => {
                                warn!(
                                    "Skipping job {} of a claimed batch: it lost its lease \
                                     while waiting (it is no longer Running this run)",
                                    job_id
                                );
                                self.refund_rate_token();
                                return (job_id, Ok(()));
                            }
                            Err(e) => warn!(
                                "Failed to record the start of job {} (and renew its lease): {}",
                                job_id, e
                            ),
                        }
                    }
                    (job_id, self.process_job(job).await)
                }));
            }
            initial_fill = false;

            tokio::select! {
                biased;
                _ = shutdown_rx.recv() => {
                    shutting_down = true;
                    break;
                }
                finished = next_finished(&mut running) => match finished {
                    Some((job_id, outcome)) => {
                        if let Err(e) = self.log_process_outcome(job_id, outcome) {
                            first_error.get_or_insert(e);
                        }
                    }
                    None => break,
                },
                () = &mut lease_keeper => {}
            }
        }

        if shutting_down {
            let unstarted = queued.drain(..).map(|claimed| claimed.job).collect();
            self.hand_back_unstarted(unstarted, &failed_batches).await;
            if !running.is_empty() {
                info!(
                    "Shutdown requested while {} job(s) of a batch are running; waiting up to \
                     {:?} for them to finish",
                    running.len(),
                    self.shutdown_grace_period
                );
                let drained = tokio::time::timeout(self.shutdown_grace_period, async {
                    while let Some((job_id, outcome)) = next_finished(&mut running).await {
                        if let Err(e) = self.log_process_outcome(job_id, outcome) {
                            first_error.get_or_insert(e);
                        }
                    }
                })
                .await;
                if drained.is_err() {
                    warn!(
                        "{} job(s) of a batch did not finish within the {:?} shutdown grace \
                         period and were cancelled; they stay Running until their leases \
                         expire and the stale job reaper reclaims them",
                        running.len(),
                        self.shutdown_grace_period
                    );
                }
            }
        }

        (first_error.map_or(Ok(()), Err), shutting_down)
    }

    /// How often held and running jobs renew their leases: a third of the lease.
    fn held_renewal_interval(&self) -> Duration {
        (self.lease_duration / 3).max(MIN_HEARTBEAT_INTERVAL)
    }

    /// Take `job_id` off the waiting list of a batch as it starts, and say whether it may
    /// run and whether its lease needs renewing first (its last renewal is older than
    /// `renew_after`, the heartbeat interval).
    fn start_held(
        held: &std::sync::Mutex<HeldJobs>,
        job_id: crate::job::JobId,
        renew_after: Duration,
    ) -> HeldStart {
        let mut held = held.lock().unwrap_or_else(|poisoned| poisoned.into_inner());
        held.waiting.retain(|run| run.id != job_id);
        let renewed = held.renewed.remove(&job_id);
        if held.lost.contains(&job_id) {
            HeldStart::Lost
        } else if renewed.is_some_and(|at| at.elapsed() < renew_after) {
            HeldStart::Run
        } else {
            HeldStart::RenewThenRun
        }
    }

    /// Renew the leases of a batch's jobs that are waiting for their turn, every third
    /// of the lease duration (like [`maintain_lease`](Self::maintain_lease) for a running
    /// job, which takes over once a job starts). A job whose lease is found lost is
    /// marked so it is not run. Never completes.
    async fn maintain_held_leases(&self, held: &std::sync::Mutex<HeldJobs>) {
        self.every_heartbeat(|| async move {
            let waiting = held
                .lock()
                .map(|held| held.waiting.clone())
                .unwrap_or_default();
            for run in waiting {
                match self.queue.heartbeat_job(&run, self.lease_duration).await {
                    Ok(true) => {
                        debug!("Renewed lease on waiting job {}", run.id);
                        let mut held = held.lock().unwrap_or_else(|poisoned| poisoned.into_inner());
                        if held.waiting.iter().any(|waiting| waiting.id == run.id) {
                            held.renewed.insert(run.id, std::time::Instant::now());
                        }
                    }
                    Ok(false) => {
                        warn!(
                            "Lost the lease on job {} while it waited in a claimed batch; it \
                             will not be run by this worker",
                            run.id
                        );
                        let mut held = held.lock().unwrap_or_else(|poisoned| poisoned.into_inner());
                        held.waiting.retain(|waiting| waiting.id != run.id);
                        held.lost.insert(run.id);
                    }
                    Err(e) => warn!("Failed to renew the lease on waiting job {}: {}", run.id, e),
                }
            }
            true
        })
        .await
    }

    /// Run `beat` every [heartbeat interval](Self::held_renewal_interval) (the first
    /// one after an interval) until it returns `false`. Never completes. With a zero
    /// lease, which never expires, `beat` never runs.
    async fn every_heartbeat<F, Fut>(&self, mut beat: F)
    where
        F: FnMut() -> Fut,
        Fut: Future<Output = bool>,
    {
        if !self.lease_duration.is_zero() {
            let interval = self.held_renewal_interval();
            loop {
                sleep(interval).await;
                if !beat().await {
                    break;
                }
            }
        }
        std::future::pending().await
    }

    /// Hand claimed jobs that never started back to the queue
    /// ([`DatabaseQueue::release_job_run`]) and refund their rate limit tokens. A job
    /// that cannot be released stays `Running` until its lease expires.
    async fn release_unstarted(&self, jobs: impl IntoIterator<Item = Job>) {
        let mut released = 0;
        for job in jobs {
            match self.queue.release_job_run(&job).await {
                Ok(true) => {
                    released += 1;
                    self.refund_rate_token();
                }
                Ok(false) => debug!(
                    "Job {} was no longer Running this run; nothing to release",
                    job.id
                ),
                Err(e) => warn!(
                    "Failed to release unstarted job {}; it stays Running until its lease \
                     expires: {}",
                    job.id, e
                ),
            }
        }
        if released > 0 {
            info!(
                "Released {} claimed job(s) that had not started back to queue {}",
                released, self.queue_name
            );
        }
    }

    /// Hand back claimed jobs that never started: the ones whose batch is in
    /// `failed_batches` are failed ([`fail_held_job`](Self::fail_held_job)), since
    /// releasing them would let them run; the others are released to the queue.
    async fn hand_back_unstarted(
        &self,
        jobs: Vec<Job>,
        failed_batches: &std::collections::HashSet<BatchId>,
    ) {
        let (doomed, released): (Vec<Job>, Vec<Job>) = jobs
            .into_iter()
            .partition(|job| job.batch_id.is_some_and(|id| failed_batches.contains(&id)));
        for job in doomed {
            if let Err(e) = self.fail_held_job(&job).await {
                warn!("Failed to fail job {} of a failed batch: {}", job.id, e);
            }
        }
        self.release_unstarted(released).await;
    }

    /// Add to `failed` the batches of the `queued` jobs that have failed. A fail-fast
    /// batch fails on its first terminal job failure, which fails its `Pending` jobs;
    /// held jobs are `Running`, so the worker has to stop them itself. A batch whose
    /// status cannot be read counts as not failed.
    async fn find_failed_batches(
        &self,
        queued: &std::collections::VecDeque<ClaimedJob>,
        failed: &mut std::collections::HashSet<BatchId>,
    ) {
        let mut batch_ids: Vec<BatchId> = queued
            .iter()
            .filter_map(|claimed| claimed.job.batch_id)
            .filter(|id| !failed.contains(id))
            .collect();
        batch_ids.sort_unstable();
        batch_ids.dedup();
        for batch_id in batch_ids {
            match self.queue.get_batch_status(batch_id).await {
                Ok(batch) if batch.status == crate::batch::BatchStatus::Failed => {
                    info!(
                        "Batch {} has failed; its claimed jobs that have not started will \
                         not run",
                        batch_id
                    );
                    failed.insert(batch_id);
                }
                Ok(_) => {}
                Err(e) => warn!("Failed to check the status of batch {}: {}", batch_id, e),
            }
        }
    }

    /// Fail a held job whose batch failed before it started, as a fail-fast batch fails
    /// its `Pending` jobs (status `Failed`, no retry). The job did not run, so its rate
    /// limit token is refunded. Nothing happens if the job is no longer `Running` the
    /// run this worker claimed.
    async fn fail_held_job(&self, job: &Job) -> Result<()> {
        // Renewing the lease confirms this worker still holds the run, and keeps the
        // reaper off it until it is failed.
        match self.queue.heartbeat_job(job, self.lease_duration).await {
            Ok(true) => {}
            Ok(false) => {
                debug!(
                    "Job {} of a failed batch is no longer Running this run; leaving it",
                    job.id
                );
                self.refund_rate_token();
                return Ok(());
            }
            Err(e) => warn!("Failed to renew the lease on job {}: {}", job.id, e),
        }
        match self
            .queue
            .fail_job(
                job.id,
                crate::queue::lifecycle::BATCH_FAILED_BEFORE_START_MESSAGE,
            )
            .await
        {
            Ok(()) => {
                info!(
                    "Failed claimed job {}: its batch failed (fail-fast) before it started",
                    job.id
                );
                self.refund_rate_token();
                Ok(())
            }
            // It already left Running (an operator, or the batch, finished it).
            Err(
                HammerworkError::InvalidJobTransition { .. } | HammerworkError::JobNotFound { .. },
            ) => {
                self.refund_rate_token();
                Ok(())
            }
            Err(e) => Err(e),
        }
    }

    /// Hand back the rate limit token of a claimed job that will not run.
    fn refund_rate_token(&self) {
        if let Some(rate_limiter) = &self.rate_limiter {
            rate_limiter.refund();
        }
    }

    /// Sleep for the poll interval (at least [`MIN_POLL_INTERVAL`]), returning early
    /// with `Shutdown` if shutdown is requested meanwhile.
    async fn idle_wait<T>(&self, shutdown_rx: &mut mpsc::Receiver<()>) -> Acquired<T> {
        tokio::select! {
            biased;
            _ = shutdown_rx.recv() => Acquired::Shutdown,
            _ = sleep(self.poll_interval.max(MIN_POLL_INTERVAL)) => Acquired::Idle,
        }
    }

    /// The number of attempts `job` gets on this worker: its own `max_attempts`,
    /// lowered by [`with_max_retries`](Self::with_max_retries) when that is set.
    fn attempt_limit(&self, job: &Job) -> i32 {
        effective_attempt_limit(job.max_attempts, self.max_retries)
    }

    /// When a failed or timed-out run of `job` should be retried.
    ///
    /// Uses, in order: the job's own retry strategy, the worker's default strategy,
    /// and the fixed retry delay.
    fn retry_at_for(&self, job: &Job) -> DateTime<Utc> {
        // The claim of this run already counted it in `attempts`, so after the first
        // failed run `attempts` is 1: the first retry (`calculate_delay(1)`, the
        // strategy's base delay).
        let retry = u32::try_from(job.attempts.max(1)).unwrap_or(u32::MAX);
        let retry_delay = if let Some(ref job_strategy) = job.retry_strategy {
            job_strategy.calculate_delay(retry)
        } else if let Some(ref default_strategy) = self.default_retry_strategy {
            default_strategy.calculate_delay(retry)
        } else {
            self.retry_delay
        };
        retry_at_from_delay(Utc::now(), retry_delay)
    }

    /// Record the outcome of this run of `job`.
    ///
    /// Returns `None` when the job is no longer `Running` this run: it was reclaimed by
    /// the stale-job reaper, changed by an operator, or is being run again by another
    /// worker. The outcome is then discarded so this (late) worker cannot overwrite the
    /// job's newer state, and the caller skips its hooks and events.
    async fn record_outcome(
        &self,
        job: &Job,
        outcome: JobOutcome,
    ) -> Result<Option<RecordedOutcome>> {
        self.record(job, outcome, None).await
    }

    /// Record the outcome of this run of `job`; with `children`, record its completion
    /// and enqueue the children in one transaction
    /// ([`DatabaseQueue::complete_job_run_with_children`]). See
    /// [`record_outcome`](Self::record_outcome).
    async fn record(
        &self,
        job: &Job,
        outcome: JobOutcome,
        children: Option<Vec<Job>>,
    ) -> Result<Option<RecordedOutcome>> {
        // Record the outcome in a spawned task so that it always runs to commit or
        // rollback, even if this future is dropped (for example when the shutdown grace
        // period expires). Cancelling a database transaction mid-flight can leave its
        // connection in the pool with the transaction still open.
        let queue = Arc::clone(&self.queue);
        let finished_job = job.clone();
        let recorded = tokio::spawn(async move {
            match children {
                Some(children) => {
                    queue
                        .complete_job_run_with_children(&finished_job, children)
                        .await
                }
                None => queue.finish_job_run(&finished_job, outcome).await,
            }
        })
        .await
        .map_err(|e| HammerworkError::Worker {
            message: format!("recording the outcome of job {} failed: {e}", job.id),
        })??;
        match recorded {
            Some(recorded) => {
                if let Some(next_run_at) = recorded.next_run_at {
                    info!(
                        "Rescheduled recurring job {} for its next run at {}",
                        job.id, next_run_at
                    );
                }
                if !recorded.unblocked.is_empty() {
                    debug!(
                        "Job {} completed; {} dependent job(s) are now runnable",
                        job.id,
                        recorded.unblocked.len()
                    );
                }
                if !recorded.cancelled.is_empty() {
                    warn!(
                        "Job {} failed; {} dependent, workflow or batch job(s) were failed with it",
                        job.id,
                        recorded.cancelled.len()
                    );
                }
                Ok(Some(recorded))
            }
            None => {
                warn!(
                    "Job {} is no longer running attempt {} (it was reclaimed as stale, \
                     changed by an operator or is running again elsewhere); discarding the \
                     outcome of this run",
                    job.id, job.attempts
                );
                Ok(None)
            }
        }
    }

    async fn process_job(&self, job: Job) -> Result<()> {
        let batch_id = job.batch_id;
        let start_time = Utc::now();

        // Create OpenTelemetry span for job processing
        #[cfg(feature = "tracing")]
        let _span = crate::tracing::create_job_span(&job, "job.process");

        // Update batch statistics if batch processing is enabled
        if self.batch_processing_enabled && batch_id.is_some() {
            self.update_batch_stats(|stats| {
                stats.jobs_processed += 1;
                stats.last_processed_job = Some(start_time);
            });
        }

        // Fire job start event hook
        self.event_hooks.fire_job_start(job.clone());

        // Record job started event
        self.record_event(
            &job,
            job_event(&job, JobEventType::Started, None, None, start_time),
        )
        .await;

        // The job's own max_attempts decides whether a failed run is retried.
        let attempts_left = job.attempts < self.attempt_limit(&job);

        // Only the handler sees an encrypted job's plaintext payload; hooks, events and
        // the recorded outcome use the stored (redacted) job. A payload that cannot be
        // decrypted (no engine, missing key, tampered data) fails this run.
        let handler_job = match self.queue.decrypt_job(job.clone()).await {
            Ok(handler_job) => handler_job,
            Err(e) => return self.handle_failure(&job, e, attempts_left).await,
        };

        // Spawning reads `_spawn_config` from the decrypted payload: for an encrypted job the
        // stored payload is a placeholder or redacted, so the configuration is only visible
        // here.
        let spawn_parent = self.spawn_manager.is_some().then(|| handler_job.clone());

        // Determine timeout duration (job-specific or default). A zero timeout, on the job
        // or as the worker default, means none (see `Job::with_timeout` and
        // `with_default_timeout`).
        let timeout_duration = job
            .timeout
            .filter(|timeout| !timeout.is_zero())
            .or(self.default_timeout.filter(|timeout| !timeout.is_zero()));

        // Ok(handler result), or Err(timeout) when the handler ran out of time.
        let handler_result = match timeout_duration {
            Some(timeout) => tokio::time::timeout(timeout, self.execute_handler(handler_job))
                .await
                .map_err(|_| timeout),
            None => Ok(self.execute_handler(handler_job).await),
        };

        match handler_result {
            Ok(Ok(job_result)) => {
                self.handle_success(&job, spawn_parent, job_result, start_time, attempts_left)
                    .await
            }
            Ok(Err(e)) => self.handle_failure(&job, e, attempts_left).await,
            Err(timeout) => self.handle_timeout(&job, timeout, attempts_left).await,
        }
    }

    /// The child jobs `job` spawns, with their configuration, or `None` when it spawns
    /// none (no spawn manager, no `_spawn_config`, or no handler for its queue).
    ///
    /// `spawn_parent` is the decrypted job: an encrypted job's stored payload does not
    /// carry the spawn configuration. A malformed configuration or a failing spawn
    /// handler is an error, which fails the run.
    async fn prepare_spawn(
        &self,
        job: &Job,
        spawn_parent: Option<&Job>,
    ) -> Result<Option<(Vec<Job>, crate::spawn::SpawnConfig)>> {
        let Some(spawn_manager) = &self.spawn_manager else {
            return Ok(None);
        };
        let parent = spawn_parent.unwrap_or(job);
        let Some(config) = crate::spawn::spawn_config_from_payload(&parent.payload)? else {
            return Ok(None);
        };
        let prepared = contain_panic(
            "Spawn handler",
            spawn_manager.prepare_spawn(parent, &config, self.queue.clone()),
        )
        .await?;
        match prepared {
            Some(children) => Ok(Some((children, config))),
            None => {
                debug!(
                    "No spawn handler registered for job type: {}",
                    job.queue_name
                );
                Ok(None)
            }
        }
    }

    async fn handle_success(
        &self,
        job: &Job,
        spawn_parent: Option<Job>,
        job_result: JobResult,
        start_time: DateTime<Utc>,
        attempts_left: bool,
    ) -> Result<()> {
        let job_id = job.id;
        debug!("Job {} completed successfully", job_id);

        let processing_time_ms = (Utc::now() - start_time).num_milliseconds().max(0) as u64;

        // Store the result before recording the completion, so that anyone who sees the
        // job as Completed can also read its result.
        if let Some(result_data) = job_result.data {
            let expires_at = job
                .result_config
                .ttl
                .map(|ttl| crate::queue::saturating_add_to(Utc::now(), ttl));

            match job.result_config.storage {
                crate::job::ResultStorage::Database => {
                    if let Err(e) = self
                        .queue
                        .store_job_result(job_id, result_data, expires_at)
                        .await
                    {
                        error!("Failed to store result for job {}: {}", job_id, e);
                    } else {
                        debug!("Stored result for job {}", job_id);
                    }
                }
                crate::job::ResultStorage::Memory => {
                    self.queue
                        .memory_results
                        .store(job_id, result_data, expires_at);
                    debug!("Stored result for job {} in memory", job_id);
                }
                crate::job::ResultStorage::None => {}
            }
        }

        // Build the children the job spawns before completing it: they are enqueued in
        // the transaction that completes the parent, and a spawn failure fails this run
        // (so it is retried) instead of completing the parent without its children.
        let spawn = match self.prepare_spawn(job, spawn_parent.as_ref()).await {
            Ok(spawn) => spawn,
            Err(e) => {
                warn!("Failed to spawn child jobs for job {}: {}", job_id, e);
                return self.handle_failure(job, e, attempts_left).await;
            }
        };
        let (children, spawn_config) = match spawn {
            Some((children, config)) => (Some(children), Some(config)),
            None => (None, None),
        };

        // Completing also reschedules a recurring job, and makes dependents whose
        // dependencies have all completed runnable, in the same transaction.
        let recorded = match self.record(job, JobOutcome::Completed, children).await {
            Ok(Some(recorded)) => recorded,
            Ok(None) => return Ok(()),
            // Enqueueing a child failed, so nothing was written and the job is still
            // Running this run: fail the run so it is retried.
            Err(e) if spawn_config.is_some() => {
                warn!("Failed to enqueue the child jobs of job {}: {}", job_id, e);
                return self.handle_failure(job, e, attempts_left).await;
            }
            Err(e) => return Err(e),
        };

        if let (Some(config), Some(spawn_manager)) = (spawn_config, &self.spawn_manager) {
            let spawn_result = crate::spawn::SpawnResult {
                parent_job_id: job_id,
                spawned_jobs: recorded.spawned.clone(),
                spawned_at: Utc::now(),
                spawn_operation_id: config.operation_id,
            };
            info!(
                "Job {} spawned {} child jobs: {:?}",
                job_id,
                spawn_result.spawned_jobs.len(),
                spawn_result.spawned_jobs
            );
            let parent = spawn_parent.as_ref().unwrap_or(job);
            if let Err(e) = contain_panic(
                "Spawn completion callback",
                spawn_manager.notify_spawn_complete(parent, &spawn_result),
            )
            .await
            {
                warn!("Spawn completion callback for job {} failed: {}", job_id, e);
            }
        }

        // Update batch statistics for successful completion
        if self.batch_processing_enabled
            && let Some(batch_id) = job.batch_id
        {
            self.update_batch_stats(|stats| {
                stats.jobs_completed += 1;
                stats.total_processing_time_ms += processing_time_ms;
                stats.update_average_processing_time();
            });

            if let Err(e) = self.check_and_update_batch_status(batch_id).await {
                warn!(
                    "Failed to update batch status for batch {}: {}",
                    batch_id, e
                );
            }
        }

        // Record span success status
        #[cfg(feature = "tracing")]
        {
            let span = tracing::Span::current();
            span.record("success", true);
            span.record("processing_time_ms", processing_time_ms);
        }

        // Fire job completion event hook
        let processing_duration = Duration::from_millis(processing_time_ms);
        self.event_hooks
            .fire_job_complete(job.clone(), processing_duration);

        // Record job completed event
        self.record_event(
            job,
            job_event(
                job,
                JobEventType::Completed,
                Some(processing_time_ms),
                None,
                Utc::now(),
            ),
        )
        .await;

        Ok(())
    }

    async fn handle_failure(
        &self,
        job: &Job,
        error: HammerworkError,
        attempts_left: bool,
    ) -> Result<()> {
        let job_id = job.id;
        error!("Job {} failed: {}", job_id, error);
        let error_message = error.to_string();

        // Record span error status
        #[cfg(feature = "tracing")]
        {
            let span = tracing::Span::current();
            span.record("error", true);
            span.record("error.type", "job_failure");
            span.record("error.message", &error_message);
            span.record("job.will_retry", attempts_left);
        }

        let outcome = if attempts_left {
            JobOutcome::Retry {
                retry_at: self.retry_at_for(job),
                error: error_message.clone(),
                timed_out: false,
            }
        } else {
            warn!(
                "Job {} failed on its last attempt ({} of {}), marking it dead",
                job_id,
                job.attempts,
                self.attempt_limit(job)
            );
            JobOutcome::Dead {
                error: error_message.clone(),
            }
        };
        let Some(recorded) = self.record_outcome(job, outcome).await? else {
            return Ok(());
        };

        self.after_failed_run(job, &error_message, attempts_left)
            .await;

        if attempts_left {
            info!(
                "Retrying job {} (attempt {} of {})",
                job_id,
                job.attempts + 1,
                self.attempt_limit(job)
            );
            self.announce_retry(job, error_message).await;
        } else {
            debug!("Job {} run ended with status {:?}", job_id, recorded.status);
            self.event_hooks
                .fire_job_fail(job.clone(), error_message.clone());
            self.record_event(
                job,
                job_event(
                    job,
                    JobEventType::Dead,
                    None,
                    Some(error_message),
                    Utc::now(),
                ),
            )
            .await;
        }

        Ok(())
    }

    /// A timed-out run is retried like a failure while the job has attempts left, and
    /// only ends the job as `TimedOut` on its last attempt.
    async fn handle_timeout(
        &self,
        job: &Job,
        timeout: Duration,
        attempts_left: bool,
    ) -> Result<()> {
        let job_id = job.id;
        let error_message = format!("Job timed out after {:?}", timeout);
        warn!("Job {} timed out after {:?}", job_id, timeout);

        let outcome = if attempts_left {
            JobOutcome::Retry {
                retry_at: self.retry_at_for(job),
                error: error_message.clone(),
                timed_out: true,
            }
        } else {
            JobOutcome::TimedOut {
                error: error_message.clone(),
            }
        };
        if self.record_outcome(job, outcome).await?.is_none() {
            return Ok(());
        }

        // Record span timeout status
        #[cfg(feature = "tracing")]
        {
            let span = tracing::Span::current();
            span.record("error", true);
            span.record("error.type", "timeout");
            span.record("error.message", &error_message);
            span.record("job.will_retry", attempts_left);
        }

        self.after_failed_run(job, &error_message, attempts_left)
            .await;

        // Fire job timeout event hook
        self.event_hooks.fire_job_timeout(job.clone(), timeout);

        // Record timeout event
        self.record_event(
            job,
            job_event(
                job,
                JobEventType::TimedOut,
                Some(timeout.as_millis() as u64),
                Some(error_message.clone()),
                Utc::now(),
            ),
        )
        .await;

        if attempts_left {
            info!(
                "Retrying timed-out job {} (attempt {} of {})",
                job_id,
                job.attempts + 1,
                self.attempt_limit(job)
            );
            self.announce_retry(job, error_message).await;
        }

        Ok(())
    }

    /// Fire the retry hook and record the `Retried` event of a failed or timed-out run.
    async fn announce_retry(&self, job: &Job, error_message: String) {
        self.event_hooks
            .fire_job_retry(job.clone(), error_message.clone());
        self.record_event(
            job,
            job_event(
                job,
                JobEventType::Retried,
                None,
                Some(error_message),
                Utc::now(),
            ),
        )
        .await;
    }

    /// Batch statistics after a failed (or timed-out) run.
    async fn after_failed_run(&self, job: &Job, error_message: &str, attempts_left: bool) {
        if !self.batch_processing_enabled {
            return;
        }
        let Some(batch_id) = job.batch_id else {
            return;
        };
        self.update_batch_stats(|stats| {
            stats.jobs_failed += 1;
        });
        if attempts_left {
            return;
        }
        if let Err(e) = self
            .handle_batch_job_failure(batch_id, job.id, error_message)
            .await
        {
            warn!(
                "Failed to handle batch job failure for batch {}: {}",
                batch_id, e
            );
        }
        if let Err(e) = self.check_and_update_batch_status(batch_id).await {
            warn!(
                "Failed to update batch status for batch {}: {}",
                batch_id, e
            );
        }
    }
    /// Create the handler future for `job`, unifying both handler types.
    fn start_handler(&self, job: Job) -> HandlerFuture {
        match &self.handler {
            JobHandlerType::Legacy(handler) => {
                // Execute legacy handler and convert () to JobResult
                let future = handler(job);
                Box::pin(async move { future.await.map(|_| JobResult::success()) })
            }
            // Execute enhanced handler directly
            JobHandlerType::WithResult(handler) => handler(job),
        }
    }

    /// Execute a job handler, returning a unified result.
    ///
    /// A panic in the handler (while creating or polling its future) is caught and
    /// turned into an error, so the job goes through the normal retry/failure path
    /// instead of killing the worker. While the handler runs, the job's lease is
    /// renewed in the background.
    async fn execute_handler(&self, job: Job) -> Result<JobResult> {
        let job_id = job.id;
        // The run whose lease is renewed: the job as dequeued (same id, attempts and
        // started_at), without its payload.
        let run = Job {
            payload: serde_json::Value::Null,
            ..job.clone()
        };
        let handler_future =
            match std::panic::catch_unwind(AssertUnwindSafe(|| self.start_handler(job))) {
                Ok(future) => future,
                Err(payload) => {
                    error!("Handler for job {} panicked", job_id);
                    return Err(handler_panic_error(payload.as_ref()));
                }
            };

        let guarded = CatchUnwind {
            inner: handler_future,
        };

        tokio::select! {
            biased;
            outcome = guarded => outcome.unwrap_or_else(|payload| {
                error!("Handler for job {} panicked", job_id);
                Err(handler_panic_error(payload.as_ref()))
            }),
            () = self.maintain_lease(&run) => Err(HammerworkError::Worker {
                message: "job lease heartbeat stopped unexpectedly".to_string(),
            }),
        }
    }

    /// Renew the lease on a running job until the surrounding future is dropped.
    ///
    /// Heartbeats every third of the lease duration; the first one is sent after that
    /// interval, so short jobs never pay for a heartbeat write. Never completes.
    async fn maintain_lease(&self, run: &Job) {
        let job_id = run.id;
        self.every_heartbeat(|| async move {
            match self.queue.heartbeat_job(run, self.lease_duration).await {
                Ok(true) => debug!("Renewed lease on job {}", job_id),
                Ok(false) => {
                    warn!(
                        "Lost the lease on job {}: it is no longer Running this run (it may \
                         have been reclaimed as stale); the handler keeps running",
                        job_id
                    );
                    return false;
                }
                Err(e) => warn!("Failed to renew the lease on job {}: {}", job_id, e),
            }
            true
        })
        .await
    }

    /// Update batch processing statistics
    fn update_batch_stats<F>(&self, updater: F)
    where
        F: FnOnce(&mut BatchProcessingStats),
    {
        if let Ok(mut stats) = self.batch_stats.write() {
            updater(&mut stats);
        }
    }

    /// Convert a JobEvent to a JobLifecycleEvent for external publishing
    #[cfg(feature = "webhooks")]
    fn convert_to_lifecycle_event(
        &self,
        event: &JobEvent,
        payload: Option<serde_json::Value>,
    ) -> JobLifecycleEvent {
        use std::collections::HashMap;

        let event_type = match event.event_type {
            JobEventType::Started => JobLifecycleEventType::Started,
            JobEventType::Completed => JobLifecycleEventType::Completed,
            JobEventType::Failed => JobLifecycleEventType::Failed,
            JobEventType::Retried => JobLifecycleEventType::Retried,
            JobEventType::Dead => JobLifecycleEventType::Dead,
            JobEventType::TimedOut => JobLifecycleEventType::TimedOut,
        };

        let error = event.error_message.as_ref().map(|message| JobError {
            message: message.clone(),
            error_type: match event.event_type {
                JobEventType::TimedOut => Some("timeout".to_string()),
                JobEventType::Failed => Some("processing_error".to_string()),
                JobEventType::Dead => Some("max_retries_exceeded".to_string()),
                _ => None,
            },
            details: None,
            retry_attempt: None,
        });

        let mut metadata = HashMap::new();
        metadata.insert("worker_queue".to_string(), self.queue_name.clone());

        if let Some(processing_time) = event.processing_time_ms {
            metadata.insert(
                "processing_time_ms".to_string(),
                processing_time.to_string(),
            );
        }

        JobLifecycleEvent {
            event_id: uuid::Uuid::new_v4(),
            job_id: event.job_id,
            queue_name: event.queue_name.clone(),
            event_type,
            priority: event.priority,
            timestamp: event.timestamp,
            processing_time_ms: event.processing_time_ms,
            error,
            // Only set when a subscriber asked for payloads (and never for an encrypted
            // job); the event manager and each subscriber's filter decide who sees it.
            payload,
            metadata,
        }
    }

    /// Check and update the status of a batch after job completion
    async fn check_and_update_batch_status(&self, batch_id: BatchId) -> Result<()> {
        // Get current batch status
        let batch_result = self.queue.get_batch_status(batch_id).await?;

        // If batch is complete, log success metrics
        if batch_result.pending_jobs == 0 {
            let completion_rate = batch_result.success_rate();

            if batch_counts_as_successful(&batch_result) {
                info!(
                    "Batch {} completed successfully with {:.1}% success rate",
                    batch_id, completion_rate
                );
            } else {
                warn!(
                    "Batch {} completed with {:.1}% success rate ({} failures)",
                    batch_id, completion_rate, batch_result.failed_jobs
                );
            }

            // Several jobs of the batch can finish at once (batch concurrency), and each
            // then sees the batch finished: only the first counts it.
            if !self.claim_finished_batch(batch_id) {
                return Ok(());
            }

            // Update batch completion statistics
            self.update_batch_stats(|stats| {
                stats.batches_completed += 1;
                if batch_counts_as_successful(&batch_result) {
                    stats.batches_successful += 1;
                }
            });
        }

        Ok(())
    }

    /// Record that `batch_id` finished; `false` when it was already counted.
    fn claim_finished_batch(&self, batch_id: BatchId) -> bool {
        match self.finished_batches.lock() {
            Ok(mut finished) => finished.insert(batch_id),
            Err(_) => false,
        }
    }

    /// Handle job failure within a batch context
    async fn handle_batch_job_failure(
        &self,
        batch_id: BatchId,
        job_id: uuid::Uuid,
        error_message: &str,
    ) -> Result<()> {
        // Get batch status to understand failure handling mode
        let batch_result = self.queue.get_batch_status(batch_id).await?;

        // Log batch-specific failure information
        warn!(
            "Job {} in batch {} failed: {}. Batch status: {}/{} jobs remaining",
            job_id, batch_id, error_message, batch_result.pending_jobs, batch_result.total_jobs
        );

        // PartialFailureMode (e.g. FailFast failing the remaining jobs) was already
        // applied by DatabaseQueue::finish_job_run, in the transaction that ended the job.

        Ok(())
    }

    /// Get current batch processing statistics
    pub fn get_batch_stats(&self) -> BatchProcessingStats {
        self.batch_stats
            .read()
            .map(|stats| stats.clone())
            .unwrap_or_default()
    }

    /// Record `event` of `job` (the job as stored, never a decrypted copy) with the
    /// statistics and metrics collectors and publish it to the event manager.
    #[cfg_attr(not(feature = "webhooks"), allow(unused_variables))]
    async fn record_event(&self, job: &Job, event: JobEvent) {
        // Record to statistics collector
        if let Some(stats_collector) = &self.stats_collector
            && let Err(e) = contain_panic(
                "Statistics collector",
                stats_collector.record_event(event.clone()),
            )
            .await
        {
            warn!("Failed to record statistics event: {}", e);
        }

        // Record to metrics collector
        #[cfg(feature = "metrics")]
        if let Some(metrics_collector) = &self.metrics_collector
            && let Err(e) = metrics_collector.record_job_event(&event).await
        {
            warn!("Failed to record metrics event: {}", e);
        }

        // Publish to event manager for external integrations
        #[cfg(feature = "webhooks")]
        if let Some(event_manager) = &self.event_manager {
            let payload = if event_manager.payload_wanted().await {
                event_payload(job)
            } else {
                None
            };
            let lifecycle_event = self.convert_to_lifecycle_event(&event, payload);
            if let Err(e) = event_manager.publish_event(lifecycle_event).await {
                warn!("Failed to publish lifecycle event: {}", e);
            }
        }

        // Update last job time for worker starvation detection: every finished run,
        // including one that failed and will be retried.
        if event.event_type != JobEventType::Started
            && let Ok(mut last_time) = self.last_job_time.write()
        {
            *last_time = event.timestamp;
        }
    }

    /// Start a background monitoring task for metrics and alerting
    #[cfg(any(feature = "metrics", feature = "alerting"))]
    fn start_monitoring_task(&self) -> tokio::task::JoinHandle<()> {
        let queue_name = self.queue_name.clone();

        let queue = Arc::clone(&self.queue);

        #[cfg(feature = "alerting")]
        let last_job_time = Arc::clone(&self.last_job_time);

        #[cfg(feature = "metrics")]
        let metrics_collector = self.metrics_collector.clone();

        #[cfg(feature = "alerting")]
        let alert_manager = self.alert_manager.clone();

        #[cfg(feature = "alerting")]
        let stats_collector = self.stats_collector.clone();

        // An explicit interval wins, then the metrics collector's update interval.
        #[cfg(feature = "metrics")]
        let collector_interval = metrics_collector.as_ref().map(|m| m.update_interval());
        #[cfg(not(feature = "metrics"))]
        let collector_interval: Option<Duration> = None;
        let monitor_every = self
            .monitoring_interval
            .or(collector_interval)
            .unwrap_or(DEFAULT_MONITORING_INTERVAL);

        tokio::spawn(async move {
            // `interval` panics on a zero period.
            let mut interval = tokio::time::interval(monitor_every.max(MIN_MONITORING_INTERVAL));

            loop {
                interval.tick().await;

                // Update queue depth metrics and check queue depth alerts
                #[cfg(feature = "metrics")]
                let wants_depth = metrics_collector.is_some();
                #[cfg(not(feature = "metrics"))]
                let wants_depth = false;
                #[cfg(feature = "alerting")]
                let wants_depth = wants_depth || alert_manager.is_some();
                let queue_depth = if wants_depth {
                    match queue.get_queue_depth(&queue_name).await {
                        Ok(depth) => Some(depth),
                        Err(e) => {
                            warn!("Failed to get queue depth for monitoring: {}", e);
                            None
                        }
                    }
                } else {
                    None
                };

                #[cfg(feature = "metrics")]
                if let (Some(metrics_collector), Some(queue_depth)) =
                    (&metrics_collector, queue_depth)
                    && let Err(e) = metrics_collector
                        .update_queue_depth(&queue_name, queue_depth)
                        .await
                {
                    warn!("Failed to update queue depth metrics: {}", e);
                }

                #[cfg(feature = "alerting")]
                if let (Some(alert_manager), Some(queue_depth)) = (&alert_manager, queue_depth)
                    && let Err(e) = alert_manager
                        .check_queue_depth(&queue_name, queue_depth)
                        .await
                {
                    warn!("Failed to check queue depth alerts: {}", e);
                }

                // Check worker starvation
                #[cfg(feature = "alerting")]
                if let Some(alert_manager) = &alert_manager {
                    let last_time_value = last_job_time.read().ok().map(|last_time| *last_time);
                    if let Some(last_time_value) = last_time_value
                        && let Err(e) = alert_manager
                            .check_worker_starvation(&queue_name, last_time_value)
                            .await
                    {
                        warn!("Failed to check worker starvation: {}", e);
                    }
                }

                // Check statistics-based alerts
                #[cfg(feature = "alerting")]
                if let (Some(alert_manager), Some(stats_collector)) =
                    (&alert_manager, &stats_collector)
                {
                    match stats_collector
                        .get_queue_statistics(&queue_name, Duration::from_secs(300))
                        .await
                    {
                        Ok(stats) => {
                            if let Err(e) =
                                alert_manager.check_thresholds(&queue_name, &stats).await
                            {
                                warn!("Failed to check alert thresholds: {}", e);
                            }
                        }
                        Err(e) => {
                            warn!("Failed to get queue statistics for alerting: {}", e);
                        }
                    }
                }
            }
        })
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use std::time::Duration;

    #[test]
    fn test_error_handling() {
        let error = HammerworkError::Worker {
            message: "Test error".to_string(),
        };

        assert_eq!(error.to_string(), "Worker error: Test error");
    }

    #[test]
    fn test_job_timeout_detection_logic() {
        use crate::job::{Job, JobStatus};
        use serde_json::json;
        use std::time::Duration;

        // Test job timeout detection scenarios
        let mut job = Job::new("timeout_test".to_string(), json!({"data": "test"}))
            .with_timeout(Duration::from_millis(100));

        // Job not started - should not timeout
        assert!(!job.should_timeout());

        // Job started recently - should not timeout
        job.started_at = Some(chrono::Utc::now() - chrono::Duration::milliseconds(50));
        job.status = JobStatus::Running;
        assert!(!job.should_timeout());

        // Job started long ago - should timeout
        job.started_at = Some(chrono::Utc::now() - chrono::Duration::milliseconds(200));
        assert!(job.should_timeout());

        // Job without timeout - should never timeout
        let mut job_no_timeout = Job::new("no_timeout".to_string(), json!({"data": "test"}));
        job_no_timeout.started_at = Some(chrono::Utc::now() - chrono::Duration::hours(1));
        assert!(!job_no_timeout.should_timeout());
    }

    #[tokio::test]
    async fn test_timeout_statistics_integration() {
        use crate::stats::{InMemoryStatsCollector, JobEvent, JobEventType};
        use std::sync::Arc;

        let stats_collector = Arc::new(InMemoryStatsCollector::new_default());

        // Simulate timeout event recording
        let timeout_event = JobEvent {
            job_id: uuid::Uuid::new_v4(),
            queue_name: "timeout_queue".to_string(),
            event_type: JobEventType::TimedOut,
            priority: crate::priority::JobPriority::Normal,
            processing_time_ms: Some(5000), // 5 seconds before timeout
            error_message: Some("Job timed out after 5s".to_string()),
            timestamp: chrono::Utc::now(),
        };

        stats_collector.record_event(timeout_event).await.unwrap();

        // Verify timeout event is tracked in statistics
        let stats = stats_collector
            .get_queue_statistics("timeout_queue", Duration::from_secs(60))
            .await
            .unwrap();

        assert_eq!(stats.total_processed, 1);
        assert_eq!(stats.timed_out, 1);
        assert_eq!(stats.error_rate, 1.0); // 1 timeout / 1 total = 100% error rate
    }

    #[test]
    fn test_worker_rate_limit_configuration() {
        use crate::rate_limit::RateLimit;

        // Test rate limit configuration
        let rate_limit = RateLimit::per_second(10).with_burst_limit(20);

        assert_eq!(rate_limit.rate, 10);
        assert_eq!(rate_limit.burst_limit, 20);
        assert_eq!(rate_limit.per, Duration::from_secs(1));

        // Test different time windows
        let per_minute = RateLimit::per_minute(60);
        assert_eq!(per_minute.rate, 60);
        assert_eq!(per_minute.per, Duration::from_secs(60));

        let per_hour = RateLimit::per_hour(3600);
        assert_eq!(per_hour.rate, 3600);
        assert_eq!(per_hour.per, Duration::from_secs(3600));
    }

    #[test]
    fn test_throttle_config_configuration() {
        use crate::rate_limit::ThrottleConfig;

        let throttle_config = ThrottleConfig::new()
            .max_concurrent(5)
            .rate_per_minute(100)
            .backoff_on_error(Duration::from_secs(30))
            .enabled(true);

        assert_eq!(throttle_config.max_concurrent, Some(5));
        assert_eq!(throttle_config.rate_per_minute, Some(100));
        assert_eq!(
            throttle_config.backoff_on_error,
            Some(Duration::from_secs(30))
        );
        assert!(throttle_config.enabled);

        // Test rate limit conversion
        let rate_limit = throttle_config.to_rate_limit().unwrap();
        assert_eq!(rate_limit.rate, 100);
        assert_eq!(rate_limit.per, Duration::from_secs(60));
    }

    #[tokio::test]
    async fn test_rate_limiter_integration() {
        use crate::rate_limit::{RateLimit, RateLimiter};

        let rate_limit = RateLimit::per_second(5); // 5 operations per second
        let rate_limiter = RateLimiter::new(rate_limit);

        // Should initially allow operations
        assert!(rate_limiter.try_acquire());
        assert!(rate_limiter.try_acquire());
        assert!(rate_limiter.try_acquire());
        assert!(rate_limiter.try_acquire());
        assert!(rate_limiter.try_acquire());

        // Should block after consuming all tokens
        assert!(!rate_limiter.try_acquire());

        // Test acquire method (will wait for token refill)
        let start = std::time::Instant::now();
        rate_limiter.acquire().await.unwrap();
        let elapsed = start.elapsed();

        // Should have waited some time for token refill (but not too long due to high test rate)
        assert!(elapsed < Duration::from_millis(500)); // Should be fast for this test rate
    }

    #[test]
    fn test_worker_backoff_configuration() {
        use crate::rate_limit::ThrottleConfig;

        // Test that backoff configuration is properly handled
        let throttle_config = ThrottleConfig::new().backoff_on_error(Duration::from_secs(60));

        assert_eq!(
            throttle_config.backoff_on_error,
            Some(Duration::from_secs(60))
        );

        // Test default poll interval fallback
        let poll_interval = Duration::from_secs(1);
        let backoff_duration = throttle_config.backoff_on_error.unwrap_or(poll_interval);
        assert_eq!(backoff_duration, Duration::from_secs(60));

        // Test with no backoff configured
        let no_backoff_config = ThrottleConfig::new();
        let backoff_duration = no_backoff_config.backoff_on_error.unwrap_or(poll_interval);
        assert_eq!(backoff_duration, poll_interval);
    }

    #[tokio::test]
    async fn test_rate_limiter_token_availability() {
        use crate::rate_limit::{RateLimit, RateLimiter};

        let rate_limit = RateLimit::per_second(10); // 10 tokens per second
        let rate_limiter = RateLimiter::new(rate_limit);

        // Check initial token availability
        let initial_tokens = rate_limiter.available_tokens();
        assert_eq!(initial_tokens, 10.0); // Should start with full burst capacity

        // Consume some tokens
        assert!(rate_limiter.try_acquire());
        assert!(rate_limiter.try_acquire());

        // Check remaining tokens (refill is continuous, so a little has come back)
        let remaining_tokens = rate_limiter.available_tokens();
        assert!((8.0..8.5).contains(&remaining_tokens), "{remaining_tokens}");
    }

    #[test]
    fn test_rate_limit_edge_cases() {
        use crate::rate_limit::RateLimit;

        // Test very low rate
        let low_rate = RateLimit::per_hour(1);
        assert_eq!(low_rate.rate, 1);
        assert_eq!(low_rate.per, Duration::from_secs(3600));

        // Test very high rate
        let high_rate = RateLimit::per_second(1000);
        assert_eq!(high_rate.rate, 1000);
        assert_eq!(high_rate.burst_limit, 1000);

        // Test custom burst limit
        let custom_burst = RateLimit::per_second(10).with_burst_limit(50);
        assert_eq!(custom_burst.burst_limit, 50);
    }

    #[test]
    fn test_throttle_config_defaults() {
        use crate::rate_limit::ThrottleConfig;

        let default_config = ThrottleConfig::default();
        assert!(default_config.enabled);
        assert!(default_config.max_concurrent.is_none());
        assert!(default_config.rate_per_minute.is_none());
        assert!(default_config.backoff_on_error.is_none());

        let new_config = ThrottleConfig::new();
        assert_eq!(new_config.enabled, default_config.enabled);
        assert_eq!(new_config.max_concurrent, default_config.max_concurrent);
    }

    #[test]
    fn test_retry_at_from_delay_never_panics_and_clamps() {
        let now = Utc::now();
        assert_eq!(
            retry_at_from_delay(now, Duration::from_secs(30)),
            now + chrono::Duration::seconds(30)
        );
        let max = now + chrono::Duration::from_std(MAX_RETRY_DELAY).unwrap();
        assert_eq!(retry_at_from_delay(now, Duration::MAX), max);
        assert_eq!(retry_at_from_delay(now, MAX_RETRY_DELAY * 2), max);
        assert_eq!(
            retry_at_from_delay(DateTime::<Utc>::MAX_UTC, Duration::from_secs(1)),
            DateTime::<Utc>::MAX_UTC
        );
    }

    #[test]
    fn test_error_backoff_grows_is_capped_and_never_zero() {
        assert_eq!(error_backoff(Duration::ZERO, 1), MIN_ERROR_BACKOFF);
        assert_eq!(
            error_backoff(Duration::from_secs(1), 1),
            Duration::from_secs(1)
        );
        assert_eq!(
            error_backoff(Duration::from_secs(1), 3),
            Duration::from_secs(4)
        );
        assert_eq!(
            error_backoff(Duration::from_secs(1), u32::MAX),
            MAX_ERROR_BACKOFF
        );
        // A base above the cap is honoured rather than shortened.
        assert_eq!(
            error_backoff(Duration::from_secs(120), 5),
            Duration::from_secs(120)
        );
    }

    #[tokio::test]
    async fn test_catch_unwind_turns_panics_into_errors() {
        let ok: HandlerFuture = Box::pin(async { Ok(JobResult::success()) });
        assert!(CatchUnwind { inner: ok }.await.unwrap().is_ok());

        let panicking: HandlerFuture = Box::pin(async {
            tokio::task::yield_now().await;
            panic!("boom")
        });
        let payload = CatchUnwind { inner: panicking }.await.unwrap_err();
        let error = handler_panic_error(payload.as_ref());
        assert_eq!(
            error.to_string(),
            "Worker error: Job handler panicked: boom"
        );

        let formatted: HandlerFuture = Box::pin(async { panic!("code {}", 42) });
        let payload = CatchUnwind { inner: formatted }.await.unwrap_err();
        assert_eq!(panic_message(payload.as_ref()), "code 42");
    }

    #[test]
    fn panic_message_describes_every_payload_kind() {
        assert_eq!(panic_message(&"static str"), "static str");
        assert_eq!(panic_message(&"owned".to_string()), "owned");
        assert_eq!(panic_message(&42_u32), "non-string panic payload");
        assert_eq!(
            handler_panic_error(&"boom").to_string(),
            "Worker error: Job handler panicked: boom"
        );
    }

    #[test]
    fn shutdown_requested_on_signal_or_dropped_sender() {
        let (tx, mut rx) = mpsc::channel(1);
        assert!(!shutdown_requested(&mut rx));
        tx.try_send(()).unwrap();
        assert!(shutdown_requested(&mut rx));
        drop(tx);
        assert!(
            shutdown_requested(&mut rx),
            "a dropped sender means shutdown"
        );
    }

    /// A worker on a pool that never connects (no database is needed to inspect how a
    /// worker is configured).
    #[cfg(feature = "postgres")]
    pub(super) fn lazy_worker(queue_name: &str) -> Worker<sqlx::Postgres> {
        let pool = sqlx::PgPool::connect_lazy("postgres://localhost/hammerwork_unused").unwrap();
        let handler: JobHandler = Arc::new(|_job| Box::pin(async { Ok(()) }));
        Worker::new(
            Arc::new(JobQueue::new(pool)),
            queue_name.to_string(),
            handler,
        )
    }

    #[cfg(feature = "postgres")]
    #[tokio::test]
    async fn with_config_applies_the_worker_section() {
        let config = crate::config::WorkerConfig {
            polling_interval: Duration::from_millis(250),
            job_timeout: Duration::from_secs(7),
            priority_weights: PriorityWeights::strict(),
            retry_strategy: RetryStrategy::fixed(Duration::from_secs(42)),
            ..Default::default()
        };
        let worker = lazy_worker("cfg").with_config(&config);
        assert_eq!(worker.poll_interval, Duration::from_millis(250));
        assert_eq!(worker.default_timeout, Some(Duration::from_secs(7)));
        assert!(worker.priority_weights.as_ref().unwrap().is_strict());
        assert_eq!(
            worker.default_retry_strategy,
            Some(RetryStrategy::fixed(Duration::from_secs(42)))
        );

        let worker = lazy_worker("cfg").with_weighted_priority();
        assert!(!worker.priority_weights.as_ref().unwrap().is_strict());
        let worker = worker.with_strict_priority();
        assert!(worker.priority_weights.as_ref().unwrap().is_strict());
    }

    #[cfg(feature = "postgres")]
    #[tokio::test]
    async fn with_hammerwork_config_applies_the_queue_throttle() {
        let mut config = crate::HammerworkConfig::new();
        config.worker.polling_interval = Duration::from_millis(100);
        config.rate_limiting.enabled = true;
        config.rate_limiting.default_throttle = ThrottleConfig::new().rate_per_minute(600);
        config.rate_limiting.queue_throttles.insert(
            "emails".to_string(),
            ThrottleConfig::new()
                .max_concurrent(2)
                .rate_per_minute(60)
                .backoff_on_error(Duration::from_secs(5)),
        );

        let emails = lazy_worker("emails").with_hammerwork_config(&config);
        assert_eq!(emails.poll_interval, Duration::from_millis(100));
        assert_eq!(
            emails.throttle_config.as_ref().unwrap().max_concurrent,
            Some(2)
        );
        assert_eq!(
            emails
                .concurrency_limit
                .as_ref()
                .unwrap()
                .available_permits(),
            2
        );
        assert!(emails.rate_limiter.is_some());
        assert_eq!(emails.error_backoff_base(), Duration::from_secs(5));

        // Other queues get the default throttle: a rate limit only.
        let other = lazy_worker("other").with_hammerwork_config(&config);
        assert_eq!(
            other.throttle_config.as_ref().unwrap().rate_per_minute,
            Some(600)
        );
        assert!(other.concurrency_limit.is_none());
        assert_eq!(
            other.error_backoff_base(),
            Duration::from_millis(100),
            "without backoff_on_error the poll interval is the backoff base"
        );

        // With rate limiting disabled no throttle applies.
        config.rate_limiting.enabled = false;
        let unthrottled = lazy_worker("emails").with_hammerwork_config(&config);
        assert!(unthrottled.throttle_config.is_none());
        assert!(unthrottled.rate_limiter.is_none());
    }

    #[cfg(all(feature = "postgres", feature = "alerting"))]
    #[tokio::test]
    async fn with_hammerwork_config_applies_alerting() {
        let mut config = crate::HammerworkConfig::new();
        config.alerting = AlertingConfig::new().alert_on_queue_depth(17);
        let worker = lazy_worker("alerts").with_hammerwork_config(&config);
        let manager = worker.alert_manager.as_ref().expect("an alert manager");
        assert_eq!(manager.config().queue_depth_threshold, Some(17));
    }

    #[cfg(feature = "postgres")]
    #[tokio::test]
    async fn disabled_throttle_is_ignored_and_max_concurrent_is_at_least_one() {
        let mut disabled = ThrottleConfig::new()
            .max_concurrent(1)
            .rate_per_minute(1)
            .backoff_on_error(Duration::from_secs(9));
        disabled.enabled = false;
        let worker = lazy_worker("t").with_throttle_config(disabled);
        assert!(worker.throttle_config.is_none());
        assert!(worker.rate_limiter.is_none());
        assert!(worker.concurrency_limit.is_none());
        assert_eq!(worker.error_backoff_base(), worker.poll_interval);

        let worker = lazy_worker("t").with_throttle_config(ThrottleConfig::new().max_concurrent(0));
        assert_eq!(
            worker
                .concurrency_limit
                .as_ref()
                .unwrap()
                .available_permits(),
            1,
            "max_concurrent = 0 would block the worker forever"
        );

        // Clones share the limit, so it applies across a pool.
        let clone = worker.clone();
        assert!(Arc::ptr_eq(
            worker.concurrency_limit.as_ref().unwrap(),
            clone.concurrency_limit.as_ref().unwrap()
        ));
    }

    /// #105: a finished batch is counted once, and the memory of finished batches is
    /// bounded.
    #[test]
    fn finished_batches_are_counted_once() {
        let mut finished = FinishedBatches::default();
        let first = uuid::Uuid::new_v4();
        assert!(finished.insert(first));
        assert!(!finished.insert(first));
        for _ in 0..FINISHED_BATCHES_REMEMBERED {
            assert!(finished.insert(uuid::Uuid::new_v4()));
        }
        assert_eq!(finished.ids.len(), FINISHED_BATCHES_REMEMBERED);
        assert_eq!(finished.order.len(), FINISHED_BATCHES_REMEMBERED);
        assert!(finished.insert(first), "the oldest batch was forgotten");
    }

    /// #101: a plain job's events carry its payload; an encrypted job's never do.
    #[cfg(feature = "webhooks")]
    #[test]
    fn event_payload_is_withheld_for_encrypted_jobs() {
        let job = Job::new("events".to_string(), serde_json::json!({"a": 1}));
        assert_eq!(event_payload(&job), Some(serde_json::json!({"a": 1})));

        let mut sealed = job.clone();
        sealed.is_encrypted = true;
        assert_eq!(event_payload(&sealed), None);

        #[cfg(feature = "encryption")]
        {
            let configured = job.with_encryption(crate::encryption::EncryptionConfig::new(
                crate::encryption::EncryptionAlgorithm::AES256GCM,
            ));
            assert_eq!(event_payload(&configured), None);
        }
    }

    #[cfg(all(feature = "postgres", feature = "webhooks"))]
    #[tokio::test]
    async fn lifecycle_events_carry_the_job_event_details() {
        let worker = lazy_worker("events");
        let job_id = uuid::Uuid::new_v4();
        let event = |event_type, error: Option<&str>| JobEvent {
            job_id,
            queue_name: "events".to_string(),
            event_type,
            priority: crate::priority::JobPriority::High,
            processing_time_ms: Some(12),
            error_message: error.map(str::to_string),
            timestamp: Utc::now(),
        };

        let completed =
            worker.convert_to_lifecycle_event(&event(JobEventType::Completed, None), None);
        assert_eq!(completed.event_type, JobLifecycleEventType::Completed);
        assert_eq!(completed.job_id, job_id);
        assert_eq!(completed.priority, crate::priority::JobPriority::High);
        assert!(completed.error.is_none());
        assert_eq!(completed.metadata["worker_queue"], "events");
        assert_eq!(completed.metadata["processing_time_ms"], "12");

        for (event_type, lifecycle, error_type) in [
            (
                JobEventType::TimedOut,
                JobLifecycleEventType::TimedOut,
                Some("timeout"),
            ),
            (
                JobEventType::Failed,
                JobLifecycleEventType::Failed,
                Some("processing_error"),
            ),
            (
                JobEventType::Dead,
                JobLifecycleEventType::Dead,
                Some("max_retries_exceeded"),
            ),
            (JobEventType::Retried, JobLifecycleEventType::Retried, None),
            (JobEventType::Started, JobLifecycleEventType::Started, None),
        ] {
            let converted =
                worker.convert_to_lifecycle_event(&event(event_type, Some("bad")), None);
            assert_eq!(converted.event_type, lifecycle);
            let error = converted.error.expect("the error message is kept");
            assert_eq!(error.message, "bad");
            assert_eq!(error.error_type.as_deref(), error_type);
        }
    }

    /// #64 M10: the monitoring task stops however the worker's `run` ends, including
    /// when its future is dropped or aborted (as a pool does when a worker panics).
    #[cfg(all(feature = "postgres", feature = "metrics"))]
    #[tokio::test]
    async fn monitoring_task_stops_when_run_is_dropped() {
        let metrics = Arc::new(
            PrometheusMetricsCollector::new(crate::metrics::MetricsConfig::default()).unwrap(),
        );
        let worker = lazy_worker("monitor")
            .with_metrics_collector(Arc::clone(&metrics))
            .with_poll_interval(Duration::from_millis(10));
        // This test and the worker.
        assert_eq!(Arc::strong_count(&metrics), 2);
        let (_shutdown_tx, shutdown_rx) = mpsc::channel(1);
        let run = tokio::spawn(async move { worker.run(shutdown_rx).await });

        // The monitoring task holds its own handle on the collector once it runs.
        let deadline = std::time::Instant::now() + Duration::from_secs(5);
        while Arc::strong_count(&metrics) < 3 && std::time::Instant::now() < deadline {
            tokio::time::sleep(Duration::from_millis(5)).await;
        }
        assert_eq!(Arc::strong_count(&metrics), 3);

        run.abort();
        let _ = run.await;
        let deadline = std::time::Instant::now() + Duration::from_secs(5);
        while Arc::strong_count(&metrics) > 1 && std::time::Instant::now() < deadline {
            tokio::time::sleep(Duration::from_millis(5)).await;
        }
        assert_eq!(
            Arc::strong_count(&metrics),
            1,
            "the monitoring task outlived the worker"
        );
    }

    /// #64 M5: a zero poll interval waits at least MIN_POLL_INTERVAL between polls.
    #[cfg(feature = "postgres")]
    #[tokio::test]
    async fn zero_poll_interval_does_not_hot_loop() {
        let worker = lazy_worker("idle").with_poll_interval(Duration::ZERO);
        let (_shutdown_tx, mut shutdown_rx) = mpsc::channel(1);
        let started = std::time::Instant::now();
        for _ in 0..5 {
            assert!(matches!(
                worker.idle_wait::<Box<Job>>(&mut shutdown_rx).await,
                Acquired::Idle
            ));
        }
        assert!(started.elapsed() >= MIN_POLL_INTERVAL * 5);
    }

    #[cfg(feature = "postgres")]
    #[tokio::test]
    async fn batch_builders_set_size_concurrency_and_mode() {
        let worker = lazy_worker("batch");
        assert!(!worker.is_batch_processing_enabled());
        assert_eq!(worker.batch_size(), DEFAULT_BATCH_SIZE);
        assert_eq!(worker.batch_concurrency(), DEFAULT_BATCH_CONCURRENCY);

        // The old builder alone enables batch mode with the default size of one job.
        let enabled = lazy_worker("batch").with_batch_processing_enabled(true);
        assert!(enabled.is_batch_processing_enabled());
        assert_eq!(enabled.batch_size(), DEFAULT_BATCH_SIZE);

        // `with_batch_size` enables batch mode; zero values are raised to one.
        let sized = lazy_worker("batch")
            .with_batch_size(0)
            .with_batch_concurrency(0);
        assert!(sized.is_batch_processing_enabled());
        assert_eq!(sized.batch_size(), 1);
        assert_eq!(sized.batch_concurrency(), 1);

        let worker = lazy_worker("batch")
            .with_batch_size(25)
            .with_batch_concurrency(5);
        let clone = worker.clone();
        assert_eq!(clone.batch_size(), 25);
        assert_eq!(clone.batch_concurrency(), 5);
        assert!(clone.is_batch_processing_enabled());
        // Batch mode can still be switched off afterwards.
        assert!(
            !worker
                .with_batch_processing_enabled(false)
                .is_batch_processing_enabled()
        );
    }

    #[tokio::test]
    async fn next_finished_yields_runs_as_they_finish() {
        let mut runs: Vec<BatchRun<'_>> = Vec::new();
        assert!(next_finished(&mut runs).await.is_none());

        let ids: Vec<crate::job::JobId> = (0..3).map(|_| uuid::Uuid::new_v4()).collect();
        for (index, id) in ids.iter().copied().enumerate() {
            // The first-pushed run finishes last.
            let delay = Duration::from_millis(60 - 20 * index as u64);
            runs.push(Box::pin(async move {
                sleep(delay).await;
                (id, Ok(()))
            }));
        }
        let mut order = Vec::new();
        while let Some((id, outcome)) = next_finished(&mut runs).await {
            assert!(outcome.is_ok());
            order.push(id);
        }
        assert_eq!(order, [ids[2], ids[1], ids[0]]);
        assert!(runs.is_empty());
    }

    #[cfg(feature = "postgres")]
    #[test]
    fn held_job_with_a_lost_lease_is_not_started() {
        let kept = Job::new("held".to_string(), serde_json::json!({}));
        let lost = Job::new("held".to_string(), serde_json::json!({}));
        let now = std::time::Instant::now();
        let held = std::sync::Mutex::new(HeldJobs {
            waiting: vec![kept.clone(), lost.clone()],
            lost: std::collections::HashSet::from([lost.id]),
            renewed: std::collections::HashMap::from([(kept.id, now), (lost.id, now)]),
        });
        let interval = Duration::from_secs(60);
        assert_eq!(
            Worker::<sqlx::Postgres>::start_held(&held, kept.id, interval),
            HeldStart::Run
        );
        assert_eq!(
            Worker::<sqlx::Postgres>::start_held(&held, lost.id, interval),
            HeldStart::Lost
        );
        assert!(held.lock().unwrap().waiting.is_empty());
    }

    #[cfg(feature = "postgres")]
    #[test]
    fn held_job_renewed_too_long_ago_is_renewed_before_it_starts() {
        // Its running heartbeat only fires an interval after it starts; together with
        // the time since its last renewal that could outlast the lease.
        let job = Job::new("held".to_string(), serde_json::json!({}));
        let held = std::sync::Mutex::new(HeldJobs {
            waiting: vec![job.clone()],
            lost: Default::default(),
            renewed: std::collections::HashMap::from([(job.id, std::time::Instant::now())]),
        });
        std::thread::sleep(Duration::from_millis(20));
        assert_eq!(
            Worker::<sqlx::Postgres>::start_held(&held, job.id, Duration::from_millis(10)),
            HeldStart::RenewThenRun
        );
    }

    fn batch_result(total: u32, completed: u32) -> crate::batch::BatchResult {
        crate::batch::BatchResult {
            batch_id: uuid::Uuid::new_v4(),
            total_jobs: total,
            completed_jobs: completed,
            failed_jobs: total - completed,
            pending_jobs: 0,
            status: crate::batch::BatchStatus::Completed,
            created_at: Utc::now(),
            completed_at: Some(Utc::now()),
            error_summary: None,
            job_errors: Default::default(),
        }
    }

    /// `BatchResult::success_rate` is a percentage (0-100), so a batch where half the
    /// jobs failed must not count as successful (the threshold used to be 0.95, i.e.
    /// 0.95%).
    #[test]
    fn batch_success_threshold_is_a_percentage() {
        assert!(!batch_counts_as_successful(&batch_result(100, 50)));
        assert!(!batch_counts_as_successful(&batch_result(100, 1)));
        assert!(!batch_counts_as_successful(&batch_result(100, 94)));
        assert!(batch_counts_as_successful(&batch_result(100, 95)));
        assert!(batch_counts_as_successful(&batch_result(10, 10)));
    }

    /// Claims count the attempt, so after a job's first failed run `attempts` is 1 and
    /// the delay must be the strategy's first retry delay (`calculate_delay(1)`).
    #[cfg(feature = "postgres")]
    #[tokio::test]
    async fn first_retry_uses_the_strategys_first_delay() {
        let worker = lazy_worker("retry_numbering").with_default_retry_strategy(
            RetryStrategy::exponential(Duration::from_secs(60), 10.0, None),
        );
        let mut job = Job::new("retry_numbering".to_string(), serde_json::json!({}));
        job.attempts = 1; // as returned by the claim of its first run
        let delay = worker.retry_at_for(&job) - Utc::now();
        assert!(
            delay <= chrono::Duration::seconds(61) && delay >= chrono::Duration::seconds(59),
            "first retry waited {delay}, expected the 60s base delay"
        );

        job.attempts = 2;
        let delay = worker.retry_at_for(&job) - Utc::now();
        assert!(
            delay <= chrono::Duration::seconds(601) && delay >= chrono::Duration::seconds(599),
            "second retry waited {delay}, expected 600s"
        );

        // A job without attempts recorded still gets the first delay.
        job.attempts = 0;
        let delay = worker.retry_at_for(&job) - Utc::now();
        assert!(delay <= chrono::Duration::seconds(61), "waited {delay}");
    }

    /// A worker whose jobs fail and are retried is processing jobs, not starving: a
    /// `Retried` event moves `last_job_time` like any other finished run.
    #[cfg(feature = "postgres")]
    #[tokio::test]
    async fn retried_runs_update_last_job_time() {
        let worker = lazy_worker("starvation");
        let long_ago = Utc::now() - chrono::Duration::hours(1);
        *worker.last_job_time.write().unwrap() = long_ago;
        let job = Job::new("starvation".to_string(), serde_json::json!({}));
        worker
            .record_event(
                &job,
                job_event(
                    &job,
                    JobEventType::Retried,
                    None,
                    Some("boom".to_string()),
                    Utc::now(),
                ),
            )
            .await;
        assert!(*worker.last_job_time.read().unwrap() > long_ago);

        // A start is not a finished run.
        *worker.last_job_time.write().unwrap() = long_ago;
        worker
            .record_event(
                &job,
                job_event(&job, JobEventType::Started, None, None, Utc::now()),
            )
            .await;
        assert_eq!(*worker.last_job_time.read().unwrap(), long_ago);
    }
}