cloacina 0.11.1

/*
 *  Copyright 2026 Colliery Software
 *
 *  Licensed under the Apache License, Version 2.0 (the "License");
 *  you may not use this file except in compliance with the License.
 *  You may obtain a copy of the License at
 *
 *      http://www.apache.org/licenses/LICENSE-2.0
 *
 *  Unless required by applicable law or agreed to in writing, software
 *  distributed under the License is distributed on an "AS IS" BASIS,
 *  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 *  See the License for the specific language governing permissions and
 *  limitations under the License.
 */

//! Stale claim sweeper — background service for expired claim recovery.
//!
//! Periodically scans for tasks with stale heartbeats (crashed runners),
//! releases their claims, and resets them to Ready for re-execution.
//!
//! ## Startup Grace Period
//!
//! In a distributed system, the scheduler may restart while runners are
//! still executing tasks. The sweeper records when it became ready and
//! ignores all claims during its first `stale_threshold` duration after
//! startup. This prevents false positives where tasks look stale simply
//! because the sweeper wasn't running to observe their heartbeats.

use std::sync::Arc;
use std::time::{Duration, Instant};

use tokio::sync::watch;
use tracing::{debug, info, warn};

use crate::dal::DAL;

/// Configuration for the stale claim sweeper.
#[derive(Debug, Clone)]
pub struct StaleClaimSweeperConfig {
    /// How often to run the sweep (default 30s).
    pub sweep_interval: Duration,
    /// How old a heartbeat must be to consider the claim stale (default 60s).
    /// Must be greater than the heartbeat interval.
    pub stale_threshold: Duration,
    /// How many stale-claim recoveries a single task gets before it is
    /// abandoned (marked `Failed` with an `ABANDONED:` error) instead of
    /// re-Readied (default 3, mirroring `cron_max_recovery_attempts`).
    /// Without a cap, a task that reliably kills its runner recycles
    /// forever (CLOACI-T-0914).
    pub max_recovery_attempts: i32,
}

impl Default for StaleClaimSweeperConfig {
    fn default() -> Self {
        Self {
            sweep_interval: Duration::from_secs(30),
            stale_threshold: Duration::from_secs(60),
            max_recovery_attempts: 3,
        }
    }
}

/// Background service that sweeps for stale task claims.
pub struct StaleClaimSweeper {
    dal: Arc<DAL>,
    config: StaleClaimSweeperConfig,
    shutdown_rx: watch::Receiver<bool>,
    /// When the sweeper became ready. Used for the startup grace period.
    ready_at: Instant,
}

impl StaleClaimSweeper {
    /// Create a new stale claim sweeper.
    pub fn new(
        dal: Arc<DAL>,
        config: StaleClaimSweeperConfig,
        shutdown_rx: watch::Receiver<bool>,
    ) -> Self {
        Self {
            dal,
            config,
            shutdown_rx,
            ready_at: Instant::now(),
        }
    }

    /// Run the sweep loop.
    pub async fn run(&mut self) {
        info!(
            "Starting stale claim sweeper (interval: {}s, threshold: {}s, grace period: {}s)",
            self.config.sweep_interval.as_secs(),
            self.config.stale_threshold.as_secs(),
            self.config.stale_threshold.as_secs(),
        );

        let mut interval = tokio::time::interval(self.config.sweep_interval);
        interval.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Skip);

