Skip to content

Commit 1f72c41

Browse files
rmcdanielrmcdaniel
authored
Merge pull request #846 from durable-workflow/fix/nexus-external-caller-namespace
Keep external Nexus results in the caller storage namespace
2 parents 562aca0 + 3531d8d commit 1f72c41

3 files changed

Lines changed: 207 additions & 2 deletions

File tree

‎composer.json‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -79,7 +79,7 @@
7979
"dev-main": "2.0.x-dev"
8080
},
8181
"durable-workflow": {
82-
"product-train": "2.6.0",
82+
"product-train": "2.6.1",
8383
"laravel-embedded-upgrade-contract": "resources/laravel-embedded-upgrade-contract.json",
8484
"laravel-dependency-security-policy": "resources/laravel-dependency-security-policy.json"
8585
},

‎src/V2/Support/ServiceCallWorkflowDelivery.php‎

Lines changed: 24 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -8,6 +8,7 @@
88
use Workflow\V2\Enums\HistoryEventType;
99
use Workflow\V2\Enums\TaskStatus;
1010
use Workflow\V2\Enums\TaskType;
11+
use Workflow\V2\Exceptions\ExternalPayloadIntegrityException;
1112
use Workflow\V2\Models\WorkflowHistoryEvent;
1213
use Workflow\V2\Models\WorkflowRun;
1314
use Workflow\V2\Models\WorkflowServiceCall;
@@ -223,7 +224,7 @@ private static function terminalPayload(WorkflowServiceCall $call, WorkflowHisto
223224
'completed_at' => $call->completed_at?->toJSON(),
224225
'failed_at' => $call->failed_at?->toJSON(),
225226
'cancelled_at' => $call->cancelled_at?->toJSON(),
226-
'result_envelope' => ServiceCallResultEnvelope::fromCall($call),
227+
'result_envelope' => self::resultForCaller($call),
227228
]);
228229
$payload = array_replace($payload, [
229230
'status' => $call->status,
@@ -253,4 +254,26 @@ private static function terminalPayload(WorkflowServiceCall $call, WorkflowHisto
253254

254255
return $payload;
255256
}
257+
258+
/**
259+
* @return array<string, mixed>|null
260+
*/
261+
private static function resultForCaller(WorkflowServiceCall $call): ?array
262+
{
263+
$result = ServiceCallResultEnvelope::fromCall($call);
264+
if ($call->caller_namespace === $call->target_namespace
265+
|| ! is_array($result['external_storage'] ?? null)) {
266+
return $result;
267+
}
268+
269+
// Terminal delivery grants the caller the value, not access to the
270+
// service namespace's provider or authenticated transport reference.
271+
$blob = ExternalPayloads::payloadBlob($result, $call->payload_codec, $call->target_namespace);
272+
if ($blob === null) {
273+
throw new ExternalPayloadIntegrityException('The external Nexus result has no encoded payload.');
274+
}
275+
$stored = ExternalPayloads::externalizeForNamespace($blob, $call->payload_codec, $call->caller_namespace);
276+
277+
return ExternalPayloads::wireEnvelope($stored, $call->payload_codec, $call->caller_namespace);
278+
}
256279
}

‎tests/Unit/V2/LinkedServiceCallCloserTest.php‎

Lines changed: 182 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -6,10 +6,13 @@
66

77
use Illuminate\Support\Carbon;
88
use Mockery;
9+
use PHPUnit\Framework\Attributes\DataProvider;
910
use RuntimeException;
1011
use Tests\TestCase;
1112
use Workflow\Serializers\Serializer;
1213
use Workflow\V2\Attributes\Type;
14+
use Workflow\V2\Contracts\ExternalPayloadStorageDriver;
15+
use Workflow\V2\Contracts\ExternalPayloadStoragePolicy;
1316
use Workflow\V2\Contracts\ServiceControlPlane;
1417
use Workflow\V2\Contracts\WorkflowControlPlane;
1518
use Workflow\V2\Contracts\WorkflowTaskBridge;
@@ -29,9 +32,12 @@
2932
use Workflow\V2\Models\WorkflowServiceOperation;
3033
use Workflow\V2\Models\WorkflowTask;
3134
use Workflow\V2\Support\DefaultServiceControlPlane;
35+
use Workflow\V2\Support\ExternalPayloads;
36+
use Workflow\V2\Support\ExternalPayloadStorage;
3237
use Workflow\V2\Support\LinkedServiceCallCloser;
3338
use Workflow\V2\Support\ServiceCallDeadline;
3439
use Workflow\V2\Support\ServiceCallView;
40+
use Workflow\V2\Support\ServiceCallWorkflowDelivery;
3541
use Workflow\V2\Support\ServiceOperationOptions;
3642
use Workflow\V2\Support\ServiceOperationResult;
3743
use Workflow\V2\TaskWatchdog;
@@ -818,6 +824,137 @@ public function testExternalResultKeepsIntegrityMetadataWithoutCopyingProviderUr
818824
], $view['result_envelope']);
819825
}
820826

