Add slingshot based exception handling

Signed-off-by: Greg Haskins <[email protected]>

Author: ghaskins

doc/workflows.md

@@ -219,3 +219,21 @@ A query consists of a `query-type` (keyword) and possibly some `args` (any seria
 ```clojure
 (query workflow :my-query {:foo "bar"})
 ```
+
+## Exceptions
+
+This SDK integrates with the [slingshot](https://github.com/scgilardi/slingshot) library.  Stones cast with slingshot's throw+ are serialized and re-thrown across activity and workflow boundaries in a transparent manner that is compatible with slingshot idiomatic try+ based catch blocks.
+
+### Managing Retries
+By default, stones cast that are not caught locally by an activity or workflow trigger ApplicationFailure semantics and are thus subject to the overall Retry Policies in place.   However, the developer may force a given stone to be non-retriable by setting the flag '::non-retriable?' within the object.
+
+Example:
+
+```clojure
+(require `[temporal.exceptions :as e])
+(require `[slingshot.slingshot :refer [throw+]])
+
+(defactivity my-activity
+   [ctx args]
+   (throw+ {:type ::my-fatal-error :msg "this error is non-retriable" ::e/non-retriable? true}))
+```

project.clj

@@ -19,7 +19,8 @@
                  [com.taoensso/timbre "6.3.1"]
                  [com.taoensso/nippy "3.3.0"]
                  [funcool/promesa "9.2.542"]
-                 [medley "1.4.0"]]
+                 [medley "1.4.0"]
+                 [slingshot "0.12.2"]]
   :repl-options {:init-ns user}
   :java-source-paths ["src"]
   :javac-options ["-target" "11" "-source" "11"]

src/temporal/activity.clj

@@ -5,6 +5,7 @@
   (:require [taoensso.timbre :as log]
             [taoensso.nippy :as nippy]
             [promesa.core :as p]
+            [temporal.internal.exceptions :as e]
             [temporal.internal.activity :as a]
             [temporal.internal.utils :as u]
             [temporal.internal.promise])                    ;; needed for IPromise protocol extention
@@ -98,6 +99,7 @@ Arguments:
      (log/trace "invoke:" activity "with" params options)
      (-> (.executeAsync stub act-name u/bytes-type (u/->objarray params))
          (p/then (partial complete-invoke activity))
+         (p/catch e/slingshot? e/recast-stone)
          (p/catch (fn [e]
                     (log/error e)
                     (throw e)))))))
@@ -138,6 +140,7 @@ Arguments:
      (log/trace "local-invoke:" activity "with" params options)
      (-> (.executeAsync stub act-name u/bytes-type (u/->objarray params))
          (p/then (partial complete-invoke activity))
+         (p/catch e/slingshot? e/recast-stone)
          (p/catch (fn [e]
                     (log/error e)
                     (throw e)))))))

src/temporal/client/core.clj

@@ -6,7 +6,8 @@
             [taoensso.nippy :as nippy]
             [promesa.core :as p]
             [temporal.internal.workflow :as w]
