diff --git a/migrations/001_tasks.sql b/migrations/001_tasks.sql index a90205f..a1c50dc 100644 --- a/migrations/001_tasks.sql +++ b/migrations/001_tasks.sql @@ -14,8 +14,8 @@ CREATE TABLE IF NOT EXISTS tasks ( expected_write_bytes INTEGER NOT NULL DEFAULT 0, retry_count INTEGER NOT NULL DEFAULT 0, last_error TEXT, - created_at TEXT NOT NULL DEFAULT (datetime('now')), - started_at TEXT, + created_at INTEGER NOT NULL, + started_at INTEGER, requeue INTEGER NOT NULL DEFAULT 0, requeue_priority INTEGER, parent_id INTEGER, @@ -42,9 +42,9 @@ CREATE TABLE IF NOT EXISTS task_history ( actual_write_bytes INTEGER, retry_count INTEGER NOT NULL DEFAULT 0, last_error TEXT, - created_at TEXT NOT NULL, - started_at TEXT, - completed_at TEXT NOT NULL DEFAULT (datetime('now')), + created_at INTEGER NOT NULL, + started_at INTEGER, + completed_at INTEGER NOT NULL, duration_ms INTEGER, parent_id INTEGER, fail_fast INTEGER NOT NULL DEFAULT 1 diff --git a/migrations/004_ttl.sql b/migrations/004_ttl.sql index e484328..18e881f 100644 --- a/migrations/004_ttl.sql +++ b/migrations/004_ttl.sql @@ -3,11 +3,11 @@ ALTER TABLE tasks ADD COLUMN ttl_seconds INTEGER; ALTER TABLE tasks ADD COLUMN ttl_from TEXT NOT NULL DEFAULT 'submission'; -ALTER TABLE tasks ADD COLUMN expires_at TEXT; +ALTER TABLE tasks ADD COLUMN expires_at INTEGER; ALTER TABLE task_history ADD COLUMN ttl_seconds INTEGER; ALTER TABLE task_history ADD COLUMN ttl_from TEXT NOT NULL DEFAULT 'submission'; -ALTER TABLE task_history ADD COLUMN expires_at TEXT; +ALTER TABLE task_history ADD COLUMN expires_at INTEGER; CREATE INDEX IF NOT EXISTS idx_tasks_expires ON tasks (expires_at ASC) - WHERE expires_at IS NOT NULL AND status IN ('pending', 'paused'); + WHERE expires_at IS NOT NULL AND status IN ('pending', 'paused', 'blocked'); diff --git a/migrations/005_scheduling.sql b/migrations/005_scheduling.sql index 69c6578..0904b8c 100644 --- a/migrations/005_scheduling.sql +++ b/migrations/005_scheduling.sql @@ -1,6 +1,6 @@ -- Delayed dispatch: task is pending but not eligible until this timestamp. -- NULL = immediately eligible (current behavior, backwards compatible). -ALTER TABLE tasks ADD COLUMN run_after TEXT; +ALTER TABLE tasks ADD COLUMN run_after INTEGER; -- Recurring task template fields. -- Only set on the "template" row that spawns recurring instances. @@ -10,7 +10,7 @@ ALTER TABLE tasks ADD COLUMN recurring_execution_count INTEGER NOT NULL DEFAULT ALTER TABLE tasks ADD COLUMN recurring_paused INTEGER NOT NULL DEFAULT 0; -- History: preserve the scheduling metadata for diagnostics. -ALTER TABLE task_history ADD COLUMN run_after TEXT; +ALTER TABLE task_history ADD COLUMN run_after INTEGER; -- Partial index: only pending tasks with a future run_after need time-gating. -- The scheduler's peek query uses this to skip not-yet-ready tasks. diff --git a/src/store/hierarchy.rs b/src/store/hierarchy.rs index 6f60576..1505113 100644 --- a/src/store/hierarchy.rs +++ b/src/store/hierarchy.rs @@ -27,9 +27,11 @@ impl TaskStore { /// Transition a waiting parent task back to `running` for finalization. pub async fn set_running_for_finalize(&self, id: i64) -> Result<(), StoreError> { + let now_ms = chrono::Utc::now().timestamp_millis(); sqlx::query( - "UPDATE tasks SET status = 'running', started_at = datetime('now') WHERE id = ? AND status = 'waiting'", + "UPDATE tasks SET status = 'running', started_at = ? WHERE id = ? AND status = 'waiting'", ) + .bind(now_ms) .bind(id) .execute(&self.pool) .await?; diff --git a/src/store/lifecycle/cancel_expire.rs b/src/store/lifecycle/cancel_expire.rs index d8b7971..80b0e3e 100644 --- a/src/store/lifecycle/cancel_expire.rs +++ b/src/store/lifecycle/cancel_expire.rs @@ -169,15 +169,18 @@ impl TaskStore { pub async fn expire_tasks(&self) -> Result, StoreError> { let mut conn = self.begin_write().await?; + let now_ms = chrono::Utc::now().timestamp_millis(); + // Find expired tasks (including blocked tasks — TTL ticks normally). let rows = sqlx::query( "SELECT * FROM tasks WHERE expires_at IS NOT NULL - AND expires_at <= datetime('now') + AND expires_at <= ? AND status IN ('pending', 'paused', 'blocked') ORDER BY expires_at ASC LIMIT 500", ) + .bind(now_ms) .fetch_all(&mut *conn) .await?; @@ -264,14 +267,16 @@ impl TaskStore { pub async fn expire_single(&self, id: i64) -> Result, StoreError> { let mut conn = self.begin_write().await?; + let now_ms = chrono::Utc::now().timestamp_millis(); let row = sqlx::query( "SELECT * FROM tasks WHERE id = ? AND expires_at IS NOT NULL - AND expires_at <= datetime('now') + AND expires_at <= ? AND status IN ('pending', 'paused')", ) .bind(id) + .bind(now_ms) .fetch_optional(&mut *conn) .await?; diff --git a/src/store/lifecycle/mod.rs b/src/store/lifecycle/mod.rs index 2058578..096ae94 100644 --- a/src/store/lifecycle/mod.rs +++ b/src/store/lifecycle/mod.rs @@ -68,13 +68,15 @@ pub(crate) async fn insert_history( } else { task.retry_count }; + let completed_at_ms = chrono::Utc::now().timestamp_millis(); + let result = sqlx::query( "INSERT INTO task_history (task_type, key, label, priority, status, payload, expected_read_bytes, expected_write_bytes, expected_net_rx_bytes, expected_net_tx_bytes, actual_read_bytes, actual_write_bytes, actual_net_rx_bytes, actual_net_tx_bytes, - retry_count, last_error, created_at, started_at, duration_ms, parent_id, fail_fast, group_key, + retry_count, last_error, created_at, started_at, completed_at, duration_ms, parent_id, fail_fast, group_key, ttl_seconds, ttl_from, expires_at, run_after, max_retries, memo) - VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)", + VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)", ) .bind(&task.task_type) .bind(&task.key) @@ -92,25 +94,17 @@ pub(crate) async fn insert_history( .bind(metrics.net_tx) .bind(retry_count) .bind(last_error) - .bind(task.created_at.format("%Y-%m-%d %H:%M:%S").to_string()) - .bind( - task.started_at - .map(|dt| dt.format("%Y-%m-%d %H:%M:%S").to_string()), - ) + .bind(task.created_at.timestamp_millis()) + .bind(task.started_at.map(|dt| dt.timestamp_millis())) + .bind(completed_at_ms) .bind(duration_ms) .bind(task.parent_id) .bind(fail_fast_val) .bind(&task.group_key) .bind(task.ttl_seconds) .bind(task.ttl_from.as_str()) - .bind( - task.expires_at - .map(|dt| dt.format("%Y-%m-%d %H:%M:%S").to_string()), - ) - .bind( - task.run_after - .map(|dt| dt.format("%Y-%m-%d %H:%M:%S").to_string()), - ) + .bind(task.expires_at.map(|dt| dt.timestamp_millis())) + .bind(task.run_after.map(|dt| dt.timestamp_millis())) .bind(task.max_retries) .bind(&task.memo) .execute(&mut **conn) diff --git a/src/store/lifecycle/transitions.rs b/src/store/lifecycle/transitions.rs index 3db0ff2..fb51b0c 100644 --- a/src/store/lifecycle/transitions.rs +++ b/src/store/lifecycle/transitions.rs @@ -27,16 +27,18 @@ impl TaskStore { /// Returns `None` if the queue is empty. Tasks with a future `run_after` /// timestamp are excluded (not yet eligible for dispatch). pub async fn peek_next(&self) -> Result, StoreError> { + let now_ms = chrono::Utc::now().timestamp_millis(); let row = sqlx::query( "SELECT * FROM tasks WHERE id = ( SELECT id FROM tasks WHERE status = 'pending' - AND (run_after IS NULL OR run_after <= strftime('%Y-%m-%d %H:%M:%f', 'now')) + AND (run_after IS NULL OR run_after <= ?) ORDER BY priority ASC, id ASC LIMIT 1 )", ) + .bind(now_ms) .fetch_optional(&self.pool) .await?; @@ -55,18 +57,21 @@ impl TaskStore { /// first pop (when `expires_at IS NULL` and `ttl_seconds IS NOT NULL`). pub async fn pop_by_id(&self, id: i64) -> Result, StoreError> { tracing::debug!(task_id = id, "store.pop_by_id: UPDATE start"); + let now_ms = chrono::Utc::now().timestamp_millis(); let row = sqlx::query( "UPDATE tasks SET status = 'running', - started_at = datetime('now'), + started_at = ?, expires_at = CASE WHEN ttl_from = 'first_attempt' AND ttl_seconds IS NOT NULL AND expires_at IS NULL - THEN datetime('now', '+' || ttl_seconds || ' seconds') + THEN ? + (ttl_seconds * 1000) ELSE expires_at END WHERE id = ? AND status = 'pending' RETURNING *", ) + .bind(now_ms) + .bind(now_ms) .bind(id) .fetch_optional(&self.pool) .await?; @@ -85,17 +90,20 @@ impl TaskStore { /// prior `peek_next` and will update its in-memory fields directly. This /// avoids the `RETURNING *` round-trip of [`pop_by_id`](Self::pop_by_id). pub(crate) async fn claim_task(&self, id: i64) -> Result { + let now_ms = chrono::Utc::now().timestamp_millis(); let result = sqlx::query( "UPDATE tasks SET status = 'running', - started_at = datetime('now'), + started_at = ?, expires_at = CASE WHEN ttl_from = 'first_attempt' AND ttl_seconds IS NOT NULL AND expires_at IS NULL - THEN datetime('now', '+' || ttl_seconds || ' seconds') + THEN ? + (ttl_seconds * 1000) ELSE expires_at END WHERE id = ? AND status = 'pending'", ) + .bind(now_ms) + .bind(now_ms) .bind(id) .execute(&self.pool) .await?; @@ -114,25 +122,30 @@ impl TaskStore { return Ok(self.pop_next().await?.into_iter().collect()); } + let now_ms = chrono::Utc::now().timestamp_millis(); let rows = sqlx::query( "UPDATE tasks SET status = 'running', - started_at = datetime('now'), + started_at = ?, expires_at = CASE WHEN ttl_from = 'first_attempt' AND ttl_seconds IS NOT NULL AND expires_at IS NULL - THEN datetime('now', '+' || ttl_seconds || ' seconds') + THEN ? + (ttl_seconds * 1000) ELSE expires_at END WHERE id IN ( SELECT id FROM tasks WHERE status = 'pending' - AND (run_after IS NULL OR run_after <= strftime('%Y-%m-%d %H:%M:%f', 'now')) - AND (expires_at IS NULL OR expires_at > strftime('%Y-%m-%d %H:%M:%f', 'now')) + AND (run_after IS NULL OR run_after <= ?) + AND (expires_at IS NULL OR expires_at > ?) ORDER BY priority ASC, id ASC LIMIT ? ) RETURNING *", ) + .bind(now_ms) + .bind(now_ms) + .bind(now_ms) + .bind(now_ms) .bind(limit as i64) .fetch_all(&self.pool) .await?; @@ -149,25 +162,30 @@ impl TaskStore { /// For tasks with `ttl_from = 'first_attempt'`, sets `expires_at` on /// the first pop. pub async fn pop_next(&self) -> Result, StoreError> { + let now_ms = chrono::Utc::now().timestamp_millis(); let row = sqlx::query( "UPDATE tasks SET status = 'running', - started_at = datetime('now'), + started_at = ?, expires_at = CASE WHEN ttl_from = 'first_attempt' AND ttl_seconds IS NOT NULL AND expires_at IS NULL - THEN datetime('now', '+' || ttl_seconds || ' seconds') + THEN ? + (ttl_seconds * 1000) ELSE expires_at END WHERE id = ( SELECT id FROM tasks WHERE status = 'pending' - AND (run_after IS NULL OR run_after <= strftime('%Y-%m-%d %H:%M:%f', 'now')) - AND (expires_at IS NULL OR expires_at > strftime('%Y-%m-%d %H:%M:%f', 'now')) + AND (run_after IS NULL OR run_after <= ?) + AND (expires_at IS NULL OR expires_at > ?) ORDER BY priority ASC, id ASC LIMIT 1 ) RETURNING *", ) + .bind(now_ms) + .bind(now_ms) + .bind(now_ms) + .bind(now_ms) .fetch_optional(&self.pool) .await?; @@ -534,16 +552,17 @@ impl TaskStore { .await?; if existing.is_none() { - let next_run = chrono::Utc::now() + chrono::Duration::seconds(interval); - let next_run_str = next_run.format("%Y-%m-%d %H:%M:%S").to_string(); + let now = chrono::Utc::now(); + let next_run = now + chrono::Duration::seconds(interval); + let next_run_ms = next_run.timestamp_millis(); + let now_ms = now.timestamp_millis(); let fail_fast_val: i32 = if task.fail_fast { 1 } else { 0 }; // Compute TTL columns for the next instance. - let expires_at_str: Option = match (task.ttl_seconds, task.ttl_from) - { + let expires_at_ms: Option = match (task.ttl_seconds, task.ttl_from) { (Some(ttl_secs), crate::task::TtlFrom::Submission) => { - let exp = chrono::Utc::now() + chrono::Duration::seconds(ttl_secs); - Some(exp.format("%Y-%m-%d %H:%M:%S").to_string()) + let exp = now + chrono::Duration::seconds(ttl_secs); + Some(exp.timestamp_millis()) } _ => None, }; @@ -556,8 +575,8 @@ impl TaskStore { ttl_seconds, ttl_from, expires_at, run_after, recurring_interval_secs, recurring_max_executions, recurring_execution_count, - recurring_paused, max_retries) - VALUES (?, ?, ?, ?, 'pending', ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, 0, ?)", + recurring_paused, max_retries, created_at) + VALUES (?, ?, ?, ?, 'pending', ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, 0, ?, ?)", ) .bind(&task.task_type) .bind(&task.key) @@ -573,12 +592,13 @@ impl TaskStore { .bind(&task.group_key) .bind(task.ttl_seconds) .bind(task.ttl_from.as_str()) - .bind(&expires_at_str) - .bind(&next_run_str) + .bind(expires_at_ms) + .bind(next_run_ms) .bind(task.recurring_interval_secs) .bind(task.recurring_max_executions) .bind(execution_count) .bind(task.max_retries) + .bind(now_ms) .execute(&mut **conn) .await?; @@ -746,7 +766,7 @@ impl TaskStore { // Delayed retry — set run_after. let run_after = chrono::Utc::now() + chrono::Duration::milliseconds(delay.as_millis() as i64); - let run_after_str = run_after.format("%Y-%m-%d %H:%M:%S%.3f").to_string(); + let run_after_ms = run_after.timestamp_millis(); sqlx::query( "UPDATE tasks SET status = 'pending', started_at = NULL, retry_count = retry_count + 1, last_error = ?, @@ -754,7 +774,7 @@ impl TaskStore { WHERE id = ?", ) .bind(error) - .bind(&run_after_str) + .bind(run_after_ms) .bind(task.id) .execute(&mut **conn) .await?; diff --git a/src/store/mod.rs b/src/store/mod.rs index 8cb0439..78912da 100644 --- a/src/store/mod.rs +++ b/src/store/mod.rs @@ -382,11 +382,11 @@ impl TaskStore { /// Prune history records older than `max_age_days` days. /// Returns the number of records deleted. pub async fn prune_history_by_age(&self, max_age_days: i64) -> Result { - let result = - sqlx::query("DELETE FROM task_history WHERE completed_at < datetime('now', ?)") - .bind(format!("-{max_age_days} days")) - .execute(&self.pool) - .await?; + let cutoff = (chrono::Utc::now() - chrono::Duration::days(max_age_days)).timestamp_millis(); + let result = sqlx::query("DELETE FROM task_history WHERE completed_at < ?") + .bind(cutoff) + .execute(&self.pool) + .await?; Ok(result.rows_affected()) } diff --git a/src/store/query/scheduling.rs b/src/store/query/scheduling.rs index 7ada5ce..5b0b9fa 100644 --- a/src/store/query/scheduling.rs +++ b/src/store/query/scheduling.rs @@ -8,7 +8,7 @@ impl TaskStore { pub async fn next_run_after( &self, ) -> Result>, StoreError> { - let row: Option<(String,)> = sqlx::query_as( + let row: Option<(i64,)> = sqlx::query_as( "SELECT run_after FROM tasks WHERE status = 'pending' AND run_after IS NOT NULL ORDER BY run_after ASC LIMIT 1", @@ -16,7 +16,7 @@ impl TaskStore { .fetch_optional(&self.pool) .await?; - Ok(row.map(|(s,)| crate::store::row_mapping::parse_datetime(&s))) + Ok(row.map(|(ms,)| crate::store::row_mapping::from_epoch_ms(ms))) } /// List active recurring schedules with their next run times. diff --git a/src/store/row_mapping.rs b/src/store/row_mapping.rs index bdb3833..8325914 100644 --- a/src/store/row_mapping.rs +++ b/src/store/row_mapping.rs @@ -9,65 +9,23 @@ use crate::task::{ TtlFrom, }; -pub(crate) fn parse_datetime(s: &str) -> DateTime { - // SQLite stores as "YYYY-MM-DD HH:MM:SS" or "YYYY-MM-DD HH:MM:SS.mmm". - // Fast fixed-position byte parser instead of the generic chrono parser. - let b = s.as_bytes(); - if b.len() < 19 { - return DateTime::::default(); - } - - let year = parse_4(b, 0); - let month = parse_2(b, 5); - let day = parse_2(b, 8); - let hour = parse_2(b, 11); - let min = parse_2(b, 14); - let sec = parse_2(b, 17); - - let nanos = if b.len() > 20 && b[19] == b'.' { - parse_frac_nanos(b, 20) - } else { - 0 - }; - - chrono::NaiveDate::from_ymd_opt(year, month, day) - .and_then(|d| d.and_hms_nano_opt(hour, min, sec, nanos)) - .map(|ndt| ndt.and_utc()) - .unwrap_or_default() -} - -#[inline(always)] -fn parse_2(b: &[u8], off: usize) -> u32 { - (b[off] - b'0') as u32 * 10 + (b[off + 1] - b'0') as u32 +/// Convert a `DateTime` to epoch milliseconds for SQLite INTEGER storage. +#[allow(dead_code)] +pub(crate) fn epoch_ms(dt: DateTime) -> i64 { + dt.timestamp_millis() } -#[inline(always)] -fn parse_4(b: &[u8], off: usize) -> i32 { - (b[off] - b'0') as i32 * 1000 - + (b[off + 1] - b'0') as i32 * 100 - + (b[off + 2] - b'0') as i32 * 10 - + (b[off + 3] - b'0') as i32 -} - -#[inline(always)] -fn parse_frac_nanos(b: &[u8], start: usize) -> u32 { - let frac_len = (b.len() - start).min(9); - let mut val: u32 = 0; - for i in 0..frac_len { - val = val * 10 + (b[start + i] - b'0') as u32; - } - // Pad to 9 digits (nanoseconds). - for _ in frac_len..9 { - val *= 10; - } - val +/// Convert epoch milliseconds from SQLite INTEGER back to `DateTime`. +/// +/// Returns `1970-01-01T00:00:00Z` on corrupt/invalid data — same silent-fallback +/// behavior as the previous `parse_datetime` on malformed TEXT. +pub(crate) fn from_epoch_ms(ms: i64) -> DateTime { + DateTime::from_timestamp_millis(ms).unwrap_or_default() } pub(crate) fn row_to_task_record(row: &sqlx::sqlite::SqliteRow) -> TaskRecord { let priority_val: i32 = row.get("priority"); let status_str: String = row.get("status"); - let created_at_str: String = row.get("created_at"); - let started_at_str: Option = row.get("started_at"); let requeue_val: i32 = row.get("requeue"); let requeue_priority_val: Option = row.get("requeue_priority"); @@ -75,8 +33,6 @@ pub(crate) fn row_to_task_record(row: &sqlx::sqlite::SqliteRow) -> TaskRecord { let fail_fast_val: i32 = row.get("fail_fast"); let ttl_from_str: String = row.get("ttl_from"); - let expires_at_str: Option = row.get("expires_at"); - let run_after_str: Option = row.get("run_after"); let recurring_paused_val: i32 = row.get("recurring_paused"); let on_dep_failure_str: String = row.get("on_dep_failure"); @@ -96,8 +52,8 @@ pub(crate) fn row_to_task_record(row: &sqlx::sqlite::SqliteRow) -> TaskRecord { }, retry_count: row.get("retry_count"), last_error: row.get("last_error"), - created_at: parse_datetime(&created_at_str), - started_at: started_at_str.map(|s| parse_datetime(&s)), + created_at: from_epoch_ms(row.get("created_at")), + started_at: row.get::, _>("started_at").map(from_epoch_ms), requeue: requeue_val != 0, requeue_priority: requeue_priority_val.map(|p| Priority::new(p as u8)), parent_id, @@ -105,8 +61,8 @@ pub(crate) fn row_to_task_record(row: &sqlx::sqlite::SqliteRow) -> TaskRecord { group_key: row.get("group_key"), ttl_seconds: row.get("ttl_seconds"), ttl_from: ttl_from_str.parse().unwrap_or(TtlFrom::Submission), - expires_at: expires_at_str.map(|s| parse_datetime(&s)), - run_after: run_after_str.map(|s| parse_datetime(&s)), + expires_at: row.get::, _>("expires_at").map(from_epoch_ms), + run_after: row.get::, _>("run_after").map(from_epoch_ms), recurring_interval_secs: row.get("recurring_interval_secs"), recurring_max_executions: row.get("recurring_max_executions"), recurring_execution_count: row.get("recurring_execution_count"), @@ -126,9 +82,6 @@ pub(crate) fn row_to_task_record(row: &sqlx::sqlite::SqliteRow) -> TaskRecord { pub(crate) fn row_to_history_record(row: &sqlx::sqlite::SqliteRow) -> TaskHistoryRecord { let priority_val: i32 = row.get("priority"); let status_str: String = row.get("status"); - let created_at_str: String = row.get("created_at"); - let started_at_str: Option = row.get("started_at"); - let completed_at_str: String = row.get("completed_at"); let parent_id: Option = row.get("parent_id"); let fail_fast_val: i32 = row.get("fail_fast"); @@ -145,8 +98,6 @@ pub(crate) fn row_to_history_record(row: &sqlx::sqlite::SqliteRow) -> TaskHistor }); let ttl_from_str: String = row.get("ttl_from"); - let expires_at_str: Option = row.get("expires_at"); - let run_after_str: Option = row.get("run_after"); TaskHistoryRecord { id: row.get("id"), @@ -165,17 +116,17 @@ pub(crate) fn row_to_history_record(row: &sqlx::sqlite::SqliteRow) -> TaskHistor actual_io, retry_count: row.get("retry_count"), last_error: row.get("last_error"), - created_at: parse_datetime(&created_at_str), - started_at: started_at_str.map(|s| parse_datetime(&s)), - completed_at: parse_datetime(&completed_at_str), + created_at: from_epoch_ms(row.get("created_at")), + started_at: row.get::, _>("started_at").map(from_epoch_ms), + completed_at: from_epoch_ms(row.get("completed_at")), duration_ms: row.get("duration_ms"), parent_id, fail_fast: fail_fast_val != 0, group_key: row.get("group_key"), ttl_seconds: row.get("ttl_seconds"), ttl_from: ttl_from_str.parse().unwrap_or(TtlFrom::Submission), - expires_at: expires_at_str.map(|s| parse_datetime(&s)), - run_after: run_after_str.map(|s| parse_datetime(&s)), + expires_at: row.get::, _>("expires_at").map(from_epoch_ms), + run_after: row.get::, _>("run_after").map(from_epoch_ms), // Tags are populated separately from the task_history_tags table. tags: std::collections::HashMap::new(), max_retries: row.get("max_retries"), @@ -188,33 +139,38 @@ mod tests { use super::*; #[test] - fn parse_whole_seconds() { - let dt = parse_datetime("2024-01-15 09:30:45"); - assert_eq!(dt.to_string(), "2024-01-15 09:30:45 UTC"); - } - - #[test] - fn parse_fractional_millis() { - let dt = parse_datetime("2024-01-15 09:30:45.123"); - assert_eq!(dt.to_string(), "2024-01-15 09:30:45.123 UTC"); - assert_eq!(dt.timestamp_subsec_millis(), 123); + fn epoch_ms_round_trip() { + let now = Utc::now(); + let ms = epoch_ms(now); + let back = from_epoch_ms(ms); + // Round-trip preserves millisecond precision. + assert_eq!(now.timestamp_millis(), back.timestamp_millis()); } #[test] - fn parse_fractional_micros() { - let dt = parse_datetime("2024-01-15 09:30:45.123456"); - assert_eq!(dt.to_string(), "2024-01-15 09:30:45.123456 UTC"); + fn epoch_ms_zero() { + let dt = from_epoch_ms(0); + assert_eq!(dt, DateTime::::default()); } #[test] - fn parse_short_string_returns_default() { - let dt = parse_datetime("bad"); - assert_eq!(dt, DateTime::::default()); + fn epoch_ms_negative() { + // Negative epoch (before 1970) should still round-trip. + let ms = -86_400_000i64; // 1969-12-31 + let dt = from_epoch_ms(ms); + assert_eq!(dt.timestamp_millis(), ms); } #[test] - fn parse_empty_returns_default() { - let dt = parse_datetime(""); - assert_eq!(dt, DateTime::::default()); + fn epoch_ms_known_value() { + // 2024-01-15 09:30:45.123 UTC + let dt = chrono::NaiveDate::from_ymd_opt(2024, 1, 15) + .unwrap() + .and_hms_milli_opt(9, 30, 45, 123) + .unwrap() + .and_utc(); + let ms = epoch_ms(dt); + let back = from_epoch_ms(ms); + assert_eq!(back.to_string(), "2024-01-15 09:30:45.123 UTC"); } } diff --git a/src/store/submit/dedup.rs b/src/store/submit/dedup.rs index ec0b0a8..66baa06 100644 --- a/src/store/submit/dedup.rs +++ b/src/store/submit/dedup.rs @@ -92,10 +92,10 @@ pub(crate) async fn supersede_existing( // Compute TTL columns for the new submission. let ttl_seconds = sub.ttl.map(|d| d.as_secs() as i64); let ttl_from_str = sub.ttl_from.as_str(); - let expires_at: Option = match (sub.ttl, sub.ttl_from) { + let expires_at: Option = match (sub.ttl, sub.ttl_from) { (Some(ttl), TtlFrom::Submission) => { let exp = chrono::Utc::now() + ttl; - Some(exp.format("%Y-%m-%d %H:%M:%S").to_string()) + Some(exp.timestamp_millis()) } _ => None, }; @@ -126,7 +126,7 @@ pub(crate) async fn supersede_existing( .bind(&sub.group_key) .bind(ttl_seconds) .bind(ttl_from_str) - .bind(&expires_at) + .bind(expires_at) .bind(sub.max_retries) .bind(replaced_id) .execute(&mut **conn) @@ -149,12 +149,13 @@ pub(crate) async fn supersede_existing( .execute(&mut **conn) .await?; + let now_ms = chrono::Utc::now().timestamp_millis(); let result = sqlx::query( "INSERT INTO tasks (task_type, key, label, priority, payload, expected_read_bytes, expected_write_bytes, expected_net_rx_bytes, expected_net_tx_bytes, parent_id, fail_fast, group_key, - ttl_seconds, ttl_from, expires_at, max_retries) - VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)", + ttl_seconds, ttl_from, expires_at, max_retries, created_at) + VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)", ) .bind(&sub.task_type) .bind(key) @@ -170,8 +171,9 @@ pub(crate) async fn supersede_existing( .bind(&sub.group_key) .bind(ttl_seconds) .bind(ttl_from_str) - .bind(&expires_at) + .bind(expires_at) .bind(sub.max_retries) + .bind(now_ms) .execute(&mut **conn) .await?; diff --git a/src/store/submit/mod.rs b/src/store/submit/mod.rs index fd6c9dc..d0ac2f2 100644 --- a/src/store/submit/mod.rs +++ b/src/store/submit/mod.rs @@ -70,18 +70,16 @@ pub(crate) async fn submit_one( // Compute TTL columns. let ttl_seconds = sub.ttl.map(|d| d.as_secs() as i64); let ttl_from_str = sub.ttl_from.as_str(); - let expires_at: Option = match (sub.ttl, sub.ttl_from) { + let expires_at: Option = match (sub.ttl, sub.ttl_from) { (Some(ttl), TtlFrom::Submission) => { let exp = chrono::Utc::now() + ttl; - Some(exp.format("%Y-%m-%d %H:%M:%S").to_string()) + Some(exp.timestamp_millis()) } _ => None, // FirstAttempt: set on pop; no TTL: NULL }; // Compute scheduling columns. - let run_after_str: Option = sub - .run_after - .map(|dt| dt.format("%Y-%m-%d %H:%M:%S").to_string()); + let run_after_ms: Option = sub.run_after.map(|dt| dt.timestamp_millis()); let recurring_interval_secs: Option = sub.recurring.as_ref().map(|r| r.interval.as_secs() as i64); let recurring_max_executions: Option = sub @@ -98,9 +96,11 @@ pub(crate) async fn submit_one( let on_dep_failure_str = sub.on_dependency_failure.as_str(); + let now_ms = chrono::Utc::now().timestamp_millis(); + let result = sqlx::query( - "INSERT OR IGNORE INTO tasks (task_type, key, label, priority, payload, expected_read_bytes, expected_write_bytes, expected_net_rx_bytes, expected_net_tx_bytes, parent_id, fail_fast, group_key, ttl_seconds, ttl_from, expires_at, run_after, recurring_interval_secs, recurring_max_executions, on_dep_failure, max_retries) - VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)", + "INSERT OR IGNORE INTO tasks (task_type, key, label, priority, payload, expected_read_bytes, expected_write_bytes, expected_net_rx_bytes, expected_net_tx_bytes, parent_id, fail_fast, group_key, ttl_seconds, ttl_from, expires_at, run_after, recurring_interval_secs, recurring_max_executions, on_dep_failure, max_retries, created_at) + VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)", ) .bind(&sub.task_type) .bind(&key) @@ -116,12 +116,13 @@ pub(crate) async fn submit_one( .bind(&sub.group_key) .bind(ttl_seconds) .bind(ttl_from_str) - .bind(&expires_at) - .bind(&run_after_str) + .bind(expires_at) + .bind(run_after_ms) .bind(recurring_interval_secs) .bind(recurring_max_executions) .bind(on_dep_failure_str) .bind(sub.max_retries) + .bind(now_ms) .execute(&mut **conn) .await?;