Add durable Saga compensation scopes (#1063)

CHANGELOG.md

@@ -2,6 +2,12 @@
 
 ## Unreleased
 
+- Experimental embedded compensation scopes retain bounded, versioned obligations
+  before effects and recover them through native Saga and recorded Activities.
+  Host checks run outside SQL locks. Recovery can restart after cancellation or
+  termination without repeating verified compensation. Scope forward ownership
+  stays with the originating run, including across continue-as-new and redrive.
+
 - A child handed to a continued run now gets its retries (#1105). Before, a
   child that failed after its parent continued as new was never retried,
   because the retry needed the parent to be waiting. It is now retried under

docs/api-stability.md

@@ -114,7 +114,26 @@ The workflow package is consumed by three distinct audiences:
    into the workflow package as a library but do not author workflows.
 
 The semver guarantee below covers all three surfaces. Breaking changes to
-anything marked `@api` require a major version bump.
+anything marked `@api` require a major version bump, except surfaces explicitly
+marked `@experimental` as described below.
+
+## Experimental compensation scopes
+
+The entire `Workflow\V2\Compensation` namespace is experimental, including
+its authoring helpers, handler and authorization contracts, activities,
+recovery helpers and persisted ledger implementation. Its types carry
+`@experimental` in code. This namespace is not covered by the stability
+guarantees in this document. Its API and storage contract may change in a
+minor release while the first increment is evaluated. `@experimental` takes
+precedence over any otherwise applicable public-API classification below.
+`@internal` types remain internal and do not become supported host entry points.
+
+This first increment supports embedded Laravel, sequential compensation, one
+nesting level and explicit release. It does not transfer forward scope ownership
+across continue-as-new or originating-run redrive. Retained obligations stay
+attached to the original run and remain recoverable through `ScopeRecovery`.
+See [durable compensation scopes](compensation-scopes.md) for the operational
+boundaries. Existing run-local Saga and its stability contract are unchanged.
 
 ## What is covered
 

regression-corpus-policy.json

@@ -74,6 +74,12 @@
             "history_sequence",
             "history_payload"
           ]
+        },
+        {
+          "glob": "src/V2/Support/FailedRunRedrivePlan.php"
+        },
+        {
+          "glob": "src/V2/Support/ActivityRetryHistory.php"
         }
       ]
     }

src/V2/Compensation/Activities/CompensateObligationActivity.php

@@ -0,0 +1,89 @@
+<?php
+
+declare(strict_types=1);
+
+namespace Workflow\V2\Compensation\Activities;
+
+use Throwable;
+use Workflow\Exceptions\NonRetryableException;
+use Workflow\V2\Activity;
+use Workflow\V2\Attributes\Type;
+use Workflow\V2\Compensation\Definition\Obligation;
+use Workflow\V2\Compensation\Definition\ScopeIdentity;
+use Workflow\V2\Compensation\Execution\HandlerRegistry;
+use Workflow\V2\Compensation\Execution\OperationContext;
+use Workflow\V2\Compensation\Execution\Resolution;
+use Workflow\V2\Compensation\Persistence\ChildScopes;
+use Workflow\V2\Compensation\Persistence\ScopeProgress;
+use Workflow\V2\Compensation\Recovery\ChildResolution;
+use Workflow\V2\Compensation\Security\ScopeBoundary;
+
+/**
+ * @internal
+ *
+ * @experimental The Compensation API is not covered by stability guarantees and may change in a minor release.
+ */
+#[Type('dw.compensation.compensate-obligation.v1')]
+final class CompensateObligationActivity extends Activity
+{
+    public int $tries = 3;
+
+    public function handle(array $identity, string $id): void
+    {
+        $scope = ScopeIdentity::fromArray($identity);
+        $boundary = app(ScopeBoundary::class);
+        $origin = null;
+        $row = $boundary->run($this, $scope, 'compensate', null, function (array $snapshot) use (
+            $scope,
+            $id,
+            &$origin
+        ): ?array {
+            $origin = $snapshot['original_run_id'];
+            return app(ScopeProgress::class)->beginCompensation($scope, $id, $this->attemptId());
+        });
+        if ($row === null) {
+            return;
+        }
+        $obligation = Obligation::fromStored($row);
+        $context = new OperationContext(
+            $scope,
+            $obligation,
+            $id,
+            $origin,
+            $this->activityId(),
+            $this->attemptId(),
+            $row['forward_status'] !== 'observed',
+            $row['forward_activity_id'],
+            $row['forward_attempt_id'],
+        );
+        $unavailable = false;
+        try {
+            if ($obligation->handlerType === ChildScopes::TYPE) {
+                $result = app(ChildResolution::class)->resolve($scope, $obligation, $id);
+            } else {
+                $handler = app(HandlerRegistry::class)->resolve($obligation);
+                $result = $row['forward_status'] === 'registered'
+                    ? Resolution::verified(['never-dispatched:' . $id], true)
+                    : $handler->compensate($context);
+            }
+        } catch (NonRetryableException) {
+            $unavailable = true;
+            $result = Resolution::unknown('handler-unavailable');
+        } catch (Throwable) {
+            $result = Resolution::unknown('provider-outcome-unknown');
+        }
+        $boundary->run(
+            $this,
+            $scope,
+            'compensate',
+            null,
+            fn () => app(ScopeProgress::class)->resolve($scope, $id, $this->attemptId(), $result)
+        );
+        if ($unavailable) {
+            throw new NonRetryableException('scope_handler_unavailable');
+        }
+        if (! $result->verified) {
+            throw new \RuntimeException('scope_compensation_unknown');
+        }
+    }
+}