827+
#[DataProvider('externalCallerNamespaces')]
828+
public function testExternalTerminalDeliveryBelongsToTheCallerNamespace(string $namespace): void
829+
{
830+
ExternalPayloadStorage::flushVerifiedCache();
831+
try {
832+
[$caller, $call, $task, $targetStorage, $callerStorage, $blob] = $this->externalCaller($namespace);
833+
$originalCall = $call->getAttributes();
834+
$claim = $this->app->make(WorkflowTaskBridge::class)->claimStatus($task->id, 'cold-caller');
835+
$this->assertTrue($claim['claimed']);
836+
$event = WorkflowHistoryEvent::query()->where('workflow_run_id', $caller->id)
837+
->where('event_type', HistoryEventType::ServiceCallCompleted->value)->sole();
838+
$envelope = $event->payload['response_payload'];
839+
$this->assertStringStartsWith('memory://' . $namespace . '/', $envelope['external_storage']['uri']);
840+
$this->assertSame($call->id, $event->payload['service_call_id']);
841+
$this->assertSame(7, $event->payload['sequence']);
842+
$this->assertSame($task->id, $event->workflow_task_id);
843+
$this->assertSame($envelope, $event->payload['service_call']['result_envelope']);
844+
$this->assertSame($envelope, $event->payload['response_or_failure_surface']['result_envelope']);
845+
$this->assertSame($originalCall, $call->refresh()->getAttributes());
846+
$this->assertSame(strlen($blob), $envelope['external_storage']['size_bytes']);
847+
$this->assertSame(hash('sha256', $blob), $envelope['external_storage']['sha256']);
848+
if ($namespace === 'service') {
849+
$this->assertSame($call->output_payload_reference, $envelope['external_storage']['uri']);
850+
$this->assertSame(1, $targetStorage->writes);
851+
} else {
852+
$this->assertNotSame($call->output_payload_reference, $envelope['external_storage']['uri']);
853+
$this->assertSame(1, $callerStorage->writes);
854+
$targetStorage->failReads = true;
855+
}
856+
ExternalPayloadStorage::flushVerifiedCache();
857+
$this->assertSame($blob, ExternalPayloads::payloadBlob($envelope, 'avro', $namespace));
858+
$originalEvent = $event->getAttributes();
859+
ServiceCallWorkflowDelivery::recordForClaim($task->fresh(), $caller->fresh());
860+
$this->assertSame($originalEvent, $event->refresh()->getAttributes());
861+
$this->assertSame(1, $callerStorage->writes);
862+
$this->assertSame(1, WorkflowHistoryEvent::query()->where('workflow_run_id', $caller->id)
863+
->where('event_type', HistoryEventType::ServiceCallCompleted->value)->count());
864+
} finally {
865+
ExternalPayloadStorage::flushVerifiedCache();
866+
}
867+
}
868+
869+
public static function externalCallerNamespaces(): iterable
870+
{
871+
yield 'cross namespace' => ['caller'];
872+
yield 'same namespace' => ['service'];
873+
}
874+
875+
public function testCallerStorageFailureRollsBackTheOriginalClaimAndCanRecover(): void
876+
{
877+
ExternalPayloadStorage::flushVerifiedCache();
878+
try {
879+
[$caller, $call, $task, , $callerStorage] = $this->externalCaller('caller');
880+
$originalCall = $call->getAttributes();
881+
$callerStorage->failWrites = true;
882+
$failure = null;
883+
try {
884+
$this->app->make(WorkflowTaskBridge::class)->claimStatus($task->id, 'refused-caller');
885+
} catch (RuntimeException $error) {
886+
$failure = $error;
887+
}
888+
$this->assertInstanceOf(
889+
RuntimeException::class,
890+
$failure,
891+
'Caller storage must be available before publishing its result history.'
892+
);
893+
$this->assertSame('caller write refused', $failure->getMessage());
894+
$this->assertSame(TaskStatus::Ready, $task->refresh()->status);
895+
$this->assertSame(0, $task->attempt_count);
896+
$this->assertNull($task->lease_owner);
897+
$this->assertTrue($task->payload['service_call_terminal_pending']);
898+
$this->assertSame(0, WorkflowHistoryEvent::query()->where('workflow_run_id', $caller->id)
899+
->where('event_type', HistoryEventType::ServiceCallCompleted->value)->count());
900+
$this->assertSame($originalCall, $call->refresh()->getAttributes());
901+
$callerStorage->failWrites = false;
902+
$this->assertTrue(
903+
$this->app->make(WorkflowTaskBridge::class)->claimStatus($task->id, 'restored-caller')['claimed']
904+
);
905+
$this->assertSame(1, $task->refresh()->attempt_count);
906+
$this->assertSame($task->id, WorkflowHistoryEvent::query()->where('workflow_run_id', $caller->id)
907+
->where('event_type', HistoryEventType::ServiceCallCompleted->value)->sole()->workflow_task_id);
908+
} finally {
909+
ExternalPayloadStorage::flushVerifiedCache();
910+
}
911+
}
912+
913+
/**
914+
* @return array{WorkflowRun, WorkflowServiceCall, WorkflowTask, NexusNamespaceExternalStorage, NexusNamespaceExternalStorage, string}
915+
*/
916+
private function externalCaller(string $namespace): array
917+
{
918+
config([
919+
'workflows.v2.task_dispatch_mode' => 'poll',
920+
]);
921+
$targetStorage = new NexusNamespaceExternalStorage('service');
922+
$callerStorage = $namespace === 'service' ? $targetStorage : new NexusNamespaceExternalStorage('caller');
923+
$drivers = [
924+
'service' => $targetStorage,
925+
$namespace => $callerStorage,
926+
];
927+
$policy = Mockery::mock(ExternalPayloadStoragePolicy::class);
928+
$policy->shouldReceive('driverFor')
929+
->andReturnUsing(static fn ($name) => $drivers[$name] ?? null);
930+
$policy->shouldReceive('thresholdBytesFor')
931+
->andReturn(32);
932+
$this->app->instance(ExternalPayloadStoragePolicy::class, $policy);
933+
$service = $this->createRun();
934+
$caller = $this->createRun($namespace);
935+
$call = $this->createCall($service, [
936+
'caller_namespace' => $namespace,
937+
'caller_workflow_instance_id' => $caller->workflow_instance_id,
938+
'caller_workflow_run_id' => $caller->id,
939+
'operation_mode' => 'sync',
940+
]);
941+
$this->recordAdmission($caller, $call);
942+
$blob = Serializer::serializeWithCodec('avro', [
943+
'payload' => str_repeat('value', 50),
944+
'bigint' => PHP_INT_MAX,
945+
'nullable' => null,
946+
'unicode' => '雪☃',
947+
]);
948+
$stored = ExternalPayloads::externalizeForNamespace($blob, 'avro', 'service');
949+
WorkflowHistoryEvent::record($service, HistoryEventType::WorkflowCompleted, [
950+
'output' => ExternalPayloads::historyValue($stored, 'avro', 'service'),
951+
'payload_codec' => 'avro',
952+
]);
953+
954+
return [$caller, $call->refresh(), WorkflowTask::query()->where('workflow_run_id', $caller->id)->sole(),
955+
$targetStorage, $callerStorage, $blob];
956+
}
957+
821958
/**
822959
* @param array<string, mixed> $options
823960
*/
@@ -909,6 +1046,51 @@ private function createCall(WorkflowRun $run, array $overrides = []): WorkflowSe
9091046
}
9101047
}
9111048

