Merge pull request #6533 from metabase/async-test-fix-race-condition

Fix async test race conditions, part II

src/metabase/feature_extraction/async.clj

@@ -19,6 +19,10 @@
   "Is the computation job still running?"
   (comp some? #{:running} :status))
 
+(def ^{:arglists '([job])} canceled?
+  "Has the computation job been canceled?"
+  (comp some? #{:canceled} :status))
+
 (defn- save-result
   [{:keys [id]} payload]
   (when-not (future-cancelled? (@running-jobs id))

test/metabase/test/async.clj

@@ -4,19 +4,7 @@
             [metabase.feature-extraction.async :as async]
             [metabase.models.computation-job :refer [ComputationJob]]))
 
-(defn result!
-  "Blocking version of async/result."
-  [job-id]
-  (when-let [f (-> #'async/running-jobs
-                   deref                  ; var
-                   deref                  ; atom
-                   (get job-id))]
-    (when-not (or (future-cancelled? f)
-                  (future-done? f))
-      @f))
-  (async/result (ComputationJob job-id)))
-
-(def ^:dynamic *max-while-runtime*
+(def ^:dynamic ^Integer *max-while-runtime*
   "Maximal time in milliseconds `while-with-timeout` runs."
   10000000)
 
@@ -30,3 +18,17 @@
        ~@body)
      (when (>= (- (System/currentTimeMillis) start#) *max-while-runtime*)
        (log/warn "While loop terminated due to exceeded max runtime."))))
+
+(defn result!
+  "Blocking version of async/result."
+  [job-id]
+  (when-let [f (-> #'async/running-jobs
+                   deref                  ; var
+                   deref                  ; atom
+                   (get job-id))]
+    (when-not (or (future-cancelled? f)
+                  (future-done? f))
+      @f))
+  ; Wait for the transaction to finish
+  (while-with-timeout (-> job-id ComputationJob async/running?))
+  (async/result (ComputationJob job-id)))