refactor: ensure tasks reach terminal state

Author: mping

.gitignore

@@ -29,4 +29,5 @@ docker/fdb.cluster
 
 # coverage
 merged.info
-coverage/
+coverage/
+.store/

DEVELOPMENT.md

@@ -36,8 +36,8 @@ clj -A:dev:doc:cljs
 # Tests
 
 ```shell
-bin/kaocha unit
-bin/kaocha unit-cljs
+bin/kaocha test
+bin/kaocha test-cljs
 
 ;; or run everything
 bin/run-coverage

bin/kaocha

@@ -2,11 +2,11 @@
 
 [ -d "node_modules/ws" ] || npm install ws
 
+#JAVA_OPTS="-agentlib:jdwp=transport=dt_socket,server=y,suspend=n,address=*:5005 -Dio.netty.tryUnsafe=false"
 JAVA_OPTS="-Dio.netty.tryUnsafe=false"
 if [[ "$OSTYPE" == "darwin"* ]]; then
   JAVA_OPTS="-Dio.netty.tryUnsafe=false -DFDB_LIBRARY_PATH_FDB_C=/usr/local/lib/libfdb_c.dylib -DFDB_LIBRARY_PATH_FDB_JAVA=/usr/local/lib/libfdb_java.jnilib"
 fi
 
-
 # bin/kaocha unit-cljs --reporter documentation --focus  cljs:intemporal.internal-failures-test --no-capture-output
 JAVA_OPTS="$JAVA_OPTS" clojure -A:dev:test:jdbc:fdb -M:test "$@"

bin/run-coverage

@@ -5,15 +5,18 @@ set -e
 BLUE='\033[0;34m'
 NC='\033[0m' # No Color
 
-
 rm -rf coverage
 mkdir -p coverage
 
+trap 'kill 0' SIGINT
+
 printf "${BLUE}running clj tests...${NC}\n"
-CLOVERAGE_OUTPUT=target/coverage-clj bin/kaocha unit --reporter documentation
+CLOVERAGE_OUTPUT=target/coverage-clj bin/kaocha test --reporter documentation &
 
 printf "${BLUE}running cljs tests...${NC}\n"
-CLOVERAGE_OUTPUT=target/coverage-cljs bin/kaocha unit-cljs --reporter documentation
+CLOVERAGE_OUTPUT=target/coverage-cljs bin/kaocha test-cljs --reporter documentation &
+
+wait
 
 printf "${BLUE}combining coverage reports...${NC}\n"
 lcov --add-tracefile target/coverage-clj/lcov.info  --add-tracefile target/coverage-cljs/lcov.info -o coverage/lcov.merged.info

docker/fdb.cluster

@@ -1 +1 @@
-docker:docker@172.19.0.2:4500
+docker:docker@172.19.0.3:4500

shadow-cljs.edn

@@ -13,6 +13,12 @@
                           :automata {:init-fn intemporal.automata/init
                                      :depends-on #{:shared}}}}
 
+          :node {:target    :node-test
+                 :output-to "target/tests.js"
+                 :ns-regexp "-test$"
+                 :compiler-options {:source-map "true"}
+                 :autorun   true}
+
           :dev {:target :browser
                 :asset-path "."
                 :output-dir "target"

src/intemporal/error.cljc

@@ -0,0 +1,16 @@
+(ns intemporal.error)
+
+(defn internal-error? [ex]
+  (when-let [t (-> ex ex-data ::type)]
+    (or (= :internal t)
+        (= :panic t))))
+
+(defn panic? [ex]
+  (and (instance? #?(:clj Exception :cljs js/Error) ex)
+       (= :panic (-> ex ex-data ::type))))
+
+(defn internal-error [msg data]
+  (ex-info msg (merge data {::type :internal})))
+
+(defn panic [msg]
+  (ex-info msg {::type :panic}))

src/intemporal/macros.cljc

@@ -74,7 +74,7 @@
                id#    (or (:id i/*env*) (i/random-id))
                fvar#  #'~wname
                task#  (i/create-workflow-task ref# root# (symbol fvar#) (macros/case :cljs fvar# :clj (var-get fvar#)) ~argv id#)]
-           (t/log! {:level :trace :_data {:env i/*env* :task task#}}  ["Invoking task with id" (:id task#)])
+           (t/log! {:level :debug :_data {:env i/*env* :task task#}}  ["Invoking task with id" (:id task#)])
            (w/enqueue-and-wait i/*env* task#))))))
 
 (defmacro stub-function
@@ -93,7 +93,7 @@
              ref#    nil ;; no enqueued task => no ref
              task#   (i/create-activity-task ref# root# (symbol fvar#) (macros/case :cljs fvar# :clj (var-get fvar#)) argv# id#)]
          ;; an embedded workflow engine doesn't need to have a task per invocation
-         (t/log! {:level :trace :_data {:env i/*env* :task task#}}  ["Invoking task"])
+         (t/log! {:level :debug :_data {:env i/*env* :task task#}}  ["Invoking task with id " id#])
          (let [res# (i/resume-task i/*env* store# protos# task#)]
            (macros/case
              :cljs res#
@@ -148,7 +148,7 @@
                                     [~@args]
                                     id#)]
 
-                      (t/log! {:level :trace :_data {:env i/*env* :task task#}}  ["Invoking task"])
+                      (t/log! {:level :debug :_data {:env i/*env* :task task#}}  ["Invoking task with id" id#])
                       (i/resume-task i/*env* store# protos# task#))))))))
                       ;(w/enqueue-and-wait i/*env* task#))))))))
 
@@ -191,7 +191,7 @@
                                   [~@args]
                                   id#)]
 
-                    (t/log! {:level :trace :_data {:env i/*env* :task task#}}  ["Invoking task"])
+                    (t/log! {:level :debug :_data {:env i/*env* :task task#}}  ["Invoking task with id" id#])
                     @(i/resume-task i/*env* store# protos# task#)))))))))
                     ;(w/enqueue-and-wait i/*env* task#)))))))))
 

src/intemporal/store.cljc

@@ -1,6 +1,6 @@
 (ns intemporal.store
   (:require [clojure.tools.reader.edn :as edn]
-            [intemporal.store.internal :refer [validate-task validate-event]]
+            [intemporal.store.internal :as si]
             [promesa.core :as p]
             [taoensso.telemere :as t]
             #?(:clj [clojure.java.io :as io]))
@@ -13,6 +13,8 @@
 
 (defprotocol TaskStore
   (list-tasks [this] "Lists all tasks")
+  (task<-panic [this task-id error]
+    "Terminates the task via panic; events should not be stored")
   (task<-event [this task-id event-descr]
     "Transitions the task. The task should be dequeued beforehand. Returns the event.
     `event-descr` is one of:
@@ -122,7 +124,7 @@
                        (when-let [w (find-task this id)]
                          (maybe-fail!)
                          (->> (apply assoc w kvs)
-                              (validate-task)
+                              (si/validate-task)
                               (swap! tasks assoc id))))]
 
      ;; deser the db
@@ -156,7 +158,7 @@
          (apply concat (vals @history)))
        (save-event [this task-id event]
          (let [evt+id (assoc event :id (swap! counter inc))]
-           (validate-event evt+id)
+           (si/validate-event evt+id)
            (swap! history (fn [v]
                             (assoc v task-id (-> (or (get v task-id) [])
                                                  (conj evt+id)))))
@@ -171,6 +173,9 @@
        (list-tasks [this]
          (vals @tasks))
 
+       (task<-panic [this task-id error]
+         (update-task this task-id :result error))
+
        (task<-event [this task-id {:keys [id ref root type sym args result error] :as event-descr}]
          ;; some redundancy between :result in task and event
          ;; note that we save the event first, because update-task can trigger some watchers
@@ -221,25 +226,22 @@
          (await-task this id {:timeout-ms default-lease}))
 
        (await-task [this id {:keys [timeout-ms] :as opts}]
-         ;; TODO handle internal errors? use a separate state?
          (maybe-fail!)
+         ;; TODO use owner
          (let [task        (find-task this id)
                deferred    (p/deferred)
-               completed?  (fn [{:keys [state]}]
-                             (or (= :success state)
-                                 (= :failure state)))
-               wrap-result (fn [{:keys [state result] :as task}]
+               wrap-result (fn [{:keys [result] :as task}]
                              (cond
-                               (= :success state) (p/resolved result)
-                               (= :failure state) (p/rejected result)
+                               (si/success? task) (p/resolved result)
+                               (si/failure? task) (p/rejected result)
                                :else (p/rejected (ex-info "Unknown state" {:task task}))))]
 
-           (if (completed? task)
+           (if (si/terminal? task)
              (wrap-result task)
              ;;else
              (do
-               (watch-task this id (fn [{:keys [state] :as task}]
-                                     (when (#{:success :failure} state)
+               (watch-task this id (fn [task]
+                                     (when (si/terminal? task)
                                        (p/resolve! deferred task))))
                ;; wait for resolution
                ;; remember: js doesnt have blocking op so we need to chain
@@ -267,7 +269,7 @@
        (enqueue-task [this task]
          ;; TODO use owner
          (maybe-fail!)
-         (validate-task task)
+         (si/validate-task task)
          (swap! tasks assoc
                 (:id task) (assoc task :order (swap! tcounter inc)))
          #?(:cljs (register this (:sym task) (:fvar task)))

src/intemporal/store/foundationdb.clj

@@ -1,7 +1,7 @@
 (ns intemporal.store.foundationdb
   (:require [intemporal.store :as store]
             [intemporal.workflow.internal :as i]
-            [intemporal.store.internal :refer [resolve-fvar serialize deserialize serializable? next-id validate-task validate-event]]
+            [intemporal.store.internal :as si :refer [resolve-fvar serialize deserialize serializable? next-id validate-task validate-event]]
             [me.vedang.clj-fdb.FDB :as cfdb]
             [me.vedang.clj-fdb.core :as fc]
             [me.vedang.clj-fdb.transaction :as ftr]
@@ -76,6 +76,14 @@
                (fc/get-range tx subspace-tasks {:valfn (comp resolve-fvar deserialize)}))
              (vals)))
 
+       (task<-panic [this task-id error]
+         (with-tx [tx (open-db)]
+           (let [task         (fc/get tx subspace-tasks task-id {:valfn (comp resolve-fvar deserialize)})
+                 updated-task (assoc task :result error)]
+             (when task
+               (validate-task updated-task)
+               (fc/set tx subspace-tasks task-id (serialize updated-task))))))
+
        (task<-event [this task-id {:keys [id ref root type sym args result error] :as event-descr}]
          ;; some redundancy between :result in task and event
          ;; note that we save the event first, because update-task can trigger some watchers
@@ -123,25 +131,20 @@
 
        (await-task [this id {:keys [timeout-ms] :as opts}]
          ;; TODO use owner
-         (let [task        (with-tx [tx (open-db)]
-                             (fc/get tx subspace-tasks [id]))
+         (let [task        (store/find-task this id)
                deferred    (p/deferred)
-               completed?  (fn [{:keys [state]}]
-                             (or (= :success state)
-                                 (= :failure state)))
                wrap-result (fn [{:keys [state result] :as task}]
                              (cond
-                               (= :success state) (p/resolved result)
-                               (= :failure state) (p/rejected result)
-                               ;; TODO throw internal error
+                               (si/success? task) (p/resolved result)
+                               (si/failure? task) (p/rejected result)
                                :else (p/rejected (ex-info "Unknown state" {:task task}))))]
 
-           (if (completed? task)
+           (if (si/terminal? task)
              (wrap-result task)
              ;;else
              (do
                (store/watch-task this id (fn [{:keys [state] :as task}]
-                                           (when (#{:success :failure} state)
+                                           (when (si/terminal? task)
                                              (p/resolve! deferred task))))
                ;; wait for resolution
                (-> (p/timeout deferred timeout-ms ::timeout)