Overhaul workflow monad to use native threading #99

iand675 · 2025-04-29T16:18:35Z

This is a relatively experimental change, but it's intended to enhance our ability to add new code in the future. The Coroutine monad, and the Workflow monad implementation are currently fairly hard to add new instances for, hard to write interceptors for, and lead to a lot of questions from engineers about the semantics of threads and exceptions in the Temporal model (as it currently stands). This PR is a step in the direction of using Haskell-native threads + thin wrappers around the IO monad to run Workflow code. This should provide us with the ability to write instances that we more-or-less can't write at the moment. The crucial one is that we can't reliably handle spans from OTel due to not having a good bracket-like function in the face of continuations.

iand675 · 2025-04-29T17:42:49Z

.envrc

pre-commit check fix.

iand675 · 2025-04-29T17:42:58Z

LICENSE

@@ -26,4 +26,3 @@ SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER
 CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY,
 OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE
 OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
-


pre-commit check fix.

iand675 · 2025-04-29T17:43:02Z

codec-encryption/LICENSE

@@ -26,4 +26,3 @@ SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER
 CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY,
 OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE
 OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
-


pre-commit check fix.

iand675 · 2025-04-29T17:43:06Z

codec-server/LICENSE

@@ -26,4 +26,3 @@ SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER
 CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY,
 OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE
 OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
-


pre-commit check fix.

iand675 · 2025-04-29T17:43:29Z

core/LICENSE

@@ -26,4 +26,3 @@ SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER
 CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY,
 OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE
 OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
-


pre-commit check fix.

iand675 · 2025-04-29T17:43:32Z

core/package.yaml

@@ -63,4 +63,3 @@ library:
  when:
    - condition: os(darwin)
      frameworks: [Security, CoreFoundation, SystemConfiguration]
-


pre-commit check fix.

iand675 · 2025-04-29T17:43:58Z

core/rust/src/client.rs

@@ -211,7 +211,6 @@ pub fn connect_client(
                    runtime,
                }),
                Err(e) => {
-                    eprintln!("Error: {:?}", e);


People occasionally see these and don't understand where they come from, and they should be reported via exceptions in Haskell land

iand675 · 2025-04-29T17:45:38Z

flake.nix

-        in
-        shells // { default = shells.ghc910; }
-      );
+  outputs = inputs @ {self, ...}: let


result of fixed pre-commit checks

iand675 · 2025-04-29T17:46:02Z

flake.nix

