From cd53d3659c65f817c7a6c34d7f36bc8e5f73d02f Mon Sep 17 00:00:00 2001
From: Andrey Antukh <niwi@niwi.nz>
Date: Wed, 5 Nov 2025 10:15:21 +0100
Subject: [PATCH] :bug: Truncate worker scheduled-at to milliseconds

The nanosecond precision has the problem with transit serialization
roundtrip used for pass data on the worker scheduler throught redis
and generates unnecesary rescheduling.
---
 backend/src/app/worker.clj        | 36 +++++++++++++++++--------------
 backend/src/app/worker/runner.clj |  7 ++++--
 2 files changed, 25 insertions(+), 18 deletions(-)

diff --git a/backend/src/app/worker.clj b/backend/src/app/worker.clj
index d1f728cdb7..a5d794e6f7 100644
--- a/backend/src/app/worker.clj
+++ b/backend/src/app/worker.clj
@@ -77,8 +77,8 @@
 ;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;
 
 (def ^:private sql:insert-new-task
-  "insert into task (id, name, props, queue, label, priority, max_retries, scheduled_at)
-   values (?, ?, ?, ?, ?, ?, ?, now() + ?)
+  "insert into task (id, name, props, queue, label, priority, max_retries, created_at, modified_at, scheduled_at)
+   values (?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
    returning id")
 
 (def ^:private
@@ -88,7 +88,7 @@
       AND queue=?
       AND label=?
       AND status = 'new'
-      AND scheduled_at > now()")
+      AND scheduled_at > ?")
 
 (def ^:private schema:options
   [:map {:title "submit-options"}
@@ -111,17 +111,19 @@
 
   (check-options! options)
 
-  (let [duration  (ct/duration delay)
-        interval  (db/interval duration)
-        props     (db/tjson params)
-        id        (uuid/next)
-        tenant    (cf/get :tenant)
-        task      (d/name task)
-        queue     (str/ffmt "%:%" tenant (d/name queue))
-        conn      (db/get-connectable options)
-        deleted   (when dedupe
-                    (-> (db/exec-one! conn [sql:remove-not-started-tasks task queue label])
-                        :next.jdbc/update-count))]
+  (let [delay        (ct/duration delay)
+        now          (ct/now)
+        scheduled-at (-> (ct/plus now delay)
+                         (ct/truncate :millisecond))
+        props        (db/tjson params)
+        id           (uuid/next)
+        tenant       (cf/get :tenant)
+        task         (d/name task)
+        queue        (str/ffmt "%:%" tenant (d/name queue))
+        conn         (db/get-connectable options)
+        deleted      (when dedupe
+                       (-> (db/exec-one! conn [sql:remove-not-started-tasks task queue label now])
+                           (db/get-update-count)))]
 
     (l/trc :hint "submit task"
            :name task
@@ -129,11 +131,13 @@
            :queue queue
            :label label
            :dedupe (boolean dedupe)
-           :delay (ct/format-duration duration)
+           :delay (ct/format-duration delay)
            :replace (or deleted 0))
 
     (db/exec-one! conn [sql:insert-new-task id task props queue
-                        label priority max-retries interval])
+                        label priority max-retries
+                        now now scheduled-at])
+
     id))
 
 (defn invoke!
diff --git a/backend/src/app/worker/runner.clj b/backend/src/app/worker/runner.clj
index f3f44bfd30..37fc55471a 100644
--- a/backend/src/app/worker/runner.clj
+++ b/backend/src/app/worker/runner.clj
@@ -158,7 +158,9 @@
             (inst-ms (:scheduled-at task)))
       (l/wrn :hint "skiping task, rescheduled"
              :task-id task-id
-             :runner-id id)
+             :runner-id id
+             :scheduled-at (ct/format-inst (:scheduled-at task))
+             :expected-scheduled-at (ct/format-inst scheduled-at))
 
       :else
       (let [result (run-task cfg task)]
@@ -179,7 +181,8 @@
                           {:error explain
                            :status "retry"
                            :modified-at now
-                           :scheduled-at (ct/plus now delay)
+                           :scheduled-at (-> (ct/plus now delay)
+                                             (ct/truncate :millisecond))
                            :retry-num nretry}
                           {:id (:id task)})
               nil))