1049+
final class NexusNamespaceExternalStorage implements ExternalPayloadStorageDriver
1050+
{
1051+
public bool $failWrites = false;
1052+
1053+
public bool $failReads = false;
1054+
1055+
public int $writes = 0;
1056+
1057+
/**
1058+
* @var array<string, string>
1059+
*/
1060+
private array $objects = [];
1061+
1062+
public function __construct(
1063+
private readonly string $namespace
1064+
) {
1065+
}
1066+
1067+
public function put(string $data, string $sha256, string $codec): string
1068+
{
1069+
if ($this->failWrites) {
1070+
throw new RuntimeException('caller write refused');
1071+
}
1072+
++$this->writes;
1073+
$uri = 'memory://' . $this->namespace . '/' . $sha256;
1074+
$this->objects[$uri] = $data;
1075+
1076+
return $uri;
1077+
}
1078+
1079+
public function get(string $uri): string
1080+
{
1081+
if ($this->failReads || ! isset($this->objects[$uri])) {
1082+
throw new RuntimeException('Reference is unavailable in this namespace.');
1083+
}
1084+
1085+
return $this->objects[$uri];
1086+
}
1087+
1088+
public function delete(string $uri): void
1089+
{
1090+
unset($this->objects[$uri]);
1091+
}
1092+
}
1093+
9121094
#[Type('nexus-waiting-caller')]
9131095
final class NexusWaitingCallerWorkflow extends Workflow
9141096
{

0 commit comments

Comments
 (0)