+            modules = [
+              ./nix/devenv/temporal-bridge.nix
+              ./nix/devenv/temporal-dev-server.nix
+              ./nix/devenv/repo-wide-checks.nix


this is the key change to this file, repo-wide-checks accidentally were left out during some refactor in the past

nice catch, ty

iand675 · 2025-04-29T17:49:25Z

sdk/src/Temporal/Client.hs

  case result of
-    Left err -> throwIO err
+    Left err -> throwM err


Here and elsewhere, I'm moving to throwM to just generalize towards any future monad changes we have to worry about in the future.

iand675 · 2025-05-09T19:12:39Z

.envrc

@@ -1,3 +1,5 @@
+#!/usr/bin/env bash


Precommit check fix

iand675 · 2025-05-09T19:15:13Z

sdk/src/Temporal/Interceptor.hs

@@ -144,3 +152,107 @@ instance Monoid ScheduleClientInterceptors where
    ScheduleClientInterceptors
      { scheduleWorkflowAction = \opts _ -> return opts
      }
+
+
+data ExecuteWorkflowInput = ExecuteWorkflowInput


Moved this stuff here from elsewhere to make the Temporal.Interceptor module the main interface for what goes in and out of Temporal actions

iand675 · 2025-05-12T16:19:18Z

sdk/package.yaml

  - Temporal.Workflow.Unsafe.Handle
-  - Temporal.Workflow.Internal.Instance


Moving away from the internal namespace usage since we can just not expose the modules

iand675 · 2025-05-12T16:20:23Z

sdk/src/Temporal/Workflow.hs

@@ -200,9 +198,6 @@ module Temporal.Workflow (
  ProvidedUpdate (..),
  KnownUpdate (..),

-  -- * Other utilities
-  unsafeAsyncEffectSink,


Not used, not really needed any more.

iand675 · 2025-05-12T16:22:54Z

sdk/src/Temporal/Workflow.hs

@@ -1644,12 +1472,12 @@ returning the original data structure with the arguments replaced with the resul
 This is actually a bit of a misnomer, since it's really 'traverseConcurrently', but this is copied to mimic the 'async' package's naming
 to slightly ease adoption.
 -}
-mapConcurrently :: Traversable t => (a -> Workflow b) -> t a -> Workflow (t b)
+mapConcurrently :: (Traversable t) => (a -> Workflow b) -> t a -> Workflow (t b)
 mapConcurrently = traverseConcurrently


I have a feeling that these folding / traversing versions of this would benefit from a specialized version that does a lot less blocking, but probably just need to come back to that later if it's an actual problem

iand675 · 2025-05-12T16:26:41Z

sdk/src/Temporal/Workflow/Instance.hs

+  , workflowRuntimeCancelRequested :: {-# UNPACK #-} !(IVar ())
+  , workflowRuntimeSequenceMaps :: {-# UNPACK #-} !SequenceMaps
+  , workflowRuntimeUnappliedJobs :: {-# UNPACK #-} !(TVar Int)
+  , signals :: {-# UNPACK #-} !SignalSupport


This is an approach I'm taking on trying to decompose portions of workflow functionality into smaller units that are a bit more independently maintainable. Hopefully this unlocks some ability to test subsystems better in future PRs.

iand675 · 2025-05-12T16:29:26Z

sdk/test/IntegrationSpec/HangingWorkflow.hs

@@ -81,22 +81,6 @@ defaultActivityOptions :: StartActivityOptions
 defaultActivityOptions = defaultStartActivityOptions $ StartToClose $ minutes 1


-opts :: StartChildWorkflowOptions
-opts =


Moved to not be a global value to fix shadowing warnings

iand675 · 2025-05-12T16:29:42Z

sdk/test/IntegrationSpec.hs

                  C.terminate h' C.TerminationOptions {terminationReason = "testing", terminationDetails = []}
              )
            `shouldThrow` \case
              RpcError {} -> True
-              _ -> False


Unnecessary since RPCError only has one constructor

iand675 · 2025-05-12T16:30:59Z

sdk/src/Temporal/Workflow/Worker.hs

-            _ <- attachContext activationCtxt
-            handleActivation activation
-          link activator
+          handleActivation activation


We don't need this extra bit of async calling, because creating the instances is cheap, and the RunningWorkflow struct is in charge of its own async-ness now.

jkachmar · 2025-05-12T16:37:33Z

sdk/src/Temporal/Client.hs

-          -- LMAO this is such a comical coercion
-          --
-          -- We just need to ignore payloads if the result type is unit to allow workflows to evolve
-          -- such that they can return something else in the future.
-          if typeRep h == typeOf ()
-            then pure $ unsafeCoerce ()
-            else case payloads of
-              (a : _) -> liftIO $ readResult a
-              _ -> error "Missing result payload"


jkachmar · 2025-05-12T16:39:11Z

sdk/src/Temporal/Client.hs

+        runReaderT
+          ( do
+              let r = RunId $ swer ^. WF.runId
+              getHandle k wfId' $
+                GetHandleOptions
+                  (Just r)
+                  (Just r)
+          )
+          c


the world if Haskell had a runReaderT variant whose arguments were flipped by default

jkachmar · 2025-05-12T17:09:13Z

sdk/src/Temporal/Client.hs

@@ -521,30 +512,28 @@ startFromPayloads
  -> StartWorkflowOptions
  -> V.Vector UnencodedPayload
  -> m (WorkflowHandle result)
-startFromPayloads k@(KnownWorkflow codec _) wfId opts payloads = do
+startFromPayloads k wfId opts payloads = do


not something to change in code, but would it be useful to have a glossary for stuff like "k is commonly used to refer to the continuation function" type stuff (for newcomers)?

jkachmar · 2025-05-13T15:09:47Z

sdk/src/Temporal/Workflow.hs

+-- startLocalActivity
+--   :: forall act
+--    . (RequireCallStack, ActivityRef act)
+--   => act
+--   -> StartLocalActivityOptions
+--   -> (ActivityArgs act :->: Workflow (Task (ActivityResult act)))
+-- startLocalActivity (activityRef -> KnownActivity codec n) opts = withWorkflowArgs @(ActivityArgs act) @(Task (ActivityResult act)) codec $ \typedPayloads -> Workflow do
+--   updateCallStack
+--   originalTime <- unWorkflow time
+--   inst <- ask
+--   runInIO <- askRunInIO
+--   let ps = fmap convertToProtoPayload typedPayloads
+--   s@(Sequence actSeq) <- nextActivitySequence
+--   resultSlot <- newIVar
+--   atomically $ modifyTVar'


also the module export up top

jkachmar · 2025-05-13T15:24:06Z

sdk/src/Temporal/Workflow.hs

+                  -- TODO
+                  -- & Command.headers .~ _
+              )


I know this is just a carryover, but what's the TODO here for? just setting the Command.headers to something if/when we want to support that?

jkachmar · 2025-05-13T15:29:23Z

sdk/src/Temporal/Workflow/Monad.hs

+import Data.Vector (Vector)
+import Data.Word
+import GHC.Conc (unsafeIOToSTM)
+import GHC.Exts (Any)


is this import actually used? I can't see Any referenced in the module at all

jkachmar · 2025-05-13T15:34:35Z

sdk/src/Temporal/Workflow/Monad.hs

+
+instance MonadReadStateVar Condition where
+  readStateVar ref = Condition do
+    tid <- unsafeIOToSTM myThreadId


might be worth documenting this the same way that Rust usage of unsafe has a convention of documenting the invariants (e.g. IO and STM are effectively identical with respect to RTS thread ID)

jkachmar · 2025-05-13T15:38:38Z

sdk/src/Temporal/Workflow/Monad.hs

+      When we get an activation, we apply the jobs, then
+
+      -}


was there more here, or was this meant to be covered in other docs?

jkachmar · 2025-05-13T15:40:26Z

sdk/src/Temporal/Workflow/Monad.hs

+      eval = mask $ \restore -> do
+        tid <- myThreadId
+        let tidText = Text.pack (show tid)
+            threadWrapper _restore = id


is the intent here the same as your previous threadWrapper? we could use this as a point to support propagating trace context?

jkachmar · 2025-05-13T15:49:59Z

sdk/src/Temporal/Workflow/Monad.hs

+        handleableSignals <- stateTVar runtime.signals.bufferedSignals $ \signals ->
+          span (\signal -> (Just (signal ^. Activation.signalName) `HashMap.member` signalHandlers) || (Nothing `HashMap.member` signalHandlers)) signals


TIL stateTVar + span as a pattern for handling state transition

jkachmar · 2025-05-13T15:51:28Z

sdk/src/Temporal/Workflow/Monad.hs

+        -- TODO, do we need to count unapplied jobs here. I think so.
+        modifyTVar' runtime.workflowRuntimeUnappliedJobs (+ length goodQueries)


not 100% sure what this comment is referring to: do you want to count unapplied jobs in one of the query counter variables here?

jkachmar · 2025-05-13T15:58:46Z

sdk/src/Temporal/Workflow/Monad.hs

+          ( do
+              liftIO $ join $ atomically $ applyActivations mainThread
+              atomically do
+                waitAllJobsHandled runtime
+                waitAllThreadsBlocked runtime
+              untilM_ do
+                (handleCount, m) <- atomically $ applyQueuedSignals mainThread
+                liftIO m
+                atomically do
+                  waitAllJobsHandled runtime
+                  waitAllThreadsBlocked runtime
+                pure (handleCount == 0)
+              -- By the time we've applied all the signals that we can,
+              -- all of the queries we can possibly handle must have been
+              -- defined.
+              liftIO . snd =<< atomically (applyQueuedQueries mainThread)
+              restore $ atomically do
+                waitAllJobsHandled runtime
+                waitAllThreadsBlocked runtime
+            )


I think I've mentioned this elsewhere, but it may be worth reiterating to the reader (either here or elsewhere in the documentation) that atomically is being called explicitly to ensure a deliberate transactional sequence.

I think a lot of people can end up with an impression that atomically is just the STM a -> IO a function — and most of the time that's no big deal — but in this case if someone were to merge some of these blocks then you could end up with re-ordering that breaks the semantics we want to enforce.

jkachmar · 2025-05-13T16:00:30Z

sdk/src/Temporal/Workflow/Monad.hs

+  case res of
+    Left err -> do
+      throwIO err
+    Right () -> do
+      pure ()


either throwM pure res? or is it worth deliberately case-ing & using throwIO

jkachmar · 2025-05-13T16:02:18Z

sdk/src/Temporal/Workflow/Monad.hs

+      -- TODO, if there's a huge amount of signals, this is algorithmically inefficient.
+      modifyTVar' runtime.signals.bufferedSignals (++ [sig])


what would the alternative be here, Vector or something not-list-like?

The alternative would be to use a difference list, such as this, although they're not hard to implement by hand: https://hackage.haskell.org/package/dlist

ah, so you mean the specific case of like... a lot of signals getting queued up before any of them get handled yea?

Right, it would be an N^2 evaluation of the list. Hard to think that gRPC size limits would let this get too out of control though

jkachmar · 2025-05-13T16:07:04Z

sdk/src/Temporal/Workflow/Monad.hs

+                    & Command.completed .~ convertToProtoPayload payload
+                 )
+
+  -- (validator, updateAction) <- do


sdk/src/Temporal/Workflow/Monad.hs

jkachmar · 2025-05-13T16:13:15Z

sdk/test/Temporal/Workflow/IVarSpec.hs

+      }
+
+
+-- -- | Creates a new thread in the test runtime