        loop {
            tokio::select! {
                _ = interval.tick() => {
                    self.sweep().await;
                }
                _ = self.shutdown_rx.changed() => {
                    if *self.shutdown_rx.borrow() {
                        info!("Stale claim sweeper shutting down");
                        break;
                    }
                }
            }
        }
    }

    /// Perform a single sweep pass.
    pub async fn sweep(&self) {
        // Startup grace period: don't sweep until we've been running for
        // at least one full stale_threshold duration. This prevents false
        // positives when the scheduler restarts — tasks that were being
        // executed by healthy runners look "stale" simply because the
        // sweeper wasn't running to see their heartbeats.
        let uptime = self.ready_at.elapsed();
        if uptime < self.config.stale_threshold {
            debug!(
                "Stale claim sweeper in grace period ({:.1}s / {}s) — skipping sweep",
                uptime.as_secs_f64(),
                self.config.stale_threshold.as_secs()
            );
            return;
        }

        // Find tasks with stale heartbeats
        let stale_claims = match self
            .dal
            .task_execution()
            .find_stale_claims(self.config.stale_threshold)
            .await
        {
            Ok(claims) => claims,
            Err(e) => {
                warn!("Stale claim sweep failed: {}", e);
                return;
            }
        };

        if stale_claims.is_empty() {
            debug!("Stale claim sweep: no stale claims found");
            return;
        }

        info!(
            "Stale claim sweep found {} stale claims",
            stale_claims.len()
        );

        for claim in &stale_claims {
            let age = chrono::Utc::now() - claim.heartbeat_at;

            // Release the claim
            if let Err(e) = self
                .dal
                .task_execution()
                .release_runner_claim(claim.task_id)
                .await
            {
                warn!(
                    "Failed to release stale claim on task {}: {}",
                    claim.task_id, e
                );
                continue;
            }

            // Recovery cap (CLOACI-T-0914): a task that has already burned
            // its recovery budget is abandoned — `mark_abandoned` writes the
            // terminal `Failed` status with an `ABANDONED:` error prefix and
            // a TaskAbandoned audit event in one transaction — instead of
            // being re-Readied forever. Failures here warn-and-continue: the
            // next sweep retries naturally.
            if claim.recovery_attempts >= self.config.max_recovery_attempts {
                let reason = format!(
                    "{} stale-claim recoveries exhausted (cap {}); last runner {}, heartbeat {}s stale",
                    claim.recovery_attempts,
                    self.config.max_recovery_attempts,
                    claim.claimed_by,
                    age.num_seconds()
                );
                if let Err(e) = self
                    .dal
                    .task_execution()
                    .mark_abandoned(claim.task_id, &reason)
                    .await
                {
                    warn!("Failed to abandon task {}: {}", claim.task_id, e);
                    continue;
                }
                metrics::counter!("cloacina_scheduler_tasks_abandoned_total").increment(1);
                warn!("Abandoned task {}: {}", claim.task_id, reason);
                continue;
            }

            // Reset to Ready for re-execution. `reset_task_for_recovery`
            // (not `mark_ready`) so the attempt is counted against the cap:
            // it re-Readies, clears started_at, increments recovery_attempts,
            // and stamps last_recovery_at.
            if let Err(e) = self
                .dal
                .task_execution()
                .reset_task_for_recovery(claim.task_id)
                .await
            {
                warn!(
                    "Failed to reset task {} to Ready after stale claim release: {}",
                    claim.task_id, e
                );
                continue;
            }

            metrics::counter!("cloacina_scheduler_stale_claims_swept_total").increment(1);

            info!(
                "Released stale claim: task {} (runner {}, last heartbeat {}s ago, recovery {}/{})",
                claim.task_id,
                claim.claimed_by,
                age.num_seconds(),
                claim.recovery_attempts + 1,
                self.config.max_recovery_attempts
            );
        }

        info!(
            "Stale claim sweep complete: {} claims released",
            stale_claims.len()
        );
    }
}

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

    #[test]
    fn config_defaults() {
        let config = StaleClaimSweeperConfig::default();
        assert_eq!(config.sweep_interval, Duration::from_secs(30));
        assert_eq!(config.stale_threshold, Duration::from_secs(60));
        assert_eq!(config.max_recovery_attempts, 3);
    }

    #[test]
    fn config_custom_values() {
        let config = StaleClaimSweeperConfig {
            sweep_interval: Duration::from_secs(10),
            stale_threshold: Duration::from_secs(120),
            max_recovery_attempts: 5,
        };
        assert_eq!(config.sweep_interval, Duration::from_secs(10));
        assert_eq!(config.stale_threshold, Duration::from_secs(120));
        assert_eq!(config.max_recovery_attempts, 5);
    }

    #[test]
    fn config_clone() {
        let config = StaleClaimSweeperConfig::default();
        let cloned = config.clone();
        assert_eq!(config.sweep_interval, cloned.sweep_interval);
        assert_eq!(config.stale_threshold, cloned.stale_threshold);
    }
}