Signal race condition

src/Workflow.php

@@ -128,14 +128,20 @@ public function handle(): void
             ->wherePivot('parent_index', '!=', StoredWorkflow::ACTIVE_WORKFLOW_INDEX)
             ->first();
 
-        $log = $this->storedWorkflow->logs()
-            ->whereIndex($this->index)
+        $logs = $this->storedWorkflow->logs()
+            ->whereIn('index', [$this->index, $this->index + 1])
+            ->get();
+
+        $log = $logs->where('index', $this->index)
+            ->first();
+
+        $nextLog = $logs->where('index', $this->index + 1)
             ->first();
 
         $this->storedWorkflow
             ->signals()
-            ->when($log, static function ($query, $log): void {
-                $query->where('created_at', '<=', $log->created_at->format('Y-m-d H:i:s.u'));
+            ->when($nextLog, static function ($query, $nextLog): void {
+                $query->where('created_at', '<=', $nextLog->created_at->format('Y-m-d H:i:s.u'));
             })
             ->each(function ($signal): void {
                 $this->{$signal->method}(...Serializer::unserialize($signal->arguments));
@@ -163,8 +169,14 @@ public function handle(): void
         while ($this->coroutine->valid()) {
             $this->index = WorkflowStub::getContext()->index;
 
-            $nextLog = $this->storedWorkflow->logs()
-                ->whereIndex($this->index)
+            $logs = $this->storedWorkflow->logs()
+                ->whereIn('index', [$this->index, $this->index + 1])
+                ->get();
+
+            $log = $logs->where('index', $this->index)
+                ->first();
+
+            $nextLog = $logs->where('index', $this->index + 1)
                 ->first();
 
             if ($log) {
@@ -179,8 +191,6 @@ public function handle(): void
                     });
             }
 
-            $log = $nextLog;
-
             $this->now = $log ? $log->now : Carbon::now();
 
             WorkflowStub::setContext([

tests/Feature/WorkflowTest.php

@@ -6,6 +6,8 @@
 
 use Tests\Fixtures\TestActivity;
 use Tests\Fixtures\TestOtherActivity;
+use Tests\Fixtures\TestSignalExceptionWorkflow;
+use Tests\Fixtures\TestSignalExceptionWorkflowLeader;
 use Tests\Fixtures\TestWorkflow;
 use Tests\TestCase;
 use Workflow\Signal;
@@ -55,4 +57,76 @@ public function testCompletedDelay(): void
             ->values()
             ->toArray());
     }
+
+    public function testTestSignalExceptionWorkflowEarly(): void
+    {
+        $workflow = WorkflowStub::make(TestSignalExceptionWorkflow::class);
+
+        $workflow->start([
+            'test' => 'data',
+        ]);
+
+        sleep(1);
+
+        $workflow->shouldRetry();
+
+        while ($workflow->running());
+
+        $this->assertSame(WorkflowCompletedStatus::class, $workflow->status());
+        $this->assertTrue($workflow->output());
+    }
+
+    public function testTestSignalExceptionWorkflowLate(): void
+    {
+        $workflow = WorkflowStub::make(TestSignalExceptionWorkflow::class);
+
+        $workflow->start([
+            'test' => 'data',
+        ]);
+
+        sleep(3);
+
+        $workflow->shouldRetry();
+
+        while ($workflow->running());
+
+        $this->assertSame(WorkflowCompletedStatus::class, $workflow->status());
+        $this->assertTrue($workflow->output());
+    }
+
+    public function testTestSignalExceptionWorkflowLeaderEarly(): void
+    {
+        $workflow = WorkflowStub::make(TestSignalExceptionWorkflowLeader::class);
+
+        $workflow->start([
+            'test' => 'data',
+        ]);
+
+        sleep(1);
+
+        $workflow->shouldRetry();
+
+        while ($workflow->running());
+
+        $this->assertSame(WorkflowCompletedStatus::class, $workflow->status());
+        $this->assertTrue($workflow->output());
+    }
+
+    public function testTestSignalExceptionWorkflowLeaderLate(): void
+    {
+        $workflow = WorkflowStub::make(TestSignalExceptionWorkflowLeader::class);
+
+        $workflow->start([
+            'test' => 'data',
+        ]);
+
+        sleep(3);
+
+        $workflow->shouldRetry();
+
+        while ($workflow->running());
+
+        $this->assertSame(WorkflowCompletedStatus::class, $workflow->status());
+        $this->assertTrue($workflow->output());
+    }
 }

tests/Fixtures/TestParentWorkflow.php

@@ -6,10 +6,17 @@
 
 use Workflow\ActivityStub;
 use Workflow\ChildWorkflowStub;
+use Workflow\SignalMethod;
 use Workflow\Workflow;
 
 class TestParentWorkflow extends Workflow
 {
+    #[SignalMethod]
+    public function ping(): void
+    {
+        // Do nothing
+    }
+
     public function execute()
     {
         $otherResult = yield ChildWorkflowStub::make(TestChildWorkflow::class);

tests/Fixtures/TestSignalExceptionWorkflow.php

@@ -0,0 +1,38 @@
+<?php
+
+declare(strict_types=1);
+
+namespace Tests\Fixtures;
+
+use Throwable;
+use Workflow\ActivityStub;
+use Workflow\SignalMethod;
+use Workflow\Workflow;
+use Workflow\WorkflowStub;
+
+class TestSignalExceptionWorkflow extends Workflow
+{
+    protected bool $shouldRetry = false;
+
+    #[SignalMethod]
+    public function shouldRetry(): void
+    {
+        $this->shouldRetry = true;
+    }
+
+    public function execute(array $data = [])
+    {
+        $shouldThrow = true;
+        while (true) {
+            try {
+                yield ActivityStub::make(TestActivity::class);
+                yield ActivityStub::make(TestSingleTryExceptionActivity::class, $shouldThrow);
+                return true;
+            } catch (Throwable) {
+                yield WorkflowStub::await(fn () => $this->shouldRetry);
+                $this->shouldRetry = false;
+                $shouldThrow = false;
+            }
+        }
+    }
+}

tests/Fixtures/TestSignalExceptionWorkflowLeader.php

@@ -0,0 +1,37 @@
+<?php
+
+declare(strict_types=1);
+
+namespace Tests\Fixtures;
+
+use Throwable;
+use Workflow\ActivityStub;
+use Workflow\SignalMethod;
+use Workflow\Workflow;
+use Workflow\WorkflowStub;
+
+class TestSignalExceptionWorkflowLeader extends Workflow
+{
+    protected bool $shouldRetry = false;
+
+    #[SignalMethod]
+    public function shouldRetry(): void
+    {
+        $this->shouldRetry = true;
+    }
+
+    public function execute(array $data = [])
+    {
+        $shouldThrow = true;
+        while (true) {
+            try {
+                yield ActivityStub::make(TestSingleTryExceptionActivity::class, $shouldThrow);
+                return true;
+            } catch (Throwable) {
+                yield WorkflowStub::await(fn () => $this->shouldRetry);
+                $this->shouldRetry = false;
+                $shouldThrow = false;
+            }
+        }
+    }
+}

tests/Fixtures/TestSingleTryExceptionActivity.php

@@ -0,0 +1,22 @@
+<?php
+
+declare(strict_types=1);
+
+namespace Tests\Fixtures;
+
+use Exception;
+use Workflow\Activity;
+
+final class TestSingleTryExceptionActivity extends Activity
+{
+    public $tries = 1;
+
+    public function execute($shouldThrow)
+    {
+        if ($shouldThrow) {
+            throw new Exception('failed');
+        }
+
+        return 'activity';
+    }
+}

tests/Unit/WorkflowTest.php

@@ -255,6 +255,14 @@ public function testParentPending(): void
                 'result' => Serializer::serialize('activity'),
             ]);
 
+        $storedParentWorkflow->signals()
+            ->create([
+                'method' => 'ping',
+                'arguments' => Serializer::serialize([]),
+                'created_at' => now()
+                    ->addSeconds(1),
+            ]);
+
         $childWorkflow = WorkflowStub::load(WorkflowStub::make(TestChildWorkflow::class)->id());
 
         $storedChildWorkflow = StoredWorkflow::findOrFail($childWorkflow->id());