testing: add with_deferred_starts to model async start_workflow

lib/temporal/testing.rb

@@ -1,6 +1,7 @@
 require 'temporal/testing/temporal_override'
 require 'temporal/testing/workflow_override'
 require 'temporal/testing/scheduled_workflows'
+require 'temporal/testing/deferred_starts'
 
 module Temporal
   module Testing
@@ -24,6 +25,40 @@ def local?
         mode == LOCAL_MODE
       end
 
+      # True when start_workflow calls should be deferred rather than run inline.
+      # Only meaningful in local mode.
+      def defer_starts?
+        local? && @defer_starts == true
+      end
+
+      # Within the block, fire-and-forget start_workflow calls are deferred instead
+      # of run inline; the queued workflows run when the block exits normally (after
+      # the caller's stack -- and any locks it held -- have unwound). This models the
+      # async ordering of a real Temporal server in local mode. See DeferredStarts.
+      #
+      # Requires local! mode and cannot be nested.
+      def with_deferred_starts
+        raise 'Temporal::Testing.with_deferred_starts requires Temporal::Testing.local!' unless local?
+
+        # The queue is a single flat array, so a nested block would drain and clear
+        # the outer block's queued starts. Forbid nesting rather than drop workflows
+        # silently.
+        raise 'Temporal::Testing.with_deferred_starts cannot be nested' if defer_starts?
+
+        @defer_starts = true
+        begin
+          result = yield
+          # Clear the flag before draining so the queued workflows run inline (and
+          # could, in turn, start and await their own children).
+          @defer_starts = false
+          DeferredStarts.execute_all
+          result
+        ensure
+          @defer_starts = false
+          DeferredStarts.clear_all
+        end
+      end
+
       private
 
       attr_reader :mode

lib/temporal/testing/deferred_starts.rb

@@ -0,0 +1,57 @@
+module Temporal
+  module Testing
+    # In local mode, Temporal.start_workflow normally runs the workflow inline and
+    # synchronously, which collapses the start/run distinction that exists against a
+    # real Temporal server (start_workflow enqueues; the workflow runs later, out of
+    # process). That inline execution is wrong for a caller that holds a resource --
+    # e.g. a lock -- across the start_workflow call and expects the workflow to run
+    # only after it has returned and released that resource.
+    #
+    # Inside a Temporal::Testing.with_deferred_starts block, start_workflow instead
+    # defers execution: the workflow is queued and run when the block exits, after the
+    # caller's stack has unwound. This lets a test model the real async ordering.
+    #
+    # Awaited workflows cannot be deferred: await_workflow_result needs the result
+    # synchronously, and local mode does not capture a workflow's return value. Only
+    # fire-and-forget start_workflow calls should run inside the block.
+    module DeferredStarts
+      def self.execute_all
+        Private::Store.execute_all
+      end
+
+      def self.clear_all
+        Private::Store.clear_all
+      end
+
+      module Private
+        module Store
+          class << self
+            def add(executor_lambda:)
+              executions << executor_lambda
+            end
+
+            def execute_all
+              # Drain FIFO. The defer flag is cleared before we drain, so any workflow
+              # started during a drain runs inline and the queue does not grow here.
+              #
+              # Each runs independently, mirroring separate executions on a real worker:
+              # WorkflowExecution#run records a failed run as FAILED rather than raising,
+              # so one workflow failing does not abort the drain of the others.
+              executions.shift.call until executions.empty?
+            end
+
+            def clear_all
+              @executions = []
+            end
+
+            private
+
+            def executions
+              @executions ||= []
+            end
+          end
+        end
+      end
+    end
+  end
+end

lib/temporal/testing/temporal_override.rb

@@ -115,8 +115,18 @@ def start_locally(workflow, schedule, *input, **args)
         )
 
         if schedule.nil?
-          execution.run do
-            workflow.execute_in_context(context, input)
+          executor = lambda do
+            execution.run do
+              workflow.execute_in_context(context, input)
+            end
+          end
+
+          if Temporal::Testing.defer_starts?
+            # Defer execution until the surrounding with_deferred_starts block exits,
+            # modeling the async start/run split of a real Temporal server.
+            Temporal::Testing::DeferredStarts::Private::Store.add(executor_lambda: executor)
+          else
+            executor.call
           end
         else
           # Defer execution; in testing mode, it'll need to be invoked manually.

spec/unit/lib/temporal/testing/temporal_override_spec.rb

@@ -101,6 +101,81 @@ def execute
       end
     end
 
+    describe 'Temporal::Testing.with_deferred_starts' do
+      let(:workflow) { TestTemporalOverrideWorkflow.new(nil) }
+
+      before do
+        allow(TestTemporalOverrideWorkflow).to receive(:new).and_return(workflow)
+        allow(workflow).to receive(:execute)
+      end
+
+      it 'defers start_workflow execution until the block exits' do
+        Temporal::Testing.with_deferred_starts do
+          client.start_workflow(TestTemporalOverrideWorkflow)
+          expect(workflow).not_to have_received(:execute)
+        end
+
+        expect(workflow).to have_received(:execute)
+      end
+
+      it 'runs start_workflow inline outside of the block' do
+        client.start_workflow(TestTemporalOverrideWorkflow)
+
+        expect(workflow).to have_received(:execute)
+      end
+
+      it 'does not run deferred workflows when the block raises' do
+        expect do
+          Temporal::Testing.with_deferred_starts do
+            client.start_workflow(TestTemporalOverrideWorkflow)
+            raise 'boom'
+          end
+        end.to raise_error('boom')
+
+        expect(workflow).not_to have_received(:execute)
+      end
+
+      it 'restores inline execution after the block' do
+        Temporal::Testing.with_deferred_starts {}
+
+        client.start_workflow(TestTemporalOverrideWorkflow)
+        expect(workflow).to have_received(:execute)
+      end
+
+      it 'runs every deferred workflow even if an earlier one fails' do
+        call_count = 0
+        allow(workflow).to receive(:execute) do
+          call_count += 1
+          raise 'boom' if call_count == 1
+        end
+
+        # The first deferred workflow is recorded as FAILED, not raised, so the drain
+        # continues to the second -- executions are independent, as on a real worker.
+        Temporal::Testing.with_deferred_starts do
+          client.start_workflow(TestTemporalOverrideWorkflow)
+          client.start_workflow(TestTemporalOverrideWorkflow)
+        end
+
+        expect(call_count).to eq(2)
+      end
+
+      it 'raises when nested' do
+        expect do
+          Temporal::Testing.with_deferred_starts do
+            Temporal::Testing.with_deferred_starts {}
+          end
+        end.to raise_error(/cannot be nested/)
+      end
+
+      it 'raises when not in local mode' do
+        Temporal::Testing.disabled! do
+          expect do
+            Temporal::Testing.with_deferred_starts {}
+          end.to raise_error(/requires Temporal::Testing.local!/)
+        end
+      end
+    end
+
     describe 'Workflow.execute_locally' do
       it 'executes the workflow' do
         workflow = TestTemporalOverrideWorkflow.new(nil)