Child exceptions

src/ChildWorkflowStub.php

@@ -8,8 +8,11 @@
 use React\Promise\Deferred;
 use React\Promise\PromiseInterface;
 use function React\Promise\resolve;
+use RuntimeException;
+use Throwable;
 use Workflow\Exceptions\TransitionNotFound;
 use Workflow\Serializers\Serializer;
+use Workflow\States\WorkflowFailedStatus;
 
 final class ChildWorkflowStub
 {
@@ -44,9 +47,41 @@ public static function make($workflow, ...$arguments): PromiseInterface
         }
 
         if ($log) {
-            ++$context->index;
-            WorkflowStub::setContext($context);
-            return resolve(Serializer::unserialize($log->result));
+            $result = Serializer::unserialize($log->result);
+            if (
+                is_array($result)
+                && array_key_exists('class', $result)
+                && is_subclass_of($result['class'], Throwable::class)
+            ) {
+                if (! $context->replaying) {
+                    $storedChildWorkflow = $context->storedWorkflow->children()
+                        ->wherePivot('parent_index', $context->index)
+                        ->first();
+                    if ($storedChildWorkflow && $storedChildWorkflow->status::class !== WorkflowFailedStatus::class) {
+                        $log->delete();
+                        $log = null;
+                    }
+                }
+
+                if ($log) {
+                    ++$context->index;
+                    WorkflowStub::setContext($context);
+                    try {
+                        $throwable = new $result['class']($result['message'] ?? '', (int) ($result['code'] ?? 0));
+                    } catch (Throwable $throwable) {
+                        throw new RuntimeException(
+                            sprintf('[%s] %s', $result['class'], (string) ($result['message'] ?? '')),
+                            (int) ($result['code'] ?? 0),
+                            $throwable
+                        );
+                    }
+                    throw $throwable;
+                }
+            } else {
+                ++$context->index;
+                WorkflowStub::setContext($context);
+                return resolve($result);
+            }
         }
 
         if (! $context->replaying) {

src/Exception.php

@@ -53,7 +53,7 @@ public function handle()
         try {
             if ($this->storedWorkflow->hasLogByIndex($this->index)) {
                 $workflow->resume();
-            } else {
+            } elseif (! $this->storedWorkflow->logs()->where('class', self::class)->exists()) {
                 $workflow->next($this->index, $this->now, self::class, $this->exception);
             }
         } catch (TransitionNotFound) {

src/WorkflowStub.php

@@ -288,12 +288,44 @@ public function fail($exception): void
 
         $this->storedWorkflow->parents()
             ->each(static function ($parentWorkflow) use ($exception) {
-                try {
-                    $parentWorkflow->toWorkflow()
-                        ->fail($exception);
-                } catch (TransitionNotFound) {
+                if (
+                    $parentWorkflow->pivot->parent_index === StoredWorkflow::CONTINUE_PARENT_INDEX
+                    || $parentWorkflow->pivot->parent_index === StoredWorkflow::ACTIVE_WORKFLOW_INDEX
+                ) {
+                    try {
+                        $parentWorkflow->toWorkflow()
+                            ->fail($exception);
+                    } catch (TransitionNotFound) {
+                        return;
+                    }
                     return;
                 }
+
+                $file = new SplFileObject($exception->getFile());
+                $iterator = new LimitIterator($file, max(0, $exception->getLine() - 4), 7);
+
+                $throwable = [
+                    'class' => get_class($exception),
+                    'message' => $exception->getMessage(),
+                    'code' => $exception->getCode(),
+                    'line' => $exception->getLine(),
+                    'file' => $exception->getFile(),
+                    'trace' => collect($exception->getTrace())
+                        ->filter(static fn ($trace) => Serializer::serializable($trace))
+                        ->toArray(),
+                    'snippet' => array_slice(iterator_to_array($iterator), 0, 7),
+                ];
+
+                $parentWf = $parentWorkflow->toWorkflow();
+
+                Exception::dispatch(
+                    $parentWorkflow->pivot->parent_index,
+                    $parentWorkflow->pivot->parent_now,
+                    $parentWorkflow,
+                    $throwable,
+                    $parentWf->connection(),
+                    $parentWf->queue()
+                );
             });
     }
 

tests/Feature/SagaChildWorkflowTest.php

@@ -0,0 +1,38 @@
+<?php
+
+declare(strict_types=1);
+
+namespace Tests\Feature;
+
+use Tests\Fixtures\TestSagaChildWorkflow;
+use Tests\Fixtures\TestSagaSingleChildWorkflow;
+use Tests\TestCase;
+use Workflow\States\WorkflowCompletedStatus;
+use Workflow\WorkflowStub;
+
+final class SagaChildWorkflowTest extends TestCase
+{
+    public function testSingleChildExceptionTriggersCompensation(): void
+    {
+        $workflow = WorkflowStub::make(TestSagaSingleChildWorkflow::class);
+
+        $workflow->start();
+
+        while ($workflow->running());
+
+        $this->assertSame(WorkflowCompletedStatus::class, $workflow->status());
+        $this->assertSame('compensated', $workflow->output());
+    }
+
+    public function testParallelChildExceptionsTriggersCompensation(): void
+    {
+        $workflow = WorkflowStub::make(TestSagaChildWorkflow::class);
+
+        $workflow->start();
+
+        while ($workflow->running());
+
+        $this->assertSame(WorkflowCompletedStatus::class, $workflow->status());
+        $this->assertSame('compensated', $workflow->output());
+    }
+}

tests/Fixtures/TestChildExceptionThrowingWorkflow.php

@@ -0,0 +1,16 @@
+<?php
+
+declare(strict_types=1);
+
+namespace Tests\Fixtures;
+
+use Exception;
+use Workflow\Workflow;
+
+class TestChildExceptionThrowingWorkflow extends Workflow
+{
+    public function execute()
+    {
+        throw new Exception('child failed');
+    }
+}

tests/Fixtures/TestRequiredArgException.php

@@ -0,0 +1,18 @@
+<?php
+
+declare(strict_types=1);
+
+namespace Tests\Fixtures;
+
+use Exception;
+
+class TestRequiredArgException extends Exception
+{
+    public function __construct(
+        string $message,
+        int $code,
+        private readonly string $requiredContext,
+    ) {
+        parent::__construct($message, $code);
+    }
+}

tests/Fixtures/TestSagaChildWorkflow.php

@@ -0,0 +1,33 @@
+<?php
+
+declare(strict_types=1);
+
+namespace Tests\Fixtures;
+
+use Workflow\Workflow;
+use function Workflow\{activity, all, child};
+
+class TestSagaChildWorkflow extends Workflow
+{
+    public function execute()
+    {
+        try {
+            yield activity(TestActivity::class);
+            $this->addCompensation(static fn () => activity(TestUndoActivity::class));
+
+            $children = [
+                child(TestChildExceptionThrowingWorkflow::class),
+                child(TestChildExceptionThrowingWorkflow::class),
+                child(TestChildExceptionThrowingWorkflow::class),
+            ];
+
+            yield all($children);
+
+            return 'success';
+        } catch (\Throwable $th) {
+            yield from $this->compensate();
+
+            return 'compensated';
+        }
+    }
+}

tests/Fixtures/TestSagaSingleChildWorkflow.php

@@ -0,0 +1,27 @@
+<?php
+
+declare(strict_types=1);
+
+namespace Tests\Fixtures;
+
+use Workflow\Workflow;
+use function Workflow\{activity, child};
+
+class TestSagaSingleChildWorkflow extends Workflow
+{
+    public function execute()
+    {
+        try {
+            yield activity(TestActivity::class);
+            $this->addCompensation(static fn () => activity(TestUndoActivity::class));
+
+            yield child(TestChildExceptionThrowingWorkflow::class);
+
+            return 'success';
+        } catch (\Throwable $th) {
+            yield from $this->compensate();
+
+            return 'compensated';
+        }
+    }
+}