Skip stale foreign exception logs during replay

Author: durable-workflow-ops

src/ActivityStub.php

@@ -66,6 +66,9 @@ public static function make($activity, ...$arguments): PromiseInterface
             ++$context->index;
             WorkflowStub::setContext($context);
             $result = Serializer::unserialize($log->result);
+            if ($log->class === Exception::class && self::isForeignExceptionResult($result, $activity)) {
+                return self::make($activity, ...$arguments);
+            }
             if (
                 is_array($result) &&
                 array_key_exists('class', $result) &&
@@ -98,4 +101,12 @@ public static function make($activity, ...$arguments): PromiseInterface
         WorkflowStub::setContext($context);
         return (new Deferred())->promise();
     }
+
+    private static function isForeignExceptionResult(mixed $result, string $activity): bool
+    {
+        return is_array($result)
+            && isset($result['sourceClass'])
+            && is_string($result['sourceClass'])
+            && $result['sourceClass'] !== $activity;
+    }
 }

src/ChildWorkflowStub.php

@@ -60,6 +60,9 @@ public static function make($workflow, ...$arguments): PromiseInterface
             ++$context->index;
             WorkflowStub::setContext($context);
             $result = Serializer::unserialize($log->result);
+            if ($log->class === Exception::class && self::isForeignExceptionResult($result, $workflow)) {
+                return self::make($workflow, ...$arguments);
+            }
             if (
                 is_array($result)
                 && array_key_exists('class', $result)
@@ -114,4 +117,12 @@ public static function make($workflow, ...$arguments): PromiseInterface
         WorkflowStub::setContext($context);
         return (new Deferred())->promise();
     }
+
+    private static function isForeignExceptionResult(mixed $result, string $workflow): bool
+    {
+        return is_array($result)
+            && isset($result['sourceClass'])
+            && is_string($result['sourceClass'])
+            && $result['sourceClass'] !== $workflow;
+    }
 }

src/Exception.php

@@ -10,6 +10,7 @@
 use Illuminate\Foundation\Bus\Dispatchable;
 use Illuminate\Queue\InteractsWithQueue;
 use Illuminate\Queue\SerializesModels;
+use Illuminate\Support\Facades\Cache;
 use Throwable;
 use Workflow\Exceptions\TransitionNotFound;
 use Workflow\Middleware\WithoutOverlappingMiddleware;
@@ -51,18 +52,30 @@ public function __construct(
 
     public function handle()
     {
-        $workflow = $this->storedWorkflow->toWorkflow();
+        $lock = Cache::lock('laravel-workflow-exception:' . $this->storedWorkflow->id, 15);
+
+        if (! $lock->get()) {
+            $this->release();
+
+            return;
+        }
 
         try {
-            if ($this->storedWorkflow->hasLogByIndex($this->index)) {
-                $workflow->resume();
-            } elseif ($this->shouldPersistAfterProbeReplay()) {
-                $workflow->next($this->index, $this->now, self::class, $this->exception);
-            }
-        } catch (TransitionNotFound) {
-            if ($workflow->running()) {
-                $this->release();
+            $workflow = $this->storedWorkflow->toWorkflow();
+
+            try {
+                if ($this->storedWorkflow->hasLogByIndex($this->index)) {
+                    $workflow->resume();
+                } elseif ($this->shouldPersistAfterProbeReplay()) {
+                    $workflow->next($this->index, $this->now, self::class, $this->exceptionPayload());
+                }
+            } catch (TransitionNotFound) {
+                if ($workflow->running()) {
+                    $this->release();
+                }
             }
+        } finally {
+            $lock->release();
         }
     }
 
@@ -132,11 +145,22 @@ private function createTentativeWorkflowState(): StoredWorkflow
             'index' => $this->index,
             'now' => $this->now,
             'class' => self::class,
-            'result' => Serializer::serialize($this->exception),
+            'result' => Serializer::serialize($this->exceptionPayload()),
         ]);
 
         $storedWorkflowClass = $this->storedWorkflow::class;
 
         return $storedWorkflowClass::query()->findOrFail($this->storedWorkflow->id);
     }
+
+    private function exceptionPayload()
+    {
+        if (! is_array($this->exception) || $this->sourceClass === null) {
+            return $this->exception;
+        }
+
+        return array_merge($this->exception, [
+            'sourceClass' => $this->sourceClass,
+        ]);
+    }
 }

