Merge pull request #4 from wildbit/utc-fix

Don’t rely on UTC for determining pending jobs

Author: temochka

CHANGELOG.md

@@ -1,3 +1,12 @@
 # Change Log
-All notable changes to this project will be documented in this file. This change log follows the conventions of [keepachangelog.com](http://keepachangelog.com/).
+All notable changes to this project are documented in this file. This change log follows the conventions of [keepachangelog.com](http://keepachangelog.com/).
+
+## [Unreleased]
+
+### Fixed
+- Internal: Fixed possible test failures in non-UTC timezones.
+- Internal: Fixed graceful shutdown test.
+
+### Changed
+- Use JVM time when querying for pending jobs instead of relying on UTC.
 

src/mysql_queue/queries.clj

@@ -70,12 +70,12 @@
                      scheduled_jobs LEFT JOIN
                      jobs ON scheduled_jobs.id=jobs.scheduled_job_id
                    WHERE
-                     scheduled_jobs.scheduled_for <= UTC_TIMESTAMP() AND
+                     scheduled_jobs.scheduled_for <= ? AND
                      jobs.id IS NULL AND
                      scheduled_jobs.name IN (" (in-query-stubs jobs-names) ") AND
                      scheduled_jobs.id NOT IN (" (in-query-stubs sieved-ids) ")
                    LIMIT ?")]
-            (concat jobs-names sieved-ids [n]))))
+            (concat [(java.util.Date.)] jobs-names sieved-ids [n]))))
 
 (defn select-jobs-by-ids
   [db ids]
@@ -95,12 +95,12 @@
                      jobs.status NOT IN (" (in-query-stubs ultimate-statuses) ") AND
                      jobs.name IN (" (in-query-stubs job-names) ") AND
                      jobs.id NOT IN (" (in-query-stubs sieved-ids) ") AND
-                     jobs.created_at + INTERVAL ? MINUTE <= UTC_TIMESTAMP()
+                     jobs.created_at + INTERVAL ? MINUTE <= ?
                    LIMIT ?")]
             ultimate-statuses
             job-names
             sieved-ids
-            [threshold-mins n])))
+            [threshold-mins (java.util.Date.) n])))
 
 (defn delete-scheduled-job-by-id!
   [db id]

test/mysql_queue/core_test.clj

@@ -160,26 +160,28 @@
         expected-set (->> num-jobs range (map inc) (into #{}))
         success? (promise)
         exception (promise)
+        lock (promise)
         check-ins (check-in-atom expected-set success?)
         jobs {:test-foo (fn [status {id :id :as args}]
+                          @lock
                           (Thread/sleep 1500)
                           (swap! check-ins conj id)
                           [:done args])}
         _ (dotimes [n num-jobs]
-            (let [scheduled-id (schedule-job db-conn :test-foo :begin {:id (inc n)} (java.util.Date.))]
-              (queries/insert-job<! db-conn scheduled-id 0 "test-foo" "begin" (pr-str {:id (inc n)}) 1)))]
+            (schedule-job db-conn :test-foo :begin {:id (inc n)} (java.util.Date.)))]
     (with-worker [wrk (worker db-conn
                               jobs
                               :num-consumer-threads 2
                               :err-fn #(deliver exception %)
                               :recovery-threshold-mins 0
                               :max-scheduler-sleep-interval 0.5
                               :max-recovery-sleep-interval 0.5)]
-      (Thread/sleep 500))
-      (is (deref success? 10 false)
-          (str "Failed to finish " num-jobs " test jobs taking 1500ms with 2s quit timeout.\n"
-               "Missing job IDs: " (clj-set/difference expected-set @check-ins) "\n"
-               "Exception?: " (deref exception 0 "nope")))
-      (is (= num-jobs (count @check-ins))
-          "The number of executed jobs doesn't match the number of jobs queued.")))
+      (deliver lock :unlocked)
+      (Thread/sleep 1000))
+    (is (deref success? 10 false)
+        (str "Failed to finish " num-jobs " test jobs taking 1500ms with 2s quit timeout.\n"
+             "Missing job IDs: " (clj-set/difference expected-set @check-ins) "\n"
+             "Exception?: " (deref exception 0 "nope")))
+    (is (= num-jobs (count @check-ins))
+        "The number of executed jobs doesn't match the number of jobs queued.")))