aboutsummaryrefslogtreecommitdiffstats
path: root/src
diff options
context:
space:
mode:
authorsalmonik <raisersoft@proton.me>2026-07-09 16:20:43 +0200
committersalmonik <raisersoft@proton.me>2026-07-09 16:20:43 +0200
commit0768a2efa68b72107f2f2e7fa0f0dc87fd12d3c4 (patch)
tree160b0f61233fef8ef5e5e62c108b8652593aed7c /src
parentsearch for mp4's too (diff)
downloadsenator-sperma-0768a2efa68b72107f2f2e7fa0f0dc87fd12d3c4.tar.gz
senator-sperma-0768a2efa68b72107f2f2e7fa0f0dc87fd12d3c4.zip
fix cron double fire
Signed-off-by: salmonik <raisersoft@proton.me>
Diffstat (limited to '')
-rw-r--r--src/database/mod.rs19
-rw-r--r--src/tasks/mod.rs41
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
213pub 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
205pub async fn set_cron_job_active(pool: &DbPool, id: i64, active: bool) -> Result<bool> { 222pub 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 @@
1use std::sync::Arc; 1use std::sync::Arc;
2 2
3use chrono::{Duration, Timelike, Utc}; 3use chrono::{DateTime, Duration, Timelike, Utc};
4use cron::Schedule; 4use cron::Schedule;
5use poise::serenity_prelude as serenity; 5use poise::serenity_prelude as serenity;
6use std::str::FromStr; 6use 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.
38const WINDOW_SECONDS: i64 = 61;
39/// Tick interval for the scheduler loop. 36/// Tick interval for the scheduler loop.
40const TICK_SECONDS: u64 = 60; 37const 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