tests/Unit/ActivityStubTest.php

@@ -7,9 +7,11 @@
 use Exception;
 use RuntimeException;
 use Tests\Fixtures\TestActivity;
+use Tests\Fixtures\TestOtherActivity;
 use Tests\Fixtures\TestWorkflow;
 use Tests\TestCase;
 use Workflow\ActivityStub;
+use Workflow\Exception as WorkflowException;
 use Workflow\Models\StoredWorkflow;
 use Workflow\Serializers\Serializer;
 use Workflow\States\WorkflowPendingStatus;
@@ -123,6 +125,43 @@ public function testLoadsStoredExceptionWithNonStandardConstructor(): void
             });
     }
 
+    public function testSkipsStoredExceptionForDifferentSourceClass(): void
+    {
+        $workflow = WorkflowStub::load(WorkflowStub::make(TestWorkflow::class)->id());
+        $storedWorkflow = StoredWorkflow::findOrFail($workflow->id());
+        $storedWorkflow->update([
+            'arguments' => Serializer::serialize([]),
+            'status' => WorkflowPendingStatus::$name,
+        ]);
+        $storedWorkflow->logs()
+            ->create([
+                'index' => 0,
+                'now' => WorkflowStub::now(),
+                'class' => WorkflowException::class,
+                'result' => Serializer::serialize([
+                    'class' => Exception::class,
+                    'message' => 'foreign',
+                    'code' => 0,
+                    'sourceClass' => TestOtherActivity::class,
+                ]),
+            ]);
+        $storedWorkflow->logs()
+            ->create([
+                'index' => 1,
+                'now' => WorkflowStub::now(),
+                'class' => TestActivity::class,
+                'result' => Serializer::serialize('test'),
+            ]);
+
+        ActivityStub::make(TestActivity::class)
+            ->then(static function ($value) use (&$result) {
+                $result = $value;
+            });
+
+        $this->assertSame('test', $result);
+        $this->assertSame(2, WorkflowStub::getContext()->index);
+    }
+
     public function testAll(): void
     {
         $workflow = WorkflowStub::load(WorkflowStub::make(TestWorkflow::class)->id());

tests/Unit/ChildWorkflowStubTest.php

@@ -4,11 +4,14 @@
 
 namespace Tests\Unit;
 
+use Exception;
 use Mockery;
 use Tests\Fixtures\TestChildWorkflow;
+use Tests\Fixtures\TestExceptionWorkflow;
 use Tests\Fixtures\TestParentWorkflow;
 use Tests\TestCase;
 use Workflow\ChildWorkflowStub;
+use Workflow\Exception as WorkflowException;
 use Workflow\Models\StoredWorkflow;
 use Workflow\Serializers\Serializer;
 use Workflow\States\WorkflowPendingStatus;
@@ -91,6 +94,43 @@ public function testLoadsChildWorkflow(): void
         $this->assertNull($result);
     }
 
+    public function testSkipsStoredExceptionForDifferentSourceClass(): void
+    {
+        $workflow = WorkflowStub::load(WorkflowStub::make(TestParentWorkflow::class)->id());
+        $storedWorkflow = StoredWorkflow::findOrFail($workflow->id());
+        $storedWorkflow->update([
+            'arguments' => Serializer::serialize([]),
+            'status' => WorkflowPendingStatus::$name,
+        ]);
+        $storedWorkflow->logs()
+            ->create([
+                'index' => 0,
+                'now' => WorkflowStub::now(),
+                'class' => WorkflowException::class,
+                'result' => Serializer::serialize([
+                    'class' => Exception::class,
+                    'message' => 'foreign child',
+                    'code' => 0,
+                    'sourceClass' => TestExceptionWorkflow::class,
+                ]),
+            ]);
+        $storedWorkflow->logs()
+            ->create([
+                'index' => 1,
+                'now' => WorkflowStub::now(),
+                'class' => TestChildWorkflow::class,
+                'result' => Serializer::serialize('test'),
+            ]);
+
+        ChildWorkflowStub::make(TestChildWorkflow::class)
+            ->then(static function ($value) use (&$result) {
+                $result = $value;
+            });
+
+        $this->assertSame('test', $result);
+        $this->assertSame(2, WorkflowStub::getContext()->index);
+    }
+
     public function testDoesNotResumeRunningStartedChildWorkflow(): void
     {
         $childWorkflow = Mockery::mock();