feat!: normalize timestamps from TEXT to epoch millisecond INTEGER

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

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');

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.

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?;

src/store/lifecycle/cancel_expire.rs

@@ -169,15 +169,18 @@ impl TaskStore {
     pub async fn expire_tasks(&self) -> Result<Vec<TaskRecord>, 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<Option<TaskRecord>, 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?;
 

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)

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<Option<TaskRecord>, 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<Option<TaskRecord>, 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<bool, StoreError> {
+        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<Option<TaskRecord>, 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<String> = match (task.ttl_seconds, task.ttl_from)
-                        {
+                        let expires_at_ms: Option<i64> = 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,15 +766,15 @@ 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 = ?,
                         run_after = ?
                      WHERE id = ?",
                 )
                 .bind(error)
-                .bind(&run_after_str)
+                .bind(run_after_ms)
                 .bind(task.id)
                 .execute(&mut **conn)
                 .await?;

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<u64, StoreError> {
-        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())
     }
 

src/store/query/scheduling.rs

@@ -8,15 +8,15 @@ impl TaskStore {
     pub async fn next_run_after(
         &self,
     ) -> Result<Option<chrono::DateTime<chrono::Utc>>, 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",
         )
         .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.