diff --git a/lib/temporal/testing.rb b/lib/temporal/testing.rb index 655cbfc1..c92300f0 100644 --- a/lib/temporal/testing.rb +++ b/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 diff --git a/lib/temporal/testing/deferred_starts.rb b/lib/temporal/testing/deferred_starts.rb new file mode 100644 index 00000000..d80adc5c --- /dev/null +++ b/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 diff --git a/lib/temporal/testing/temporal_override.rb b/lib/temporal/testing/temporal_override.rb index c67515e4..4a9afb76 100644 --- a/lib/temporal/testing/temporal_override.rb +++ b/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. diff --git a/spec/unit/lib/temporal/testing/temporal_override_spec.rb b/spec/unit/lib/temporal/testing/temporal_override_spec.rb index 4d8bef64..e0483a08 100644 --- a/spec/unit/lib/temporal/testing/temporal_override_spec.rb +++ b/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)