From 440e2572902ba2fd3034e3636376e59261006334 Mon Sep 17 00:00:00 2001 From: Dmitriy Derepko Date: Fri, 31 Jul 2026 14:27:04 +0400 Subject: [PATCH] fix: report every failure from parallel saga compensation Parallel Saga::compensate() used Promise::all, which short-circuits on the first rejected compensation, so sibling failures were lost and the caller received only the first (raw) exception. Await every compensation scope and aggregate all failures into a CompensationException (first as cause, rest as suppressed), matching the sdk-java Saga semantics. The success return value stays null and the sequential branch is unchanged. Catch \Exception (not \Throwable) so a \Error thrown by a compensation (e.g. a TypeError from a deployment bug) still propagates raw. The default ExceptionInterceptor classifies \Error as retryable, so wrapping it in CompensationException would turn a retryable task failure into a permanent workflow failure. This matches sdk-java, which catches Exception (not Error). Tests (unit, mock harness): - parallel happy path runs every compensation (ports sdk-java testSagaParallelCompensation) - sequential runs compensations in reverse order (ports sdk-java testSaga) - sequential stops at first failure and throws the raw exception - compensate() return value stays null on success - parallel compensation reports every failure via CompensationException - parallel compensation propagates a raw \Error for retry --- src/Workflow/Saga.php | 19 +- .../SagaParallelCompensationTestCase.php | 236 ++++++++++++++++++ 2 files changed, 253 insertions(+), 2 deletions(-) create mode 100644 tests/Unit/Workflow/SagaParallelCompensationTestCase.php diff --git a/src/Workflow/Saga.php b/src/Workflow/Saga.php index edde8d070..b17ab48ef 100644 --- a/src/Workflow/Saga.php +++ b/src/Workflow/Saga.php @@ -12,7 +12,6 @@ namespace Temporal\Workflow; use Temporal\Exception\CompensationException; -use Temporal\Promise; use Temporal\Workflow; final class Saga @@ -67,7 +66,23 @@ function () { $scopes[] = Workflow::asyncDetached($handler); } - yield Promise::all($scopes); + $sagaException = null; + foreach ($scopes as $scope) { + try { + yield $scope; + } catch (\Exception $e) { + if ($sagaException === null) { + $sagaException = new CompensationException($e->getMessage(), (int) $e->getCode(), $e); + } else { + $sagaException->addSuppressed($e); + } + } + } + + if ($sagaException !== null) { + throw $sagaException; + } + return; } diff --git a/tests/Unit/Workflow/SagaParallelCompensationTestCase.php b/tests/Unit/Workflow/SagaParallelCompensationTestCase.php new file mode 100644 index 000000000..e5a48582e --- /dev/null +++ b/tests/Unit/Workflow/SagaParallelCompensationTestCase.php @@ -0,0 +1,236 @@ +factory = WorkerFactoryMock::create(); + $this->worker = $this->factory->newWorker(); + + parent::setUp(); + } + + public function testParallelCompensationHappyPathCompletes(): void + { + $this->addToAssertionCount(1); + $this->worker->registerWorkflowObject( + new + #[Workflow\WorkflowInterface] + class { + #[WorkflowMethod(name: 'SagaParallelWorkflow')] + public function handler(): iterable + { + $saga = new Saga(); + $saga->setParallelCompensation(true); + $log = []; + $saga->addCompensation(static function () use (&$log): void { + $log[] = 'A'; + }); + $saga->addCompensation(static function () use (&$log): void { + $log[] = 'B'; + }); + + yield $saga->compensate(); + + \sort($log); + + return \implode('|', $log); + } + } + ); + + $this->worker->runWorkflow('SagaParallelWorkflow'); + $this->worker->assertWorkflowReturns('A|B'); + $this->factory->run($this->worker); + } + + public function testParallelCompensationPropagatesRawErrorForRetry(): void + { + $this->addToAssertionCount(1); + $this->worker->registerWorkflowObject( + new + #[Workflow\WorkflowInterface] + class { + #[WorkflowMethod(name: 'SagaParallelWorkflow')] + public function handler(): iterable + { + $saga = new Saga(); + $saga->setParallelCompensation(true); + $saga->addCompensation(static function (): void { + throw new \TypeError('deployment bug'); + }); + + try { + yield $saga->compensate(); + } catch (\Throwable $e) { + return $e::class . '|' . $e->getMessage(); + } + + return 'NO_THROW'; + } + } + ); + + $this->worker->runWorkflow('SagaParallelWorkflow'); + $this->worker->assertWorkflowReturns('TypeError|deployment bug'); + $this->factory->run($this->worker); + } + + public function testSequentialRunsCompensationsInReverseOrder(): void + { + $this->addToAssertionCount(1); + $this->worker->registerWorkflowObject( + new + #[Workflow\WorkflowInterface] + class { + #[WorkflowMethod(name: 'SagaParallelWorkflow')] + public function handler(): iterable + { + $saga = new Saga(); + $saga->setParallelCompensation(false); + + $order = []; + $saga->addCompensation(static function () use (&$order): void { + $order[] = 'first-added'; + }); + $saga->addCompensation(static function () use (&$order): void { + $order[] = 'second-added'; + }); + + yield $saga->compensate(); + + return \implode(',', $order); + } + } + ); + + $this->worker->runWorkflow('SagaParallelWorkflow'); + $this->worker->assertWorkflowReturns('second-added,first-added'); + $this->factory->run($this->worker); + } + + public function testSequentialStopsAtFirstFailureAndThrowsRaw(): void + { + $this->addToAssertionCount(1); + $this->worker->registerWorkflowObject( + new + #[Workflow\WorkflowInterface] + class { + #[WorkflowMethod(name: 'SagaParallelWorkflow')] + public function handler(): iterable + { + $saga = new Saga(); + $saga->setParallelCompensation(false); + + $ran = []; + $saga->addCompensation(static function () use (&$ran): void { + $ran[] = 'first-added'; + }); + $saga->addCompensation(static function (): void { + throw new \RuntimeException('boom-last'); + }); + + try { + yield $saga->compensate(); + } catch (\Throwable $e) { + return $e::class . '|' . $e->getMessage() . '|ran=' . \implode(',', $ran); + } + + return 'NO_THROW'; + } + } + ); + + $this->worker->runWorkflow('SagaParallelWorkflow'); + $this->worker->assertWorkflowReturns('RuntimeException|boom-last|ran='); + $this->factory->run($this->worker); + } + + public function testCompensateReturnValueProbe(): void + { + $this->addToAssertionCount(1); + $this->worker->registerWorkflowObject( + new + #[Workflow\WorkflowInterface] + class { + #[WorkflowMethod(name: 'SagaParallelWorkflow')] + public function handler(): iterable + { + $saga = new Saga(); + $saga->setParallelCompensation(true); + $saga->addCompensation(static fn(): string => 'A'); + $saga->addCompensation(static fn(): string => 'B'); + + $result = yield $saga->compensate(); + + return 'RET:' . \var_export($result, true); + } + } + ); + + $this->worker->runWorkflow('SagaParallelWorkflow'); + $this->worker->assertWorkflowReturns('RET:NULL'); + $this->factory->run($this->worker); + } + + public function testParallelCompensationReportsEveryFailure(): void + { + $this->addToAssertionCount(1); + $this->worker->registerWorkflowObject( + new + #[Workflow\WorkflowInterface] + class { + #[WorkflowMethod(name: 'SagaParallelWorkflow')] + public function handler(): iterable + { + $saga = new Saga(); + $saga->setParallelCompensation(true); + $saga->addCompensation(static function (): void { + throw new \RuntimeException('comp-A'); + }); + $saga->addCompensation(static function (): void { + throw new \RuntimeException('comp-B'); + }); + + try { + yield $saga->compensate(); + } catch (\Throwable $e) { + $seen = [$e->getMessage()]; + if ($e instanceof CompensationException) { + foreach ($e->getSuppressed() as $suppressed) { + $seen[] = $suppressed->getMessage(); + } + } + \sort($seen); + + return \implode('|', \array_unique($seen)); + } + + return 'NO_THROW'; + } + } + ); + + $this->worker->runWorkflow('SagaParallelWorkflow'); + $this->worker->assertWorkflowReturns('comp-A|comp-B'); + $this->factory->run($this->worker); + } +}