src/V2/Compensation/Activities/ForwardOperationActivity.php

@@ -0,0 +1,104 @@
+<?php
+
+declare(strict_types=1);
+
+namespace Workflow\V2\Compensation\Activities;
+
+use Throwable;
+use Workflow\V2\Activity;
+use Workflow\V2\Attributes\Type;
+use Workflow\V2\Compensation\Definition\Obligation;
+use Workflow\V2\Compensation\Definition\PortableDescriptor;
+use Workflow\V2\Compensation\Definition\ScopeIdentity;
+use Workflow\V2\Compensation\Execution\ForwardResult;
+use Workflow\V2\Compensation\Execution\HandlerRegistry;
+use Workflow\V2\Compensation\Execution\OperationContext;
+use Workflow\V2\Compensation\Execution\ScopedOperation;
+use Workflow\V2\Compensation\Persistence\ScopeProgress;
+use Workflow\V2\Compensation\Persistence\ScopeRepository;
+use Workflow\V2\Compensation\Security\ScopeBoundary;
+
+/**
+ * @internal
+ *
+ * @experimental The Compensation API is not covered by stability guarantees and may change in a minor release.
+ */
+#[Type('dw.compensation.forward-operation.v1')]
+final class ForwardOperationActivity extends Activity
+{
+    public int $tries = 3;
+
+    public function handle(array $identity, string $id, string $principal): array
+    {
+        $scope = ScopeIdentity::fromArray($identity);
+        $boundary = app(ScopeBoundary::class);
+        $handler = null;
+        $fingerprint = null;
+        $admission = $boundary->run(
+            $this,
+            $scope,
+            'forward',
+            $principal,
+            function () use ($scope, $id, &$handler, &$fingerprint): array {
+                return app(ScopeProgress::class)->admitForward(
+                    $scope,
+                    $id,
+                    $this->attemptId(),
+                    $this->activityId(),
+                    static function (array $row) use (&$handler, &$fingerprint): void {
+                        if (! $handler instanceof ScopedOperation || $row['fingerprint'] !== $fingerprint) {
+                            throw new \LogicException('The prepared forward descriptor changed before admission.');
+                        }
+                    }
+                );
+            },
+            static function () use ($scope, $id, &$handler, &$fingerprint): void {
+                $snapshot = app(ScopeRepository::class)->inspect($scope);
+                foreach ($snapshot['obligations'] as $row) {
+                    if ($row['id'] !== $id) {
+                        continue;
+                    }
+                    if ($row['forward_status'] !== 'observed') {
+                        $row['descriptor'] = PortableDescriptor::encode($row['descriptor']);
+                        $handler = app(HandlerRegistry::class)->resolve(Obligation::fromStored($row));
+                        $fingerprint = $row['fingerprint'];
+                    }
+                    return;
+                }
+                throw new \LogicException('The prepared forward obligation is unavailable.');
+            }
+        );
+        if ($admission['cached'] !== null) {
+            return $admission['cached']['value'];
+        }
+        if (! $handler instanceof ScopedOperation) {
+            throw new \LogicException('The forward handler was not validated before dispatch.');
+        }
+        $obligation = Obligation::fromStored($admission['row']);
+        $context = new OperationContext(
+            $scope,
+            $obligation,
+            $id,
+            $this->runId(),
+            $this->activityId(),
+            $this->attemptId(),
+            $admission['uncertain']
+        );
+        try {
+            $result = $handler->execute($context);
+        } catch (Throwable) {
+            $result = ForwardResult::unknown('provider-outcome-unknown');
+        }
+        $boundary->run(
+            $this,
+            $scope,
+            'forward',
+            $principal,
+            fn () => app(ScopeProgress::class)->forwardResult($scope, $id, $this->attemptId(), $result)
+        );
+        if (! $result->observed) {
+            throw new \RuntimeException('scope_forward_unknown');
+        }
+        return $result->value;
+    }
+}

src/V2/Compensation/Activities/PrepareRecoveryActivity.php