Suggested change

-- -- | Creates a new thread in the test runtime

-- | Creates a new thread in the test runtime

jkachmar · 2025-05-13T16:13:23Z

sdk/test/Temporal/Workflow/IVarSpec.hs

+      `finally` atomically (modifyTVar' runtime.testThreadManager $ HashMap.delete tid)
+
+
+-- -- | Check if a thread is blocked in the runtime


Suggested change

-- -- | Check if a thread is blocked in the runtime

-- | Check if a thread is blocked in the runtime

jkachmar

huge quality of life improvement; I wish we had some benchmarks to do a before/after comparison because I'd be pretty shocked if this didn't measurably improve overhead.

iand675 · 2025-05-13T16:22:09Z

@jkachmar @andrewsmith Informally benchmarking this, they're effectively identical performance wise.

Both test suites run in 26.5 seconds +/- .2 seconds

jkachmar · 2025-05-13T18:53:22Z

sdk/src/Temporal/Workflow.hs

+        let anyHandle :: forall h. ChildWorkflowHandle h -> ChildWorkflowHandle Any
+            anyHandle = unsafeCoerce
+        atomically $
+          modifyTVar'
+            runtime.workflowRuntimeSequenceMaps.childWorkflows
+            (HashMap.insert s $ anyHandle wfHandle)


okay so I understand what unsafeCoerce is doing here, but I don't really follow what opposite side of this looks like.

you stash the ChildWorkflowHandle Any in the map, which gets used in handleResolveChildWorkflowExecution but how is the result type information handled here?

iand675 commented Apr 29, 2025

View reviewed changes

.envrc

Copy link

Contributor Author

iand675 Apr 29, 2025

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

pre-commit check fix.

iand675 commented Apr 29, 2025

View reviewed changes

iand675 force-pushed the haxl-be-gone branch from 60ad71d to 8cc4804 Compare April 29, 2025 20:05

johnmastro mentioned this pull request Apr 30, 2025

add support for update #75

Merged

iand675 force-pushed the haxl-be-gone branch from 11b2955 to eef11be Compare May 5, 2025 14:26

iand675 added 15 commits May 8, 2025 14:09

some progress

eb03252

fix child workflow handles

f697446

Bug fixes

efa6945

fix signals

995f9ef

Fix everything except for weird child workflow deadlock

2603931

hanging workflow shutdown is weird

46189a8

get rid of all the rogue / debug logging

4144ba6

Add docs folder explaining how things fit together, mostly

54eaf43

Fix log levels in rust sdk

b6734a8

Fix all the precommit hooks

bbfe108

Tidy tests a bit

16d4d5f

get rid of resourcet

b528aa6

we have passing tests

663fa24

Tests fixed?

1d7e11a

Rebase fixups

51a3f5b

iand675 force-pushed the haxl-be-gone branch from 4df2bd5 to 51a3f5b Compare May 8, 2025 13:12

iand675 added 2 commits May 9, 2025 20:43

Get update tests passing

57b937f

ghc 96 happy now?

d6763ba

iand675 commented May 9, 2025

View reviewed changes

.envrc

@@ -1,3 +1,5 @@

#!/usr/bin/env bash

Copy link

Contributor Author

iand675 May 9, 2025

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Precommit check fix

iand675 commented May 9, 2025

View reviewed changes

iand675 added 3 commits May 9, 2025 21:20

Diff minimization

b3b25b3

Docs adjustments

19f8464

Get rid of coroutine module

156cb80

iand675 commented May 12, 2025

View reviewed changes

Delete sdk/test/IntegrationSpec/Replay.hs

b74c336

iand675 commented May 12, 2025

View reviewed changes

Remove some dead code

31e3d67

iand675 commented May 12, 2025

View reviewed changes

iand675 requested a review from jkachmar May 12, 2025 16:31

iand675 changed the title ~~Haxl be gone~~ Overhaul workflow monad to use native threading May 12, 2025

jkachmar reviewed May 12, 2025

View reviewed changes

jkachmar reviewed May 13, 2025

View reviewed changes

sdk/src/Temporal/Workflow/Monad.hs Show resolved Hide resolved

jkachmar reviewed May 13, 2025

View reviewed changes

jkachmar approved these changes May 13, 2025

View reviewed changes

jkachmar reviewed May 13, 2025

View reviewed changes

		- Temporal.Workflow.Unsafe.Handle
		- Temporal.Workflow.Internal.Instance

		handleableSignals <- stateTVar runtime.signals.bufferedSignals $ \signals ->
		span (\signal -> (Just (signal ^. Activation.signalName) `HashMap.member` signalHandlers) \|\| (Nothing `HashMap.member` signalHandlers)) signals

		-- TODO, do we need to count unapplied jobs here. I think so.
		modifyTVar' runtime.workflowRuntimeUnappliedJobs (+ length goodQueries)

		-- TODO, if there's a huge amount of signals, this is algorithmically inefficient.
		modifyTVar' runtime.signals.bufferedSignals (++ [sig])

	-- -- \| Creates a new thread in the test runtime
	-- \| Creates a new thread in the test runtime

		`finally` atomically (modifyTVar' runtime.testThreadManager $ HashMap.delete tid)


		-- -- \| Check if a thread is blocked in the runtime

	-- -- \| Check if a thread is blocked in the runtime
	-- \| Check if a thread is blocked in the runtime

Overhaul workflow monad to use native threading #99

Are you sure you want to change the base?

Overhaul workflow monad to use native threading #99

Uh oh!

Conversation

iand675 commented Apr 29, 2025 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

iand675 Apr 29, 2025 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

iand675 Apr 29, 2025 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

iand675 commented Apr 29, 2025 •

edited

Loading

iand675 Apr 29, 2025 •

edited

Loading

iand675 Apr 29, 2025 •

edited

Loading

jkachmar May 13, 2025 •

edited

Loading

iand675 commented May 13, 2025 •

edited

Loading