-            [temporal.internal.utils :as u])
+            [temporal.internal.utils :as u]
+            [temporal.internal.exceptions :as e])
   (:import [java.time Duration]
            [io.temporal.client WorkflowClient WorkflowClientOptions WorkflowClientOptions$Builder WorkflowStub]
            [io.temporal.serviceclient WorkflowServiceStubs WorkflowServiceStubsOptions WorkflowServiceStubsOptions$Builder]
@@ -156,7 +157,8 @@ defworkflow once the workflow concludes.
 "
   [{:keys [^WorkflowStub stub] :as workflow}]
   (-> (.getResultAsync stub u/bytes-type)
-      (p/then nippy/thaw)))
+      (p/then nippy/thaw)
+      (p/catch e/slingshot? e/recast-stone)))
 
 (defn query
   "

src/temporal/exceptions.clj

@@ -0,0 +1,5 @@
+;; Copyright © 2024 Manetu, Inc.  All rights reserved
+
+(ns temporal.exceptions)
+
+(def flags {::non-retriable? "'true' indicates this exception is not going to be retried even if it is not included into retry policy doNotRetry list."})

src/temporal/internal/activity.clj

@@ -4,9 +4,11 @@
   (:require [clojure.core.protocols :as p]
             [clojure.datafy :as d]
             [clojure.core.async :refer [go <!] :as async]
+            [slingshot.slingshot :refer [try+ throw+]]
             [taoensso.timbre :as log]
             [taoensso.nippy :as nippy]
             [temporal.internal.utils :as u]
+            [temporal.internal.exceptions :as e]
             [temporal.common :as common])
   (:import [java.time Duration]
            [io.temporal.activity Activity ActivityInfo DynamicActivity ActivityCancellationType]
@@ -103,11 +105,14 @@
         f (u/find-dispatch-fn dispatch activity-type)
         a (u/->args args)]
     (log/trace activity-id "calling" f "with args:" a)
-    (try
-      (result-> activity-id (f ctx a))
-      (catch Exception e
-        (log/error e)
-        (throw e)))))
+    (try+
+     (result-> activity-id (f ctx a))
+     (catch Exception e
+       (log/error e)
+       (throw e))
+     (catch Object o
+       (log/error &throw-context)
+       (e/freeze &throw-context)))))
 
 (defn dispatcher [ctx dispatch]
   (reify DynamicActivity

src/temporal/internal/exceptions.clj

@@ -0,0 +1,25 @@
+;; Copyright © 2024 Manetu, Inc.  All rights reserved
+
+(ns temporal.internal.exceptions
+  (:require [slingshot.slingshot :refer [throw+]]
+            [temporal.exceptions :as e]
+            [temporal.internal.utils :as u])
+  (:import [io.temporal.failure ApplicationFailure]))
+
+(def exception-type (name ::slingshot))
+
+(defn slingshot? [ex]
+  (and (instance? ApplicationFailure (ex-cause ex))
+       (= exception-type (.getType (cast ApplicationFailure (ex-cause ex))))))
+
+(defn freeze
+  [{{:keys [::e/non-retriable?] :or {non-retriable? false}} :object :as context}]
+  (let [o (u/->objarray context)
+        t (if non-retriable?
+            (ApplicationFailure/newNonRetryableFailure nil exception-type o)
+            (ApplicationFailure/newFailure nil exception-type o))]
+    (throw t)))
+
+(defn recast-stone [ex]
+  (let [stone (->> ex ex-cause (cast ApplicationFailure) (.getDetails) u/->args :object)]
+    (throw+ stone)))                                        ;; FIXME: Does not preserve the original stack-trace

src/temporal/internal/workflow.clj

@@ -3,11 +3,13 @@
 (ns ^:no-doc temporal.internal.workflow
   (:require [clojure.core.protocols :as p]
             [clojure.datafy :as d]
+            [slingshot.slingshot :refer [try+]]
             [taoensso.timbre :as log]
             [taoensso.nippy :as nippy]
             [temporal.common :as common]
             [temporal.internal.utils :as u]
-            [temporal.internal.signals :as s])
+            [temporal.internal.signals :as s]
+            [temporal.internal.exceptions :as e])
   (:import [io.temporal.workflow Workflow WorkflowInfo]
            [io.temporal.client WorkflowOptions WorkflowOptions$Builder]))
 
@@ -51,17 +53,20 @@
 
 (defn execute
   [ctx dispatch args]
-  (try
-    (let [{:keys [workflow-type workflow-id]} (get-info)
-          d (u/find-dispatch dispatch workflow-type)
-          f (:fn d)
-          a (u/->args args)
-          _ (log/trace workflow-id "calling" f "with args:" a)
-          r (if (-> d :type (= :legacy))
-              (f ctx {:args a :signals (s/create-signal-chan)})
-              (f a))]
-      (log/trace workflow-id "result:" r)
-      (nippy/freeze r))
-    (catch Exception e
-      (log/error e)
-      (throw e))))
+  (try+
+   (let [{:keys [workflow-type workflow-id]} (get-info)
+         d (u/find-dispatch dispatch workflow-type)
+         f (:fn d)
+         a (u/->args args)
+         _ (log/trace workflow-id "calling" f "with args:" a)
+         r (if (-> d :type (= :legacy))
+             (f ctx {:args a :signals (s/create-signal-chan)})
+             (f a))]
+     (log/trace workflow-id "result:" r)
+     (nippy/freeze r))
+   (catch Exception e
+     (log/error e)
+     (throw e))
+   (catch Object o
+     (log/error &throw-context)
+     (e/freeze &throw-context))))

test/temporal/test/slingshot.clj

@@ -0,0 +1,46 @@
+;; Copyright © Manetu, Inc.  All rights reserved
+
+(ns temporal.test.slingshot
+  (:require [clojure.test :refer :all]
+            [taoensso.timbre :as log]
+            [slingshot.slingshot :refer [try+ throw+]]
+            [temporal.client.core :as c]
+            [temporal.workflow :refer [defworkflow]]
+            [temporal.activity :refer [defactivity] :as a]
+            [temporal.exceptions :as e]
+            [temporal.test.utils :as t]))
+
+(use-fixtures :once t/wrap-service)
+
+(defactivity slingshot-nonretriable-activity
+  [ctx {:keys [name] :as args}]
+  (log/info "slingshot-nonretriable-activity:" args)
+  (throw+ {:type ::test1 ::e/non-retriable? true}))
+
+(defactivity slingshot-retriable-activity
+  [ctx {:keys [name] :as args}]
+  (log/info "slingshot-retriable-activity:" args)
+  (throw+ {:type ::test2}))
+
+(defworkflow slingshot-workflow
+  [args]
+  (log/info "slingshot-workflow:" args)
+  (try+
+   @(a/invoke slingshot-nonretriable-activity args)
+   (catch [:type ::test1] _
+     (log/info "caught stone 1")
+     (try+
+      @(a/invoke slingshot-retriable-activity args)
+      (catch [:type ::test2] _
+        (log/info "caught stone 2")
+        (throw+ {:type ::test3}))))))
+
+(deftest the-test
+  (testing "Verifies that we can catch slingshot stones across activity/workflow boundaries"
+    (let [workflow (t/create-workflow slingshot-workflow)]
+      (c/start workflow {})
+      (try+
+       @(c/get-result workflow)
+       (throw (ex-info "should not get here" {}))
+       (catch [:type ::test3] _
+         (log/info "caught stone 3"))))))