@@ -0,0 +1,38 @@
+<?php
+
+declare(strict_types=1);
+
+namespace Workflow\V2\Compensation\Activities;
+
+use Workflow\V2\Activity;
+use Workflow\V2\Attributes\Type;
+use Workflow\V2\Compensation\Definition\ScopeIdentity;
+use Workflow\V2\Compensation\Security\ScopeBoundary;
+
+/**
+ * @internal
+ *
+ * @experimental The Compensation API is not covered by stability guarantees and may change in a minor release.
+ */
+#[Type('dw.compensation.prepare-recovery.v1')]
+final class PrepareRecoveryActivity extends Activity
+{
+    public int $tries = 3;
+
+    public function handle(array $identity): array
+    {
+        $scope = ScopeIdentity::fromArray($identity);
+        return app(ScopeBoundary::class)->run(
+            $this,
+            $scope,
+            'compensate',
+            null,
+            static function (array $snapshot): array {
+                if (! in_array($snapshot['status'], ['requested', 'verified'], true)) {
+                    throw new \LogicException('The scope has no accepted compensation request.');
+                }
+                return array_column($snapshot['obligations'], 'id');
+            }
+        );
+    }
+}

src/V2/Compensation/Activities/RegisterChildScopeActivity.php

@@ -0,0 +1,66 @@
+<?php
+
+declare(strict_types=1);
+
+namespace Workflow\V2\Compensation\Activities;
+
+use Throwable;
+use Workflow\Exceptions\NonRetryableException;
+use Workflow\V2\Activity;
+use Workflow\V2\Attributes\Type;
+use Workflow\V2\Compensation\Definition\PortableDescriptor;
+use Workflow\V2\Compensation\Definition\ScopeIdentity;
+use Workflow\V2\Compensation\Persistence\ChildScopes;
+use Workflow\V2\Compensation\Persistence\ScopeStorage;
+use Workflow\V2\Compensation\Security\ScopeAuthorization;
+use Workflow\V2\Compensation\Security\ScopeBoundary;
+
+/**
+ * @internal
+ *
+ * @experimental The Compensation API is not covered by stability guarantees and may change in a minor release.
+ */
+#[Type('dw.compensation.register-child.v1')]
+final class RegisterChildScopeActivity extends Activity
+{
+    public int $tries = 3;
+
+    public function handle(array $identity, string $key, array $childIdentity, string $principal): string
+    {
+        $scope = ScopeIdentity::fromArray($identity);
+        $child = ScopeIdentity::fromArray($childIdentity);
+        return app(ScopeBoundary::class)->run(
+            $this,
+            $scope,
+            'register',
+            $principal,
+            function (?array $snapshot, string $parentPolicy, string $policy) use (
+                $scope,
+                $key,
+                $child,
+                $principal
+            ): string {
+                $id = app(ChildScopes::class)->register($scope, $this->runId(), $key, $child);
+                ScopeStorage::append(ScopeStorage::locked($scope), 'child_authorized', [
+                    'childScopeId' => $child->id(),
+                    'principal' => $principal,
+                    'operation' => 'participate',
+                    'policyVersion' => $policy,
+                    'runId' => $this->runId(),
+                    'activityId' => $this->activityId(),
+                    'attemptId' => $this->attemptId(),
+                ]);
+                return $id;
+            },
+            static function () use ($child, $principal): string {
+                try {
+                    $policy = app(ScopeAuthorization::class)->assertAllowed($child, $principal, 'participate');
+                    PortableDescriptor::identifier($policy, 'authorization policy version');
+                    return $policy;
+                } catch (Throwable) {
+                    throw new NonRetryableException('scope_child_authorization_refused');
+                }
+            }
+        );
+    }
+}

src/V2/Compensation/Activities/RegisterObligationActivity.php

@@ -0,0 +1,38 @@
+<?php
+
+declare(strict_types=1);
+
+namespace Workflow\V2\Compensation\Activities;
+
+use Workflow\V2\Activity;
+use Workflow\V2\Attributes\Type;
+use Workflow\V2\Compensation\Definition\Obligation;
+use Workflow\V2\Compensation\Definition\ScopeIdentity;
+use Workflow\V2\Compensation\Execution\HandlerRegistry;
+use Workflow\V2\Compensation\Persistence\ScopeRepository;
+use Workflow\V2\Compensation\Security\ScopeBoundary;
+
+/**
+ * @internal
+ *
+ * @experimental The Compensation API is not covered by stability guarantees and may change in a minor release.
+ */
+#[Type('dw.compensation.register-obligation.v1')]
+final class RegisterObligationActivity extends Activity
+{
+    public int $tries = 3;
+
+    public function handle(array $identity, array $descriptor, string $principal): string
+    {
+        $scope = ScopeIdentity::fromArray($identity);
+        $obligation = Obligation::fromArray($descriptor);
+        return app(ScopeBoundary::class)->run($this, $scope, 'register', $principal, function () use (
+            $scope,
+            $obligation
+        ): string {
+            $repository = app(ScopeRepository::class);
+            $repository->open($scope, $this->runId());
+            return $repository->register($scope, $obligation);
+        }, static fn () => app(HandlerRegistry::class)->resolve($obligation));
+    }
+}