From 9f480484720c32a7072c3242570e90c1c9dabac6 Mon Sep 17 00:00:00 2001
From: Michiel Borkent <michielborkent@gmail.com>
Date: Tue, 18 May 2021 13:19:09 +0200
Subject: [PATCH] [#847] Better error handling for parallel tasks (2)

---
 src/babashka/impl/tasks.clj   | 12 +++++++-----
 test/babashka/bb_edn_test.clj | 14 +++++++++++++-
 2 files changed, 20 insertions(+), 6 deletions(-)

diff --git a/src/babashka/impl/tasks.clj b/src/babashka/impl/tasks.clj
index 24c07a4f..a21a18c8 100644
--- a/src/babashka/impl/tasks.clj
+++ b/src/babashka/impl/tasks.clj
@@ -3,7 +3,7 @@
             [babashka.impl.common :refer [ctx bb-edn debug]]
             [babashka.impl.deps :as deps]
             [babashka.process :as p]
-            [clojure.core.async :refer [chan <!! alts!! thread]]
+            [clojure.core.async :refer [<!!]]
             [clojure.java.io :as io]
             [clojure.string :as str]
             [rewrite-clj.node :as node]
@@ -12,7 +12,7 @@
             [sci.core :as sci])
   (:import [clojure.core.async.impl.channels ManyToManyChannel]))
 
-(defn chan? [x]
+(defn -chan? [x]
   (instance? ManyToManyChannel x))
 
 (def sci-ns (sci/create-ns 'babashka.tasks nil))
@@ -123,7 +123,7 @@
 
 (defn -wait [res]
   (when res
-    (if (chan? res)
+    (if (-chan? res)
       (let [[_task-name res] (<!! res)]
         (if (instance? Throwable res)
           (throw (ex-info (ex-message res)
@@ -165,13 +165,14 @@
 (defn wait-tasks [deps]
   (if deps
     (format "
-(let [chans %s]
+(let [chans (filter babashka.tasks/-chan? %s)]
   (loop [cs chans]
     (let [[v p] (clojure.core.async/alts!! cs)
           [task-name v] v
           cs (filterv #(not= p %%) cs)
           ;; _ (.println System/err (str \"n: \" task-name \" v: \" v))
-          _ (intern *ns* (symbol task-name) v)]
+          ;; check for existence of v, as the channel may already have been consumed once
+          _ (when v (intern *ns* (symbol task-name) v))]
       (when (instance? Throwable v)
         (throw (ex-info (ex-message v)
                         {:babashka/exit 1
@@ -447,6 +448,7 @@
   {'shell (sci/copy-var shell sci-ns)
    'clojure (sci/copy-var clojure sci-ns)
    '-wait (sci/copy-var -wait sci-ns)
+   '-chan? (sci/copy-var -chan? sci-ns)
    '-err-thread (sci/copy-var -err-thread sci-ns)
    '*task* task
    'current-task current-task
diff --git a/test/babashka/bb_edn_test.clj b/test/babashka/bb_edn_test.clj
index 110c81f5..ae73f7fa 100644
--- a/test/babashka/bb_edn_test.clj
+++ b/test/babashka/bb_edn_test.clj
@@ -224,7 +224,19 @@
                                               (throw (ex-info "0 noes" {})))
                                         c {:depends [a b]}}}
         (is (thrown-with-msg? Exception #"0 noes"
-                              (bb "run" "--parallel" "c")))))))
+                              (bb "run" "--parallel" "c")))))
+    (testing "edge case"
+      (test-utils/with-config '{:tasks
+                                {a   (run '-a {:parallel true})
+                                 -a  {:depends [a:a a:b c]
+                                      :task (prn [a:a a:b c])}
+                                 a:a {:depends [c]
+                                      :task (+ 1 2 3)}
+                                 a:b {:depends [c]
+                                      :task (do (Thread/sleep 10)
+                                                (+ 1 2 3))}
+                                 c (do (Thread/sleep 10) :c)}}
+        (is (= [6 6 :c] (bb "run" "--prn" "a")))))))
 
 (deftest list-tasks-test
   (test-utils/with-config {}