diff options
Diffstat (limited to '')
| -rw-r--r-- | src/database/mod.rs | 19 | ||||
| -rw-r--r-- | src/tasks/mod.rs | 41 |
2 files changed, 46 insertions, 14 deletions
diff --git a/src/database/mod.rs b/src/database/mod.rs index 53f4bf4..e803da9 100644 --- a/src/database/mod.rs +++ b/src/database/mod.rs | |||
| @@ -50,12 +50,19 @@ async fn run_migrations(pool: &DbPool) -> Result<()> { | |||
| 50 | active INTEGER NOT NULL DEFAULT 1, | 50 | active INTEGER NOT NULL DEFAULT 1, |
| 51 | audio_filename TEXT, | 51 | audio_filename TEXT, |
| 52 | message TEXT, | 52 | message TEXT, |
| 53 | created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP | 53 | created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP, |
| 54 | last_run TIMESTAMP | ||
| 54 | )", | 55 | )", |
| 55 | ) | 56 | ) |
| 56 | .execute(pool) | 57 | .execute(pool) |
| 57 | .await?; | 58 | .await?; |
| 58 | 59 | ||
| 60 | // Add last_run column for pre-existing databases that lack it. | ||
| 61 | sqlx::query("ALTER TABLE cron_jobs ADD COLUMN last_run TIMESTAMP") | ||
| 62 | .execute(pool) | ||
| 63 | .await | ||
| 64 | .ok(); | ||
| 65 | |||
| 59 | sqlx::query( | 66 | sqlx::query( |
| 60 | "CREATE TABLE IF NOT EXISTS triggers ( | 67 | "CREATE TABLE IF NOT EXISTS triggers ( |
| 61 | id INTEGER PRIMARY KEY AUTOINCREMENT, | 68 | id INTEGER PRIMARY KEY AUTOINCREMENT, |
| @@ -157,6 +164,7 @@ pub struct CronJob { | |||
| 157 | pub message: Option<String>, | 164 | pub message: Option<String>, |
| 158 | #[allow(dead_code)] | 165 | #[allow(dead_code)] |
| 159 | pub created_at: Option<DateTime<Utc>>, | 166 | pub created_at: Option<DateTime<Utc>>, |
| 167 | pub last_run: Option<DateTime<Utc>>, | ||
| 160 | } | 168 | } |
| 161 | 169 | ||
| 162 | #[allow(dead_code)] | 170 | #[allow(dead_code)] |
| @@ -202,6 +210,15 @@ pub async fn delete_cron_job(pool: &DbPool, id: i64) -> Result<bool> { | |||
| 202 | Ok(result.rows_affected() > 0) | 210 | Ok(result.rows_affected() > 0) |
| 203 | } | 211 | } |
| 204 | 212 | ||
| 213 | pub async fn update_cron_last_run(pool: &DbPool, id: i64, last_run: DateTime<Utc>) -> Result<()> { | ||
| 214 | sqlx::query("UPDATE cron_jobs SET last_run = ? WHERE id = ?") | ||
| 215 | .bind(last_run) | ||
| 216 | .bind(id) | ||
| 217 | .execute(pool) | ||
| 218 | .await?; | ||
| 219 | Ok(()) | ||
| 220 | } | ||
| 221 | |||
| 205 | pub async fn set_cron_job_active(pool: &DbPool, id: i64, active: bool) -> Result<bool> { | 222 | pub async fn set_cron_job_active(pool: &DbPool, id: i64, active: bool) -> Result<bool> { |
| 206 | let result = sqlx::query("UPDATE cron_jobs SET active = ? WHERE id = ?") | 223 | let result = sqlx::query("UPDATE cron_jobs SET active = ? WHERE id = ?") |
| 207 | .bind(if active { 1 } else { 0 }) | 224 | .bind(if active { 1 } else { 0 }) |
diff --git a/src/tasks/mod.rs b/src/tasks/mod.rs index f3f8811..1519c12 100644 --- a/src/tasks/mod.rs +++ b/src/tasks/mod.rs | |||
| @@ -1,6 +1,6 @@ | |||
| 1 | use std::sync::Arc; | 1 | use std::sync::Arc; |
| 2 | 2 | ||
| 3 | use chrono::{Duration, Timelike, Utc}; | 3 | use chrono::{DateTime, Duration, Timelike, Utc}; |
| 4 | use cron::Schedule; | 4 | use cron::Schedule; |
| 5 | use poise::serenity_prelude as serenity; | 5 | use poise::serenity_prelude as serenity; |
| 6 | use std::str::FromStr; | 6 | use std::str::FromStr; |
| @@ -33,9 +33,6 @@ fn parse_cron(raw: &str) -> Result<Schedule, cron::error::Error> { | |||
| 33 | Schedule::from_str(&normalized) | 33 | Schedule::from_str(&normalized) |
| 34 | } | 34 | } |
| 35 | 35 | ||
| 36 | /// Check window: look for cron fires in the last WINDOW_SECONDS. | ||
| 37 | /// 61s gives 1s overlap with 60s tick to avoid missing exact-boundary fires. | ||
| 38 | const WINDOW_SECONDS: i64 = 61; | ||
| 39 | /// Tick interval for the scheduler loop. | 36 | /// Tick interval for the scheduler loop. |
| 40 | const TICK_SECONDS: u64 = 60; | 37 | const TICK_SECONDS: u64 = 60; |
| 41 | 38 | ||
| @@ -49,7 +46,6 @@ pub async fn check_and_execute_jobs(ctx: &CronContext) { | |||
| 49 | }; | 46 | }; |
| 50 | 47 | ||
| 51 | let now = Utc::now(); | 48 | let now = Utc::now(); |
| 52 | let window_start = now - Duration::seconds(WINDOW_SECONDS); | ||
| 53 | 49 | ||
| 54 | for job in jobs { | 50 | for job in jobs { |
| 55 | let schedule = match parse_cron(&job.cron_expression) { | 51 | let schedule = match parse_cron(&job.cron_expression) { |
| @@ -60,26 +56,45 @@ pub async fn check_and_execute_jobs(ctx: &CronContext) { | |||
| 60 | } | 56 | } |
| 61 | }; | 57 | }; |
| 62 | 58 | ||
| 63 | // Check if any fire time fell within the last WINDOW_SECONDS. | 59 | // Find the next scheduled fire strictly after last_run. |
| 64 | let mut has_fired = false; | 60 | // - If last_run is None (freshly created or never run), start from the |
| 65 | for next_fire in schedule.after(&window_start) { | 61 | // last fire at/after the previous tick boundary so we still catch the |
| 62 | // fire scheduled for "now" on the first scan without doubling up. | ||
| 63 | // - If last_fire <= now, the job is due; we record last_run = last_fire | ||
| 64 | // (not `now`) so a long execution delay can't skip a legitimately | ||
| 65 | // pending later fire. | ||
| 66 | let reference: DateTime<Utc> = job.last_run.unwrap_or_else(|| { | ||
| 67 | // Step back a tick so the first scan after startup catches the fire | ||
| 68 | // scheduled for the minute we wake up on. | ||
| 69 | now - chrono::Duration::seconds(TICK_SECONDS as i64) | ||
| 70 | }); | ||
| 71 | |||
| 72 | let mut due_fire: Option<DateTime<Utc>> = None; | ||
| 73 | for next_fire in schedule.after(&reference) { | ||
| 66 | if next_fire > now { | 74 | if next_fire > now { |
| 67 | break; | 75 | break; |
| 68 | } | 76 | } |
| 69 | has_fired = true; | 77 | due_fire = Some(next_fire); |
| 70 | break; // one fire per window is enough | 78 | break; |
| 71 | } | 79 | } |
| 72 | 80 | ||
| 73 | if !has_fired { | 81 | let Some(fire_time) = due_fire else { |
| 74 | continue; | 82 | continue; |
| 75 | } | 83 | }; |
| 76 | 84 | ||
| 77 | info!("Executing cron job {}", job.id); | 85 | info!("Executing cron job {} (fire {})", job.id, fire_time); |
| 78 | match job.r#type.as_str() { | 86 | match job.r#type.as_str() { |
| 79 | "audio" => execute_audio_job(ctx, &job).await, | 87 | "audio" => execute_audio_job(ctx, &job).await, |
| 80 | "wiadomość" => execute_message_job(ctx, &job).await, | 88 | "wiadomość" => execute_message_job(ctx, &job).await, |
| 81 | _ => {} | 89 | _ => {} |
| 82 | } | 90 | } |
| 91 | |||
| 92 | // Record last_run so we never re-fire the same schedule instant on the | ||
| 93 | // next tick. Kept outside the match so a failed/empty send still marks | ||
| 94 | // the fire as consumed and doesn't retry on every tick. | ||
| 95 | if let Err(e) = database::update_cron_last_run(&ctx.pool, job.id, fire_time).await { | ||
| 96 | error!("Failed to update last_run for cron job {}: {e}", job.id); | ||
| 97 | } | ||
| 83 | } | 98 | } |
| 84 | } | 99 | } |
| 85 | 100 | ||
