From dbb3f20126c99674e7a95f4cb9dd6cb70a481e6d Mon Sep 17 00:00:00 2001 From: Durable Workflow Date: Tue, 8 Sep 2026 16:29:11 +0000 Subject: [PATCH 1/5] Allow bounded leased completion uploads during storage draining --- .../Api/RuntimeExternalPayloadController.php | 28 ++- .../Middleware/EnforceStorageAdmission.php | 17 ++ app/Models/RuntimePayloadCompletionBudget.php | 18 ++ app/Support/RuntimeExternalPayloadCleanup.php | 2 + .../RuntimeExternalPayloadReference.php | 12 + .../RuntimeExternalPayloadRegistry.php | 25 +-- .../RuntimePayloadCompletionContext.php | 109 +++++++++ app/Support/RuntimePayloadCompletionLease.php | 119 ++++++++++ .../RuntimePayloadCompletionUploads.php | 147 ++++++++++++ config/server.php | 5 + ...ate_runtime_payload_completion_budgets.php | 26 +++ docs/storage-admission.md | 34 +++ .../external-payload-transport.openapi.yaml | 19 ++ .../RuntimePayloadCompletionUploadsTest.php | 209 ++++++++++++++++++ .../RuntimePayloadCompletionContextTest.php | 88 ++++++++ 15 files changed, 839 insertions(+), 19 deletions(-) create mode 100644 app/Models/RuntimePayloadCompletionBudget.php create mode 100644 app/Support/RuntimePayloadCompletionContext.php create mode 100644 app/Support/RuntimePayloadCompletionLease.php create mode 100644 app/Support/RuntimePayloadCompletionUploads.php create mode 100644 database/migrations/2026_09_08_000000_create_runtime_payload_completion_budgets.php create mode 100644 tests/Feature/RuntimePayloadCompletionUploadsTest.php create mode 100644 tests/Unit/RuntimePayloadCompletionContextTest.php diff --git a/app/Http/Controllers/Api/RuntimeExternalPayloadController.php b/app/Http/Controllers/Api/RuntimeExternalPayloadController.php index 5916af24..3f234f2f 100644 --- a/app/Http/Controllers/Api/RuntimeExternalPayloadController.php +++ b/app/Http/Controllers/Api/RuntimeExternalPayloadController.php @@ -7,6 +7,10 @@ use App\Support\RuntimeExternalPayloadReference; use App\Support\RuntimeExternalPayloadRegistry; use App\Support\RuntimeExternalPayloadUploadBody; +use App\Support\RuntimePayloadCompletionContext; +use App\Support\RuntimePayloadCompletionUploads; +use App\Support\StorageAdmissionPaused; +use App\Support\StoragePressure; use Illuminate\Http\JsonResponse; use Illuminate\Http\Request; use Symfony\Component\HttpFoundation\Response; @@ -62,12 +66,24 @@ public function store(Request $request): JsonResponse } $namespace = (string) $request->attributes->get('namespace', config('server.default_namespace')); - $reference = $this->registry->upload( - $namespace, - $data, - (string) $request->header('X-Durable-Workflow-Payload-Codec'), - strtolower((string) $request->header('X-Durable-Workflow-Payload-SHA256')), - ); + $context = $request->attributes->get(RuntimePayloadCompletionContext::class); + $codec = (string) $request->header('X-Durable-Workflow-Payload-Codec'); + $sha256 = strtolower((string) $request->header('X-Durable-Workflow-Payload-SHA256')); + try { + if ($context instanceof RuntimePayloadCompletionContext) { + $reference = app(RuntimePayloadCompletionUploads::class)->upload($namespace, $context, $data, $codec, $sha256, $this->registry); + } else { + // Pressure may change while an admitted HTTP upload is being read. + $snapshot = app(StoragePressure::class)->snapshot(); + if ($snapshot['state'] !== 'normal') { + throw new StorageAdmissionPaused($snapshot); + } + $reference = $this->registry->upload($namespace, $data, $codec, $sha256, + fn () => app(StoragePressure::class)->requireNewWork()); + } + } finally { + $data->close(); + } $this->audit->record($request, 'external_payload.uploaded', [ 'reference_identity_sha256' => hash('sha256', $reference['reference_id']), diff --git a/app/Http/Middleware/EnforceStorageAdmission.php b/app/Http/Middleware/EnforceStorageAdmission.php index 96088125..4e9baf79 100644 --- a/app/Http/Middleware/EnforceStorageAdmission.php +++ b/app/Http/Middleware/EnforceStorageAdmission.php @@ -3,6 +3,8 @@ namespace App\Http\Middleware; use App\Support\ControlPlaneProtocol; +use App\Support\RuntimePayloadCompletionContext; +use App\Support\RuntimePayloadCompletionUploads; use App\Support\StoragePressure; use App\Support\WorkerProtocol; use Closure; @@ -39,6 +41,21 @@ public function handle(Request $request, Closure $next): Response { $snapshot = $this->pressure->snapshot(); $action = class_basename((string) $request->route()?->getActionName()); + if ($action === 'RuntimeExternalPayloadController@store' + && $request->headers->has(RuntimePayloadCompletionContext::HEADER) + && in_array($snapshot['state'], ['normal', 'draining'], true)) { + return app(RequireRole::class)->handle($request, function (Request $request) use ($next): Response { + $context = RuntimePayloadCompletionContext::parse((string) $request->header(RuntimePayloadCompletionContext::HEADER)); + app(RuntimePayloadCompletionUploads::class)->authorize( + (string) $request->attributes->get('namespace'), $context, + strtolower((string) $request->header('X-Durable-Workflow-Payload-SHA256')), + (int) $request->header('X-Durable-Workflow-Payload-Size'), + ); + $request->attributes->set(RuntimePayloadCompletionContext::class, $context); + + return $next($request); + }, 'worker'); + } if ($snapshot['state'] === 'normal' || $request->isMethodSafe() || $action === 'WorkerController@workflowTaskHistory' || ($snapshot['state'] === 'draining' && in_array($action, self::DRAIN_ACTIONS, true))) { diff --git a/app/Models/RuntimePayloadCompletionBudget.php b/app/Models/RuntimePayloadCompletionBudget.php new file mode 100644 index 00000000..0eff3d8a --- /dev/null +++ b/app/Models/RuntimePayloadCompletionBudget.php @@ -0,0 +1,18 @@ + 'array', 'slots' => 'array', 'objects' => 'array', 'expires_at' => 'immutable_datetime', + ]; +} diff --git a/app/Support/RuntimeExternalPayloadCleanup.php b/app/Support/RuntimeExternalPayloadCleanup.php index d8c731c0..700c8d2f 100644 --- a/app/Support/RuntimeExternalPayloadCleanup.php +++ b/app/Support/RuntimeExternalPayloadCleanup.php @@ -44,6 +44,8 @@ public function runPass( $limit = min(self::MAX_BATCH_SIZE, max(1, $limit)); $cutoff ??= now(); + app(RuntimePayloadCompletionUploads::class)->cleanup($namespace, $limit, $cutoff); + $candidates = $this->expiredQuery($namespace, $cutoff) ->orderBy('expires_at') ->orderBy('id') diff --git a/app/Support/RuntimeExternalPayloadReference.php b/app/Support/RuntimeExternalPayloadReference.php index def5af90..91931776 100644 --- a/app/Support/RuntimeExternalPayloadReference.php +++ b/app/Support/RuntimeExternalPayloadReference.php @@ -72,6 +72,14 @@ public static function transportManifest(): array 'X-Durable-Workflow-Payload-SHA256', ], 'idempotency' => 'content_addressed_per_namespace', + 'completion_context' => [ + 'schema' => RuntimePayloadCompletionContext::SCHEMA, + 'header' => RuntimePayloadCompletionContext::HEADER, + 'use' => 'retry_draining_refusal_for_current_worker_completion_only', + 'max_bytes_per_lease' => RuntimePayloadCompletionUploads::maxBytes(), + 'max_slots_per_lease' => RuntimePayloadCompletionUploads::MAX_SLOTS, + 'fenced_or_stale_admission' => 'refuse', + ], ], 'fetch' => [ 'method' => 'GET', @@ -121,6 +129,10 @@ public static function transportManifest(): array 'external_payload_namespace_bytes_exhausted' => ['status' => 429, 'retryable' => true], 'external_payload_namespace_objects_exhausted' => ['status' => 429, 'retryable' => true], 'external_payload_namespace_quota_unavailable' => ['status' => 503, 'retryable' => true], + 'external_payload_completion_invalid' => ['status' => 422, 'retryable' => false], + 'external_payload_completion_lease_rejected' => ['status' => 409, 'retryable' => false], + 'external_payload_completion_conflict' => ['status' => 409, 'retryable' => false], + 'external_payload_completion_budget_exhausted' => ['status' => 429, 'retryable' => true], ], 'audit_events' => [ 'external_payload.uploaded', diff --git a/app/Support/RuntimeExternalPayloadRegistry.php b/app/Support/RuntimeExternalPayloadRegistry.php index 39759a09..374bd0f9 100644 --- a/app/Support/RuntimeExternalPayloadRegistry.php +++ b/app/Support/RuntimeExternalPayloadRegistry.php @@ -19,7 +19,7 @@ public function __construct( /** * @return array{schema: string, reference_id: string, codec: string, size_bytes: int, sha256: string} */ - public function upload(string $namespace, string|ExternalPayloadStream $data, string $codec, string $sha256): array + public function upload(string $namespace, string|ExternalPayloadStream $data, string $codec, string $sha256, ?\Closure $beforeWrite = null): array { $namespace = $this->namespace($namespace); try { @@ -56,6 +56,7 @@ public function upload(string $namespace, string|ExternalPayloadStream $data, st sizeBytes: $sizeBytes, retained: false, expiresAt: $expiresAt, + beforeWrite: $beforeWrite, )); } @@ -465,6 +466,7 @@ private function store( int $sizeBytes, bool $retained, mixed $expiresAt, + ?\Closure $beforeWrite = null, ): RuntimeExternalPayload { $observedSize = $data instanceof ExternalPayloadStream ? $data->sizeBytes : strlen($data); $observedHash = $data instanceof ExternalPayloadStream ? $data->sha256 : hash('sha256', $data); @@ -479,17 +481,12 @@ private function store( } try { - $this->objectLock->transaction($uri, fn (): RuntimeExternalPayload => $this->track( - $namespace, - $uri, - $codec, - $sha256, - $sizeBytes, - false, - $expiresAt, - RuntimeExternalPayload::UPLOAD_WRITING, - )); - } catch (RuntimeExternalPayloadException $exception) { + $this->objectLock->transaction($uri, function () use ($namespace, $uri, $codec, $sha256, $sizeBytes, $expiresAt, $beforeWrite): RuntimeExternalPayload { + $beforeWrite?->__invoke(); + + return $this->track($namespace, $uri, $codec, $sha256, $sizeBytes, false, $expiresAt, RuntimeExternalPayload::UPLOAD_WRITING); + }); + } catch (RuntimeExternalPayloadException|StorageAdmissionPaused $exception) { throw $exception; } catch (Throwable $exception) { throw $this->unavailable('External payload reference registration failed before storage commit.', $exception); @@ -506,7 +503,9 @@ private function store( $sizeBytes, $retained, $expiresAt, + $beforeWrite, ): RuntimeExternalPayload { + $beforeWrite?->__invoke(); $row = RuntimeExternalPayload::query() ->where('namespace', $namespace) ->where('storage_uri_sha256', hash('sha256', $uri)) @@ -547,7 +546,7 @@ private function store( RuntimeExternalPayload::UPLOAD_READY, ); }); - } catch (RuntimeExternalPayloadException $exception) { + } catch (RuntimeExternalPayloadException|StorageAdmissionPaused $exception) { throw $exception; } catch (Throwable $exception) { // The writing row was committed before the backing write. If this diff --git a/app/Support/RuntimePayloadCompletionContext.php b/app/Support/RuntimePayloadCompletionContext.php new file mode 100644 index 00000000..a38a0de8 --- /dev/null +++ b/app/Support/RuntimePayloadCompletionContext.php @@ -0,0 +1,109 @@ + $slot */ + private function __construct( + public string $kind, + public string $taskId, + public int|string $attempt, + public string $leaseOwner, + public string $operation, + public array $slot, + ) {} + + public static function parse(string $header): self + { + if (strlen($header) > 4096) { + throw self::invalid(); + } + try { + $value = json_decode($header, true, 8, JSON_THROW_ON_ERROR); + } catch (JsonException) { + throw self::invalid(); + } + $keys = ['schema', 'kind', 'task_id', 'attempt', 'lease_owner', 'operation', 'slot']; + if (! is_array($value) || count($value) !== count($keys) + || array_diff($keys, array_keys($value)) !== [] + || ($value['schema'] ?? null) !== self::SCHEMA + || ! in_array($value['kind'] ?? null, ['activity', 'workflow', 'query'], true) + || ! is_string($value['operation']) + || ! self::identifier($value['task_id']) || ! self::identifier($value['lease_owner']) + || ! is_array($value['slot']) || ! array_is_list($value['slot'])) { + throw self::invalid(); + } + if ($value['kind'] === 'activity' + ? ! self::identifier($value['attempt']) + : (! is_int($value['attempt']) || $value['attempt'] < 1)) { + throw self::invalid(); + } + $validSlot = match ($value['kind'].'.'.($value['operation'] ?? '')) { + 'activity.complete' => $value['slot'] === ['result'], + 'activity.fail' => $value['slot'] === ['failure', 'details'], + 'query.complete' => $value['slot'] === ['result_envelope'], + 'workflow.complete' => self::workflowSlot($value['slot']), + default => false, + }; + if (! $validSlot) { + throw self::invalid(); + } + + return new self($value['kind'], $value['task_id'], $value['attempt'], + $value['lease_owner'], $value['operation'], $value['slot']); + } + + public function scope(string $namespace): string + { + // Complete/fail and every payload slot share one allowance for this lease. + return hash('sha256', json_encode([$namespace, $this->kind, $this->taskId, + $this->attempt, $this->leaseOwner], JSON_THROW_ON_ERROR)); + } + + public function slotIdentity(): string + { + return hash('sha256', json_encode([$this->operation, $this->slot], JSON_THROW_ON_ERROR)); + } + + public function toArray(): array + { + return ['schema' => self::SCHEMA, 'kind' => $this->kind, 'task_id' => $this->taskId, + 'attempt' => $this->attempt, 'lease_owner' => $this->leaseOwner, + 'operation' => $this->operation, 'slot' => $this->slot]; + } + + private static function identifier(mixed $value): bool + { + return is_string($value) && $value !== '' && strlen($value) <= 255 + && preg_match('/[\x00-\x1f\x7f]/', $value) === 0; + } + + private static function workflowSlot(array $slot): bool + { + if (($slot[0] ?? null) !== 'commands' || ! is_int($slot[1] ?? null) || $slot[1] < 0) { + return false; + } + if (count($slot) === 3) { + return in_array($slot[2], ['arguments', 'entries', 'request_payload', 'result'], true); + } + if (count($slot) === 4) { + return array_slice($slot, 2) === ['exception', 'details']; + } + + return count($slot) === 6 && $slot[2] === 'workflow_stream' && $slot[3] === 'items' + && is_int($slot[4]) && $slot[4] >= 0 && $slot[5] === 'payload'; + } + + private static function invalid(): RuntimeExternalPayloadException + { + return new RuntimeExternalPayloadException('external_payload_completion_invalid', 422, false, + 'A canonical worker completion context is required.'); + } +} diff --git a/app/Support/RuntimePayloadCompletionLease.php b/app/Support/RuntimePayloadCompletionLease.php new file mode 100644 index 00000000..2d20e4ed --- /dev/null +++ b/app/Support/RuntimePayloadCompletionLease.php @@ -0,0 +1,119 @@ +where('namespace', $namespace) + ->where('worker_id', $context->leaseOwner)->first(); + if (! $worker instanceof WorkerRegistration || ! WorkerPollFence::isFresh($worker)) { + throw self::rejected(); + } + + $expires = match ($context->kind) { + 'activity' => $this->activity($namespace, $context), + 'workflow' => $this->workflow($namespace, $context), + 'query' => $this->query($namespace, $context), + }; + if (! is_string($expires) || $expires === '') { + throw self::rejected(); + } + try { + $expiresAt = CarbonImmutable::parse($expires); + } catch (\Exception) { + throw self::rejected(); + } + if ($expiresAt->lte(now())) { + throw self::rejected(); + } + + return $expiresAt; + } + + public function mayResume(string $namespace, RuntimePayloadCompletionContext $context): bool + { + // An expired activity can still be renewed before lease recovery. Keep its + // budget until its identity is replaced or closed, not merely until TTL. + if ($context->kind === 'query') { + $task = $this->queries->task($context->taskId); + + return ($task['namespace'] ?? null) === $namespace && ($task['status'] ?? null) === 'leased' + && ($task['lease_owner'] ?? null) === $context->leaseOwner + && ($task['attempt_count'] ?? null) === $context->attempt; + } + $task = NamespaceWorkflowScope::task($namespace, $context->taskId); + if ($task === null || $task->status->value !== 'leased' || $task->lease_owner !== $context->leaseOwner) { + return false; + } + if ($context->kind === 'workflow') { + return $task->attempt_count === $context->attempt; + } + $status = $this->activities->status($context->attempt); + + return ($status['can_continue'] ?? false) === true + && ($status['workflow_task_id'] ?? null) === $context->taskId + && ($status['lease_owner'] ?? null) === $context->leaseOwner; + } + + private function activity(string $namespace, RuntimePayloadCompletionContext $context): ?string + { + $task = NamespaceWorkflowScope::task($namespace, $context->taskId); + if ($task === null || $task->task_type !== TaskType::Activity || $task->lease_owner !== $context->leaseOwner + || $task->lease_expires_at === null || $task->lease_expires_at->lte(now())) { + throw self::rejected(); + } + // The bridge checks the current attempt, task and run without renewing the lease. + $status = $this->activities->status($context->attempt); + if (($status['can_continue'] ?? false) !== true + || ($status['workflow_task_id'] ?? null) !== $context->taskId + || ($status['lease_owner'] ?? null) !== $context->leaseOwner) { + throw self::rejected(); + } + + return $status['lease_expires_at'] ?? null; + } + + private function workflow(string $namespace, RuntimePayloadCompletionContext $context): ?string + { + $guard = $this->workflows->guard( + fn (string $ns, string $id) => NamespaceWorkflowScope::task($ns, $id), + $namespace, $context->taskId, $context->attempt, $context->leaseOwner, + ); + if (! $guard['valid'] || $guard['task']?->task_type !== TaskType::Workflow || in_array($guard['status']['run_status'] ?? null, + ['completed', 'failed', 'cancelled', 'terminated'], true)) { + throw self::rejected(); + } + + return $guard['status']['lease_expires_at'] ?? null; + } + + private function query(string $namespace, RuntimePayloadCompletionContext $context): ?string + { + if ($this->queries->guardCompletion($namespace, $context->taskId, + $context->leaseOwner, $context->attempt) !== null) { + throw self::rejected(); + } + + return $this->queries->task($context->taskId)['lease_expires_at'] ?? null; + } + + public static function rejected(): RuntimeExternalPayloadException + { + return new RuntimeExternalPayloadException('external_payload_completion_lease_rejected', 409, false, + 'The payload upload does not belong to a current worker completion lease.'); + } +} diff --git a/app/Support/RuntimePayloadCompletionUploads.php b/app/Support/RuntimePayloadCompletionUploads.php new file mode 100644 index 00000000..ff2cc9df --- /dev/null +++ b/app/Support/RuntimePayloadCompletionUploads.php @@ -0,0 +1,147 @@ +requireWritable(); + if ($this->reference($namespace, $context, $sha256, $size) === null) { + $this->leases->expiresAt($namespace, $context); + } + } + + public function upload( + string $namespace, RuntimePayloadCompletionContext $context, ExternalPayloadStream $data, + string $codec, string $sha256, RuntimeExternalPayloadRegistry $registry, + ): array { + $this->requireWritable(); + try { + PayloadCodecContract::canonicalize($codec); + } catch (\InvalidArgumentException $exception) { + throw new RuntimeExternalPayloadException('external_payload_unsupported', 422, false, $exception->getMessage()); + } + if (! hash_equals($data->sha256, $sha256)) { + throw new RuntimeExternalPayloadException('external_payload_integrity_mismatch', 422, false, + 'Declared external payload integrity metadata does not match the uploaded bytes.'); + } + if (($reference = $this->reference($namespace, $context, $sha256, $data->sizeBytes)) !== null) { + // Lost upload/completion responses may be reconciled after the lease closes. + // This path never creates, rewrites, retains or extends an object. + return $reference; + } + + $expiresAt = $this->leases->expiresAt($namespace, $context); + $scope = $context->scope($namespace); + $this->locks->transaction($scope, function () use ($namespace, $context, $sha256, $data, $expiresAt, $scope): void { + $this->requireWritable(); + $budget = RuntimePayloadCompletionBudget::query()->find($scope) + ?? new RuntimePayloadCompletionBudget(['id' => $scope, 'namespace' => $namespace, + 'context' => $context->toArray(), 'slots' => [], 'objects' => []]); + $slots = $budget->slots; + $objects = $budget->objects; + $slot = $context->slotIdentity(); + $identity = ['sha256' => $sha256, 'size_bytes' => $data->sizeBytes]; + if (isset($slots[$slot]) && array_intersect_key($slots[$slot], $identity) !== $identity) { + throw new RuntimeExternalPayloadException('external_payload_completion_conflict', 409, false, + 'A completion payload slot is already bound to different bytes.'); + } + if (! isset($slots[$slot]) && count($slots) >= self::MAX_SLOTS + || ! isset($objects[$sha256]) && $data->sizeBytes > self::maxBytes() - array_sum($objects)) { + throw new RuntimeExternalPayloadException('external_payload_completion_budget_exhausted', 429, true, + 'The bounded payload allowance for this completion lease is exhausted.', retryAfterSeconds: 5); + } + $slots[$slot] ??= $identity; + $objects[$sha256] = $data->sizeBytes; + $budget->forceFill(['slots' => $slots, 'objects' => $objects, + 'expires_at' => $expiresAt->addSeconds($this->retryRetention())])->save(); + }); + + $reference = $registry->upload($namespace, $data, $codec, $sha256, function () use ($namespace, $context): void { + $this->requireWritable(); + $this->leases->expiresAt($namespace, $context); + }); + $this->locks->transaction($scope, function () use ($scope, $context, $reference): void { + $this->requireWritable(); + $budget = RuntimePayloadCompletionBudget::query()->findOrFail($scope); + $slots = $budget->slots; + $slots[$context->slotIdentity()]['reference'] = $reference; + $budget->forceFill(['slots' => $slots])->save(); + }); + + return $reference; + } + + public static function maxBytes(): int + { + return max(1, (int) (config('server.external_payload_transport.completion_max_bytes') + ?? config('server.external_payload_transport.max_payload_bytes'))); + } + + public function cleanup(?string $namespace, int $limit, CarbonInterface $cutoff): void + { + $candidates = RuntimePayloadCompletionBudget::query()->where('expires_at', '<=', $cutoff) + ->when($namespace !== null, fn ($query) => $query->where('namespace', $namespace)) + ->orderBy('expires_at')->limit($limit)->get(); + foreach ($candidates as $candidate) { + $context = RuntimePayloadCompletionContext::parse(json_encode($candidate->context, JSON_THROW_ON_ERROR)); + $this->locks->transaction($candidate->id, function () use ($candidate, $context, $cutoff): void { + $this->pressure->requireNewWork(); + $row = RuntimePayloadCompletionBudget::query()->find($candidate->id); + if ($row === null || $row->expires_at->gt($cutoff)) { + return; + } + // A heartbeat may keep an old lease alive: never reset its allowance. + if ($this->leases->mayResume($candidate->namespace, $context)) { + $row->forceFill(['expires_at' => now()->addSeconds($this->retryRetention())])->save(); + } else { + $row->delete(); + } + }); + } + } + + private function reference(string $namespace, RuntimePayloadCompletionContext $context, string $sha256, int $size): ?array + { + $budget = RuntimePayloadCompletionBudget::query()->find($context->scope($namespace)); + $slot = $budget?->slots[$context->slotIdentity()] ?? null; + $reference = $slot['reference'] ?? null; + if (! is_array($reference) || ($slot['sha256'] ?? null) !== $sha256 || ($slot['size_bytes'] ?? null) !== $size) { + return null; + } + $row = RuntimeExternalPayload::query()->whereKey($reference['reference_id'])->where('namespace', $namespace) + ->where('sha256', $sha256)->where('size_bytes', $size)->where('codec', 'avro') + ->where('upload_status', RuntimeExternalPayload::UPLOAD_READY)->first(); + if ($row === null || ($row->retained_at === null && ($row->expires_at === null || $row->expires_at->lte(now())))) { + return null; + } + + return $reference; + } + + private function requireWritable(): void + { + $snapshot = $this->pressure->snapshot(); + if (! in_array($snapshot['state'], ['normal', 'draining'], true)) { + throw new StorageAdmissionPaused($snapshot); + } + } + + private function retryRetention(): int + { + return max(1, (int) config('server.external_payload_transport.abandoned_upload_expiry_seconds')); + } +} diff --git a/config/server.php b/config/server.php index f0119516..8b6be75e 100644 --- a/config/server.php +++ b/config/server.php @@ -487,6 +487,11 @@ 'external_payload_transport' => [ 's3_disk' => 'external-payload-s3', + 'completion_max_bytes' => EnvAuditor::env( + 'DW_EXTERNAL_PAYLOAD_COMPLETION_MAX_BYTES', + 'WORKFLOW_SERVER_EXTERNAL_PAYLOAD_COMPLETION_MAX_BYTES', + null, + ), 'max_payload_bytes' => (int) EnvAuditor::env( 'DW_EXTERNAL_PAYLOAD_MAX_BYTES', 'WORKFLOW_SERVER_EXTERNAL_PAYLOAD_MAX_BYTES', diff --git a/database/migrations/2026_09_08_000000_create_runtime_payload_completion_budgets.php b/database/migrations/2026_09_08_000000_create_runtime_payload_completion_budgets.php new file mode 100644 index 00000000..08e3fe97 --- /dev/null +++ b/database/migrations/2026_09_08_000000_create_runtime_payload_completion_budgets.php @@ -0,0 +1,26 @@ +char('id', 64)->primary(); + $table->string('namespace', 128)->index(); + $table->json('context'); + $table->json('slots'); + $table->json('objects'); + $table->timestamp('expires_at')->index(); + $table->timestamps(); + }); + } + + public function down(): void + { + Schema::dropIfExists('runtime_payload_completion_budgets'); + } +}; diff --git a/docs/storage-admission.md b/docs/storage-admission.md index d193ab9d..af602046 100644 --- a/docs/storage-admission.md +++ b/docs/storage-admission.md @@ -54,6 +54,40 @@ history pruning, and payload cleanup do not start a new pass under pressure. ## Deployment Requirements +### External Completion Payloads + +When discovery advertises `upload.completion_context`, a worker may retry a +draining upload refusal with `X-Durable-Workflow-Payload-Completion`. The header +is a JSON object, at most 4096 bytes, containing exactly: + +```json +{"schema":"durable-workflow.v2.payload-completion-context.v1","kind":"activity","task_id":"task-id","attempt":"activity-attempt-id","lease_owner":"worker-id","operation":"complete","slot":["result"]} +``` + +`activity` uses its attempt ID; `workflow` and `query` use their positive integer +attempt counter. The context must describe the current namespace-scoped lease. +An activity allows `complete`/`["result"]` or `fail`/`["failure","details"]`; +a query allows `complete`/`["result_envelope"]`. Workflow completion slots are +`["commands", index, field]` for `arguments`, `entries`, `request_payload`, or +`result`, plus `exception/details` and `workflow_stream/items/index/payload`. +Indices are non-negative integers, not numeric strings. + +This is not a general upload permission. The worker role and current lease are +checked before reading and before committing new bytes. Every slot is immutable +within a lease. All its slots and complete/fail outcomes share at most 128 slots +and `DW_EXTERNAL_PAYLOAD_COMPLETION_MAX_BYTES` distinct bytes, defaulting to the +ordinary maximum external payload size. Matching retries do not consume another +allowance. A recorded ready reference can be returned without a new write after +the lease closes. Expired references cannot be recreated without a current lease. +Existing namespace quotas still apply, and fenced/stale admission refuses uploads. + +Normal uploads without this header keep their existing behavior. SDKs must not +attach it to workflow starts, signals, updates, or other new input. The allowance +does not reserve physical disk space or promise that arbitrarily many leased +results fit; operators still need the headroom qualification below. + +### Physical Reserve + This is cooperative admission, **not a database write fence or an exact byte reservation system**. A request, queue job, or maintenance pass already in progress can still write. A storage transition can race a successful check. diff --git a/resources/platform-protocol-specs/external-payload-transport.openapi.yaml b/resources/platform-protocol-specs/external-payload-transport.openapi.yaml index 4060c391..4f874b14 100644 --- a/resources/platform-protocol-specs/external-payload-transport.openapi.yaml +++ b/resources/platform-protocol-specs/external-payload-transport.openapi.yaml @@ -32,6 +32,7 @@ paths: - $ref: "#/components/parameters/PayloadCodecHeader" - $ref: "#/components/parameters/PayloadSizeHeader" - $ref: "#/components/parameters/PayloadSha256Header" + - $ref: "#/components/parameters/PayloadCompletionHeader" requestBody: required: true content: @@ -51,6 +52,7 @@ paths: "401": { $ref: "#/components/responses/Unauthorized" } "403": { $ref: "#/components/responses/Unauthorized" } "404": { $ref: "#/components/responses/NotFound" } + "409": { $ref: "#/components/responses/InvalidReference" } "413": { $ref: "#/components/responses/Oversized" } "415": { $ref: "#/components/responses/Unsupported" } "422": { $ref: "#/components/responses/InvalidReference" } @@ -122,6 +124,19 @@ components: in: header required: true schema: { type: string, pattern: '^[a-f0-9]{64}$' } + PayloadCompletionHeader: + name: X-Durable-Workflow-Payload-Completion + in: header + required: false + description: > + Optional JSON worker completion context, at most 4096 UTF-8 bytes. + Requires the worker role and a current namespace-scoped task lease for + new writes. Use only after a draining refusal when discovery advertises + upload.completion_context. Contains exactly schema, kind, task_id, + attempt, lease_owner, operation and slot; see docs/storage-admission.md. + Fenced or stale admission remains closed. Recorded matching references + can be reconciled read-only after the lease closes. + schema: { type: string, maxLength: 4096, contentMediaType: application/json } headers: PayloadCodec: schema: { type: string, const: avro } @@ -182,6 +197,10 @@ components: - external_payload_namespace_bytes_exhausted - external_payload_namespace_objects_exhausted - external_payload_namespace_quota_unavailable + - external_payload_completion_invalid + - external_payload_completion_lease_rejected + - external_payload_completion_conflict + - external_payload_completion_budget_exhausted message: { type: string, minLength: 1 } retryable: { type: boolean } status: { type: integer } diff --git a/tests/Feature/RuntimePayloadCompletionUploadsTest.php b/tests/Feature/RuntimePayloadCompletionUploadsTest.php new file mode 100644 index 00000000..07e6ccaf --- /dev/null +++ b/tests/Feature/RuntimePayloadCompletionUploadsTest.php @@ -0,0 +1,209 @@ +directory = storage_path('framework/testing/completion-uploads'); + File::deleteDirectory($this->directory); + foreach (['default', 'other'] as $namespace) { + $this->createNamespace($namespace); + WorkflowNamespace::query()->where('name', $namespace)->update(['external_payload_storage' => [ + 'driver' => 'local', 'enabled' => true, 'threshold_bytes' => 32, + 'config' => ['uri' => 'file://'.$this->directory.'/'.$namespace], + ]]); + } + $this->registerWorker('worker', 'queue'); + config(['server.external_payload_transport.max_payload_bytes' => 4096, + 'server.external_payload_transport.completion_max_bytes' => null]); + $this->configureStoragePressure(); + } + + protected function tearDown(): void + { + RuntimePayloadCompletionBudget::flushEventListeners(); + $this->removeStoragePressure(); + File::deleteDirectory($this->directory); + parent::tearDown(); + } + + public function test_owned_activity_result_can_upload_complete_and_reconcile_after_lease_closes(): void + { + $context = $this->activity(); + $payload = Serializer::serializeWithCodec('avro', str_repeat('result', 200)); + $this->observeStoragePressure('draining'); + $reference = $this->upload($payload, $context)->assertCreated()->json('reference'); + $this->upload($payload, $context)->assertCreated()->assertJsonPath('reference', $reference); + $this->postJson('/api/worker/activity-tasks/'.$context['task_id'].'/complete', [ + 'activity_attempt_id' => $context['attempt'], 'lease_owner' => 'worker', + 'result' => ['codec' => 'avro', 'external_payload' => $reference], + ], $this->workerHeaders())->assertOk(); + $before = RuntimeExternalPayload::query()->sole()->getRawOriginal(); + $budgetBefore = RuntimePayloadCompletionBudget::query()->sole()->getRawOriginal(); + $this->upload($payload, $context)->assertCreated()->assertJsonPath('reference', $reference); + $this->assertSame($before, RuntimeExternalPayload::query()->sole()->getRawOriginal()); + $this->assertSame($budgetBefore, RuntimePayloadCompletionBudget::query()->sole()->getRawOriginal()); + $this->getJson('/api/activities/activity', $this->apiHeaders())->assertOk() + ->assertJsonPath('activity_status', 'completed') + ->assertJsonPath('result.external_payload.reference_id', $reference['reference_id']); + $this->call('GET', '/api/external-payloads/v1/'.$reference['reference_id'], [], [], [], [ + 'HTTP_X_NAMESPACE' => 'default', 'HTTP_X_DURABLE_WORKFLOW_PAYLOAD_CODEC' => 'avro', + 'HTTP_X_DURABLE_WORKFLOW_PAYLOAD_SIZE' => (string) strlen($payload), + 'HTTP_X_DURABLE_WORKFLOW_PAYLOAD_SHA256' => hash('sha256', $payload), + ])->assertOk()->assertStreamedContent($payload); + } + + #[DataProvider('wrongContexts')] + public function test_foreign_or_unrelated_context_rejects_without_reserving_storage(array $changes, string $namespace): void + { + $context = array_replace($this->activity(), $changes); + $this->observeStoragePressure('draining'); + $this->upload('bytes', $context, $namespace)->assertStatus(409) + ->assertJsonPath('reason', 'external_payload_completion_lease_rejected'); + $this->assertDatabaseCount('runtime_payload_completion_budgets', 0); + $this->assertDatabaseCount('runtime_external_payloads', 0); + } + + public static function wrongContexts(): array + { + return [[['task_id' => 'missing'], 'default'], [['attempt' => 'missing'], 'default'], + [['lease_owner' => 'another-worker'], 'default'], [[], 'other']]; + } + + public function test_expired_lease_cannot_create_an_upload(): void + { + $context = $this->activity(); + WorkflowTask::query()->whereKey($context['task_id'])->update(['lease_expires_at' => now()->subSecond()]); + $this->observeStoragePressure('draining'); + $this->upload('bytes', $context)->assertStatus(409); + $this->assertDatabaseCount('runtime_payload_completion_budgets', 0); + $this->assertDatabaseCount('runtime_external_payloads', 0); + } + + public function test_one_slot_cannot_rotate_content_and_complete_fail_share_the_budget(): void + { + $context = $this->activity(); + config(['server.external_payload_transport.completion_max_bytes' => 6]); + $this->observeStoragePressure('draining'); + $this->upload('first', $context)->assertCreated(); + $this->upload('other', $context)->assertStatus(409)->assertJsonPath('reason', 'external_payload_completion_conflict'); + $context['operation'] = 'fail'; + $context['slot'] = ['failure', 'details']; + $this->upload('other', $context)->assertStatus(429)->assertJsonPath('reason', 'external_payload_completion_budget_exhausted'); + $this->upload('first', $context)->assertCreated(); + $this->assertDatabaseCount('runtime_external_payloads', 1); + $this->assertDatabaseCount('runtime_payload_completion_budgets', 1); + } + + #[DataProvider('closedAdmission')] + public function test_fenced_or_stale_state_does_not_admit_even_an_owned_upload(string $state): void + { + $context = $this->activity(); + $this->observeStoragePressure($state === 'stale' ? 'normal' : $state); + if ($state === 'stale') { + $this->travel(120)->seconds(); + } + $this->upload('bytes', $context)->assertStatus(503); + $this->assertDatabaseCount('runtime_external_payloads', 0); + $this->assertDatabaseCount('runtime_payload_completion_budgets', 0); + } + + public static function closedAdmission(): array + { + return [['fenced'], ['stale']]; + } + + public function test_lease_is_rechecked_after_reservation_and_before_object_write(): void + { + $context = $this->activity(); + $this->observeStoragePressure('draining'); + RuntimePayloadCompletionBudget::created(function () use ($context): void { + WorkflowTask::query()->whereKey($context['task_id'])->update(['lease_expires_at' => now()->subSecond()]); + }); + $this->upload('bytes', $context)->assertStatus(409); + $this->assertDatabaseCount('runtime_external_payloads', 0); + } + + public function test_hard_fence_is_rechecked_before_object_write(): void + { + $context = $this->activity(); + $this->observeStoragePressure('draining'); + RuntimePayloadCompletionBudget::created(fn () => $this->observeStoragePressure('fenced')); + $this->upload('bytes', $context)->assertStatus(503); + $this->assertDatabaseCount('runtime_external_payloads', 0); + } + + public function test_namespace_quota_still_applies_to_owned_uploads(): void + { + $context = $this->activity(); + config(['server.external_payload_transport.hard_max_bytes_per_namespace' => 2]); + $this->observeStoragePressure('draining'); + $this->upload('bytes', $context)->assertStatus(429)->assertJsonPath('reason', 'external_payload_namespace_bytes_exhausted'); + $this->assertDatabaseCount('runtime_external_payloads', 0); + } + + public function test_cleanup_does_not_reset_an_expired_but_renewable_activity_budget(): void + { + $context = $this->activity(); + $this->observeStoragePressure('draining'); + $this->upload('bytes', $context)->assertCreated(); + $this->observeStoragePressure('normal'); + RuntimePayloadCompletionBudget::query()->update(['expires_at' => now()->subHour()]); + WorkflowTask::query()->whereKey($context['task_id'])->update(['lease_expires_at' => now()->subMinute()]); + app(RuntimeExternalPayloadCleanup::class)->runPass(); + $this->assertDatabaseCount('runtime_payload_completion_budgets', 1); + WorkflowTask::query()->whereKey($context['task_id'])->update(['status' => 'completed']); + RuntimePayloadCompletionBudget::query()->update(['expires_at' => now()->subHour()]); + app(RuntimeExternalPayloadCleanup::class)->runPass(); + $this->assertDatabaseCount('runtime_payload_completion_budgets', 0); + } + + private function activity(): array + { + $this->postJson('/api/activities', ['activity_id' => 'activity', + 'activity_type' => 'tests.external-greeting-activity', 'task_queue' => 'queue', 'input' => []], $this->apiHeaders())->assertCreated(); + $task = $this->postJson('/api/worker/activity-tasks/poll', ['worker_id' => 'worker', 'task_queue' => 'queue'], + $this->workerHeaders())->assertOk()->json('task'); + $this->assertIsArray($task); + + return ['schema' => RuntimePayloadCompletionContext::SCHEMA, 'kind' => 'activity', + 'task_id' => $task['task_id'], 'attempt' => $task['activity_attempt_id'], + 'lease_owner' => 'worker', 'operation' => 'complete', 'slot' => ['result']]; + } + + private function upload(string $bytes, array $context, string $namespace = 'default'): TestResponse + { + return $this->call('POST', '/api/external-payloads/v1', [], [], [], [ + 'CONTENT_TYPE' => 'application/octet-stream', 'HTTP_X_NAMESPACE' => $namespace, + 'HTTP_X_DURABLE_WORKFLOW_PAYLOAD_CODEC' => 'avro', 'HTTP_X_DURABLE_WORKFLOW_PAYLOAD_SIZE' => (string) strlen($bytes), + 'HTTP_X_DURABLE_WORKFLOW_PAYLOAD_SHA256' => hash('sha256', $bytes), + 'HTTP_X_DURABLE_WORKFLOW_PAYLOAD_COMPLETION' => json_encode($context, JSON_THROW_ON_ERROR), + ], $bytes); + } +} diff --git a/tests/Unit/RuntimePayloadCompletionContextTest.php b/tests/Unit/RuntimePayloadCompletionContextTest.php new file mode 100644 index 00000000..868eec12 --- /dev/null +++ b/tests/Unit/RuntimePayloadCompletionContextTest.php @@ -0,0 +1,88 @@ +value($kind, $operation, $slot); + $context = RuntimePayloadCompletionContext::parse(json_encode($value, JSON_THROW_ON_ERROR)); + $this->assertSame($kind, $context->kind); + $this->assertSame($operation, $context->operation); + $this->assertSame($slot, $context->slot); + $this->assertSame($context->scope('default'), + RuntimePayloadCompletionContext::parse(json_encode(array_reverse($value, true), JSON_THROW_ON_ERROR))->scope('default')); + $this->assertNotSame($context->scope('default'), $context->scope('other')); + } + + public static function validSlots(): array + { + return [ + ['activity', 'complete', ['result']], + ['activity', 'fail', ['failure', 'details']], + ['query', 'complete', ['result_envelope']], + ['workflow', 'complete', ['commands', 0, 'arguments']], + ['workflow', 'complete', ['commands', 1, 'entries']], + ['workflow', 'complete', ['commands', 2, 'request_payload']], + ['workflow', 'complete', ['commands', 3, 'result']], + ['workflow', 'complete', ['commands', 4, 'exception', 'details']], + ['workflow', 'complete', ['commands', 5, 'workflow_stream', 'items', 0, 'payload']], + ]; + } + + public function test_all_slots_and_outcomes_for_the_same_lease_share_one_budget(): void + { + $complete = RuntimePayloadCompletionContext::parse(json_encode($this->value(), JSON_THROW_ON_ERROR)); + $fail = RuntimePayloadCompletionContext::parse(json_encode($this->value('activity', 'fail', ['failure', 'details']), JSON_THROW_ON_ERROR)); + $this->assertSame($complete->scope('default'), $fail->scope('default')); + $this->assertNotSame($complete->slotIdentity(), $fail->slotIdentity()); + foreach (['task_id', 'attempt', 'lease_owner'] as $key) { + $other = $this->value(); + $other[$key] .= '-other'; + $this->assertNotSame($complete->scope('default'), + RuntimePayloadCompletionContext::parse(json_encode($other, JSON_THROW_ON_ERROR))->scope('default')); + } + } + + #[DataProvider('invalidContexts')] + public function test_malformed_or_unrelated_context_cannot_authorize_writes(array $changes): void + { + $this->expectException(RuntimeExternalPayloadException::class); + RuntimePayloadCompletionContext::parse(json_encode(array_replace($this->value(), $changes), JSON_THROW_ON_ERROR)); + } + + public static function invalidContexts(): array + { + return array_map(static fn (array $changes): array => [$changes], [ + ['schema' => 'unknown'], ['extra' => true], ['kind' => 'client'], ['operation' => 'start'], + ['operation' => []], ['operation' => null], ['slot' => null], ['task_id' => []], + ['task_id' => ''], ['task_id' => str_repeat('a', 256)], ['lease_owner' => "worker\n"], + ['attempt' => 1], ['slot' => ['input']], ['slot' => ['failure', 'details']], + ['kind' => 'workflow', 'attempt' => '1', 'slot' => ['commands', 0, 'result']], + ['kind' => 'workflow', 'attempt' => 0, 'slot' => ['commands', 0, 'result']], + ['kind' => 'workflow', 'attempt' => 1, 'slot' => ['commands', '0', 'result']], + ['kind' => 'workflow', 'attempt' => 1, 'slot' => ['commands', -1, 'result']], + ['kind' => 'query', 'attempt' => 1, 'slot' => ['result']], + ]); + } + + public function test_oversized_header_is_rejected_before_json_decode(): void + { + $this->expectException(RuntimeExternalPayloadException::class); + RuntimePayloadCompletionContext::parse(str_repeat(' ', 4097)); + } + + private function value(string $kind = 'activity', string $operation = 'complete', array $slot = ['result']): array + { + return ['schema' => RuntimePayloadCompletionContext::SCHEMA, 'kind' => $kind, + 'task_id' => 'task', 'attempt' => $kind === 'activity' ? 'attempt' : 1, + 'lease_owner' => 'worker', 'operation' => $operation, 'slot' => $slot]; + } +} From 62d2e3ca36b3a03036422650cc537a725905c789 Mon Sep 17 00:00:00 2001 From: Durable Workflow Date: Tue, 8 Sep 2026 16:36:41 +0000 Subject: [PATCH 2/5] Cover workflow and query drain completion ownership --- .../RuntimePayloadCompletionUploadsTest.php | 97 +++++++++++++++++++ ...timeExternalPayloadOpenApiContractTest.php | 4 + 2 files changed, 101 insertions(+) diff --git a/tests/Feature/RuntimePayloadCompletionUploadsTest.php b/tests/Feature/RuntimePayloadCompletionUploadsTest.php index 07e6ccaf..2720dd30 100644 --- a/tests/Feature/RuntimePayloadCompletionUploadsTest.php +++ b/tests/Feature/RuntimePayloadCompletionUploadsTest.php @@ -4,9 +4,11 @@ use App\Models\RuntimeExternalPayload; use App\Models\RuntimePayloadCompletionBudget; +use App\Models\WorkerRegistration; use App\Models\WorkflowNamespace; use App\Support\RuntimeExternalPayloadCleanup; use App\Support\RuntimePayloadCompletionContext; +use App\Support\WorkflowQueryTaskBroker; use Illuminate\Foundation\Testing\RefreshDatabase; use Illuminate\Support\Facades\File; use Illuminate\Support\Facades\Queue; @@ -16,6 +18,7 @@ use Tests\Feature\Concerns\StoragePressureFixture; use Tests\TestCase; use Workflow\Serializers\Serializer; +use Workflow\V2\Models\WorkflowRun; use Workflow\V2\Models\WorkflowTask; class RuntimePayloadCompletionUploadsTest extends TestCase @@ -79,6 +82,86 @@ public function test_owned_activity_result_can_upload_complete_and_reconcile_aft ])->assertOk()->assertStreamedContent($payload); } + public function test_workflow_completion_upload_can_be_committed_during_draining(): void + { + $context = $this->workflow(); + $payload = Serializer::serializeWithCodec('avro', ['result' => str_repeat('x', 100)]); + $this->observeStoragePressure('draining'); + $reference = $this->upload($payload, $context)->assertCreated()->json('reference'); + $this->postJson('/api/worker/workflow-tasks/'.$context['task_id'].'/complete', [ + 'lease_owner' => 'worker', 'workflow_task_attempt' => $context['attempt'], + 'commands' => [['type' => 'complete_workflow', 'sequence' => 1, + 'result' => ['codec' => 'avro', 'external_payload' => $reference]]], + ], $this->workerHeaders())->assertOk(); + $this->assertNotNull(RuntimeExternalPayload::query()->sole()->retained_at); + $this->assertSame('completed', WorkflowRun::query()->sole()->status->value); + $this->upload($payload, $context)->assertCreated()->assertJsonPath('reference', $reference); + } + + public function test_query_completion_upload_uses_its_own_current_lease(): void + { + $this->postJson('/api/workflows', ['workflow_id' => 'query-workflow', + 'workflow_type' => 'tests.external-greeting-workflow', 'task_queue' => 'queue', 'input' => []], + $this->apiHeaders())->assertCreated(); + WorkerRegistration::query()->where('worker_id', 'worker')->update(['capabilities' => ['query_tasks']]); + $broker = app(WorkflowQueryTaskBroker::class); + $broker->enqueue('default', WorkflowRun::query()->sole(), 'status', [ + 'codec' => 'avro', 'blob' => Serializer::serializeWithCodec('avro', []), + ]); + $task = $this->postJson('/api/worker/query-tasks/poll', ['worker_id' => 'worker', 'task_queue' => 'queue'], + $this->workerHeaders())->assertOk()->json('task'); + $this->assertIsArray($task); + $context = ['schema' => RuntimePayloadCompletionContext::SCHEMA, 'kind' => 'query', + 'task_id' => $task['query_task_id'], 'attempt' => $task['query_task_attempt'], + 'lease_owner' => 'worker', 'operation' => 'complete', 'slot' => ['result_envelope']]; + $this->observeStoragePressure('draining'); + $wrong = $context; + $wrong['attempt']++; + $this->upload('bytes', $wrong)->assertStatus(409); + $payload = Serializer::serializeWithCodec('avro', ['status' => str_repeat('ready', 50)]); + $reference = $this->upload($payload, $context)->assertCreated()->json('reference'); + $this->postJson('/api/worker/query-tasks/'.$context['task_id'].'/complete', [ + 'lease_owner' => 'worker', 'query_task_attempt' => $context['attempt'], + 'result_envelope' => ['codec' => 'avro', 'external_payload' => $reference], + ], $this->workerHeaders())->assertOk()->assertJsonPath('outcome', 'completed'); + $this->upload($payload, $context)->assertCreated()->assertJsonPath('reference', $reference); + } + + public function test_activity_cannot_claim_another_budget_as_a_workflow_task(): void + { + $context = $this->activity(); + $context['kind'] = 'workflow'; + $context['attempt'] = 1; + $context['slot'] = ['commands', 0, 'result']; + $this->observeStoragePressure('draining'); + $this->upload('bytes', $context)->assertStatus(409); + $this->assertDatabaseCount('runtime_payload_completion_budgets', 0); + } + + public function test_workflow_slots_are_bounded_even_when_the_bytes_are_deduplicated(): void + { + $context = $this->workflow(); + $this->observeStoragePressure('draining'); + for ($index = 0; $index < 128; $index++) { + $context['slot'] = ['commands', $index, 'arguments']; + $this->upload('same', $context)->assertCreated(); + } + $context['slot'] = ['commands', 128, 'arguments']; + $this->upload('same', $context)->assertStatus(429) + ->assertJsonPath('reason', 'external_payload_completion_budget_exhausted'); + $this->assertDatabaseCount('runtime_external_payloads', 1); + $this->assertCount(128, RuntimePayloadCompletionBudget::query()->sole()->slots); + } + + public function test_renewed_worker_registration_does_not_authorize_a_wrong_workflow_attempt(): void + { + $context = $this->workflow(); + $context['attempt']++; + $this->observeStoragePressure('draining'); + $this->upload('bytes', $context)->assertStatus(409); + $this->assertDatabaseCount('runtime_payload_completion_budgets', 0); + } + #[DataProvider('wrongContexts')] public function test_foreign_or_unrelated_context_rejects_without_reserving_storage(array $changes, string $namespace): void { @@ -197,6 +280,20 @@ private function activity(): array 'lease_owner' => 'worker', 'operation' => 'complete', 'slot' => ['result']]; } + private function workflow(): array + { + $this->postJson('/api/workflows', ['workflow_id' => 'workflow', + 'workflow_type' => 'tests.external-greeting-workflow', 'task_queue' => 'queue', 'input' => []], + $this->apiHeaders())->assertCreated(); + $task = $this->postJson('/api/worker/workflow-tasks/poll', ['worker_id' => 'worker', 'task_queue' => 'queue'], + $this->workerHeaders())->assertOk()->json('task'); + $this->assertIsArray($task); + + return ['schema' => RuntimePayloadCompletionContext::SCHEMA, 'kind' => 'workflow', + 'task_id' => $task['task_id'], 'attempt' => $task['workflow_task_attempt'], + 'lease_owner' => 'worker', 'operation' => 'complete', 'slot' => ['commands', 0, 'result']]; + } + private function upload(string $bytes, array $context, string $namespace = 'default'): TestResponse { return $this->call('POST', '/api/external-payloads/v1', [], [], [], [ diff --git a/tests/Unit/RuntimeExternalPayloadOpenApiContractTest.php b/tests/Unit/RuntimeExternalPayloadOpenApiContractTest.php index 130e28b4..d5e8b1c7 100644 --- a/tests/Unit/RuntimeExternalPayloadOpenApiContractTest.php +++ b/tests/Unit/RuntimeExternalPayloadOpenApiContractTest.php @@ -49,6 +49,10 @@ public function test_runtime_external_payload_openapi_is_parseable_and_complete( 'external_payload_namespace_bytes_exhausted', 'external_payload_namespace_objects_exhausted', 'external_payload_namespace_quota_unavailable', + 'external_payload_completion_invalid', + 'external_payload_completion_lease_rejected', + 'external_payload_completion_conflict', + 'external_payload_completion_budget_exhausted', ], $reasons); $this->assertSame( From 9bff92d41b5bc3cef551d608c866d9c62988aeff Mon Sep 17 00:00:00 2001 From: Durable Workflow Date: Tue, 8 Sep 2026 17:15:24 +0000 Subject: [PATCH 3/5] Preserve drain retries and qualify cold concurrent completion uploads --- .../RuntimeExternalPayloadObjectLock.php | 7 +- .../RuntimeExternalPayloadRegistry.php | 4 +- .../RuntimePayloadCompletionUploads.php | 20 ++- app/Support/StorageAdmissionPaused.php | 4 +- docs/storage-admission.md | 3 + .../RuntimePayloadCompletionProcessTest.php | 149 ++++++++++++++++++ .../RuntimePayloadCompletionUploadsTest.php | 49 +++++- .../RuntimePayloadCompletionProcess.php | 122 ++++++++++++++ 8 files changed, 342 insertions(+), 16 deletions(-) create mode 100644 tests/Feature/RuntimePayloadCompletionProcessTest.php create mode 100644 tests/Support/RuntimePayloadCompletionProcess.php diff --git a/app/Support/RuntimeExternalPayloadObjectLock.php b/app/Support/RuntimeExternalPayloadObjectLock.php index 66752921..a9ad8701 100644 --- a/app/Support/RuntimeExternalPayloadObjectLock.php +++ b/app/Support/RuntimeExternalPayloadObjectLock.php @@ -22,11 +22,14 @@ final class RuntimeExternalPayloadObjectLock * @param Closure(): TReturn $callback * @return TReturn */ - public function transaction(string $uri, Closure $callback): mixed + public function transaction(string $uri, Closure $callback, ?Closure $beforeLock = null): mixed { $bucket = hexdec(substr(hash('sha256', $uri), 0, 4)) % self::BUCKETS; - return DB::transaction(function () use ($bucket, $callback): mixed { + return DB::transaction(function () use ($bucket, $callback, $beforeLock): mixed { + // Completion ownership locks must precede payload locks, as they do + // when the engine commits an activity result. + $beforeLock?->__invoke(); $lock = DB::table('runtime_external_payload_object_locks') ->where('bucket', $bucket) ->lockForUpdate() diff --git a/app/Support/RuntimeExternalPayloadRegistry.php b/app/Support/RuntimeExternalPayloadRegistry.php index 374bd0f9..8dd2e253 100644 --- a/app/Support/RuntimeExternalPayloadRegistry.php +++ b/app/Support/RuntimeExternalPayloadRegistry.php @@ -485,7 +485,7 @@ private function store( $beforeWrite?->__invoke(); return $this->track($namespace, $uri, $codec, $sha256, $sizeBytes, false, $expiresAt, RuntimeExternalPayload::UPLOAD_WRITING); - }); + }, $beforeWrite); } catch (RuntimeExternalPayloadException|StorageAdmissionPaused $exception) { throw $exception; } catch (Throwable $exception) { @@ -545,7 +545,7 @@ private function store( $expiresAt, RuntimeExternalPayload::UPLOAD_READY, ); - }); + }, $beforeWrite); } catch (RuntimeExternalPayloadException|StorageAdmissionPaused $exception) { throw $exception; } catch (Throwable $exception) { diff --git a/app/Support/RuntimePayloadCompletionUploads.php b/app/Support/RuntimePayloadCompletionUploads.php index ff2cc9df..c7deea05 100644 --- a/app/Support/RuntimePayloadCompletionUploads.php +++ b/app/Support/RuntimePayloadCompletionUploads.php @@ -48,7 +48,7 @@ public function upload( $scope = $context->scope($namespace); $this->locks->transaction($scope, function () use ($namespace, $context, $sha256, $data, $expiresAt, $scope): void { $this->requireWritable(); - $budget = RuntimePayloadCompletionBudget::query()->find($scope) + $budget = RuntimePayloadCompletionBudget::query()->lockForUpdate()->find($scope) ?? new RuntimePayloadCompletionBudget(['id' => $scope, 'namespace' => $namespace, 'context' => $context->toArray(), 'slots' => [], 'objects' => []]); $slots = $budget->slots; @@ -61,6 +61,12 @@ public function upload( } if (! isset($slots[$slot]) && count($slots) >= self::MAX_SLOTS || ! isset($objects[$sha256]) && $data->sizeBytes > self::maxBytes() - array_sum($objects)) { + $snapshot = $this->pressure->snapshot(); + if ($snapshot['state'] === 'draining') { + // No slot or object mutation was admitted. Preserve the worker's + // existing pause/retry contract, not an activity failure. + throw new StorageAdmissionPaused($snapshot, unadmitted: true); + } throw new RuntimeExternalPayloadException('external_payload_completion_budget_exhausted', 429, true, 'The bounded payload allowance for this completion lease is exhausted.', retryAfterSeconds: 5); } @@ -76,7 +82,7 @@ public function upload( }); $this->locks->transaction($scope, function () use ($scope, $context, $reference): void { $this->requireWritable(); - $budget = RuntimePayloadCompletionBudget::query()->findOrFail($scope); + $budget = RuntimePayloadCompletionBudget::query()->lockForUpdate()->findOrFail($scope); $slots = $budget->slots; $slots[$context->slotIdentity()]['reference'] = $reference; $budget->forceFill(['slots' => $slots])->save(); @@ -98,18 +104,22 @@ public function cleanup(?string $namespace, int $limit, CarbonInterface $cutoff) ->orderBy('expires_at')->limit($limit)->get(); foreach ($candidates as $candidate) { $context = RuntimePayloadCompletionContext::parse(json_encode($candidate->context, JSON_THROW_ON_ERROR)); - $this->locks->transaction($candidate->id, function () use ($candidate, $context, $cutoff): void { + $mayResume = true; + $this->locks->transaction($candidate->id, function () use ($candidate, $cutoff, &$mayResume): void { $this->pressure->requireNewWork(); - $row = RuntimePayloadCompletionBudget::query()->find($candidate->id); + $row = RuntimePayloadCompletionBudget::query()->lockForUpdate()->find($candidate->id); if ($row === null || $row->expires_at->gt($cutoff)) { return; } // A heartbeat may keep an old lease alive: never reset its allowance. - if ($this->leases->mayResume($candidate->namespace, $context)) { + if ($mayResume) { $row->forceFill(['expires_at' => now()->addSeconds($this->retryRetention())])->save(); } else { $row->delete(); } + }, function () use ($candidate, $context, &$mayResume): void { + $this->pressure->requireNewWork(); + $mayResume = $this->leases->mayResume($candidate->namespace, $context); }); } } diff --git a/app/Support/StorageAdmissionPaused.php b/app/Support/StorageAdmissionPaused.php index 96800bd0..2d09f4af 100644 --- a/app/Support/StorageAdmissionPaused.php +++ b/app/Support/StorageAdmissionPaused.php @@ -9,13 +9,13 @@ final class StorageAdmissionPaused extends RuntimeException { - public function __construct(public readonly array $snapshot) + public function __construct(public readonly array $snapshot, private readonly bool $unadmitted = false) { parent::__construct('Storage admission paused before a new claim.'); } public function render(Request $request): Response { - return app(EnforceStorageAdmission::class)->reject($request, $this->snapshot, false); + return app(EnforceStorageAdmission::class)->reject($request, $this->snapshot, $this->unadmitted); } } diff --git a/docs/storage-admission.md b/docs/storage-admission.md index af602046..10d4a6d4 100644 --- a/docs/storage-admission.md +++ b/docs/storage-admission.md @@ -80,6 +80,9 @@ ordinary maximum external payload size. Matching retries do not consume another allowance. A recorded ready reference can be returned without a new write after the lease closes. Expired references cannot be recreated without a current lease. Existing namespace quotas still apply, and fenced/stale admission refuses uploads. +An exhausted drain allowance returns the ordinary retryable storage-pressure +refusal, so workers hold their result instead of reporting an activity failure. +After capacity recovers, retry the ordinary upload without the completion header. Normal uploads without this header keep their existing behavior. SDKs must not attach it to workflow starts, signals, updates, or other new input. The allowance diff --git a/tests/Feature/RuntimePayloadCompletionProcessTest.php b/tests/Feature/RuntimePayloadCompletionProcessTest.php new file mode 100644 index 00000000..cfe22175 --- /dev/null +++ b/tests/Feature/RuntimePayloadCompletionProcessTest.php @@ -0,0 +1,149 @@ +initialize($driver); + $this->runProbe('init', $kind); + $first = $this->probe('upload', $kind, 'alpha', '0', 'first'); + $second = $this->probe('upload', $kind, 'bravo', '0', 'second'); + $first->start(); + $second->start(); + foreach (['first' => $first, 'second' => $second] as $name => $process) { + $deadline = microtime(true) + 10; + while (! is_file($this->directory.'/'.$name.'.ready') && $process->isRunning() && microtime(true) < $deadline) { + usleep(10000); + } + self::assertFileExists($this->directory.'/'.$name.'.ready', $process->getErrorOutput()); + } + touch($this->directory.'/go'); + $a = $this->completedResult($first); + $b = $this->completedResult($second); + foreach ([$a, $b] as $response) { + if ($response['status'] === 503) { + self::assertSame('sqlite', $driver); + self::assertSame('backend_lock_pressure', $response['body']['reason']); + self::assertTrue($response['body']['retryable']); + } + } + $statuses = [$a['status'], $b['status']]; + sort($statuses); + self::assertContains($statuses, [[201, 409], [201, 503]], json_encode([$a, $b])); + $winner = $a['status'] === 201 ? 'alpha' : 'bravo'; + $loser = $winner === 'alpha' ? 'bravo' : 'alpha'; + $accepted = $a['status'] === 201 ? $a : $b; + self::assertSame($accepted, $this->runProbe('upload', $kind, $winner)); + self::assertSame(409, $this->runProbe('upload', $kind, $loser)['status']); + $exhausted = $this->runProbe('upload', $kind, $loser, '1'); + self::assertSame(503, $exhausted['status']); + self::assertSame('storage_pressure', $exhausted['body']['reason']); + self::assertFalse($exhausted['body']['request_admitted']); + $status = $this->runProbe('status', $kind); + self::assertSame(['budgets' => 1, 'slots' => 1, 'objects' => 1, + 'bytes' => $accepted['body']['reference']['size_bytes'], 'rows' => 1], $status); + self::assertSame(200, $this->runProbe('complete', $kind)['status']); + // Every operation above and below boots a new application/database connection. + self::assertSame($accepted, $this->runProbe('upload', $kind, $winner)); + self::assertSame($status, $this->runProbe('status', $kind)); + self::assertSame(409, $this->runProbe('upload', $kind, $loser, '1')['status']); + } + + protected function tearDown(): void + { + foreach ($this->processes as $process) { + if ($process->isRunning()) { + $process->stop(1); + } + } + if ($this->databaseAdmin !== null) { + $this->databaseAdmin->exec('DROP DATABASE '.$this->database); + } + if (isset($this->directory)) { + (new Filesystem)->deleteDirectory($this->directory); + } + parent::tearDown(); + } + + private function initialize(string $driver): void + { + $host = getenv('DW_TEST_COMPLETION_MYSQL_HOST'); + if ($driver === 'mysql' && ! $host) { + $this->markTestSkipped('Set DW_TEST_COMPLETION_MYSQL_HOST to a disposable MySQL test server.'); + } + $this->directory = sys_get_temp_dir().'/dw-completion-process-'.bin2hex(random_bytes(6)); + mkdir($this->directory, 0700); + $this->environment = ['APP_ENV' => 'testing', 'APP_CONFIG_CACHE' => $this->directory.'/config.php', + 'CACHE_STORE' => 'array', 'QUEUE_CONNECTION' => 'database', 'DW_AUTH_DRIVER' => 'none', + 'DW_WORKER_POLL_TIMEOUT' => '0', 'DB_URL' => false, 'DB_CONNECTION' => $driver, + 'DB_DATABASE' => $this->directory.'/database.sqlite']; + if ($driver === 'sqlite') { + touch($this->environment['DB_DATABASE']); + } else { + $this->database = 'dw_completion_'.bin2hex(random_bytes(6)); + $user = getenv('DW_TEST_COMPLETION_MYSQL_USER') ?: 'root'; + $password = getenv('DW_TEST_COMPLETION_MYSQL_PASSWORD') ?: ''; + $this->databaseAdmin = new PDO('mysql:host='.$host.';port=3306', $user, $password, + [PDO::ATTR_ERRMODE => PDO::ERRMODE_EXCEPTION]); + $this->databaseAdmin->exec('CREATE DATABASE '.$this->database); + $this->environment = array_replace($this->environment, ['DB_HOST' => $host, 'DB_PORT' => '3306', + 'DB_DATABASE' => $this->database, 'DB_USERNAME' => $user, 'DB_PASSWORD' => $password]); + } + } + + private function probe(string $action, string $kind, string $variant = 'alpha', string $slot = '0', string $barrier = ''): Process + { + $process = new Process([PHP_BINARY, 'tests/Support/RuntimePayloadCompletionProcess.php', $action, + $this->directory, $kind, $variant, $slot, $barrier], dirname(__DIR__, 2), $this->environment, + timeout: $action === 'init' ? 120 : 30); + $this->processes[] = $process; + + return $process; + } + + private function runProbe(string $action, string $kind, string $variant = 'alpha', string $slot = '0'): array + { + $process = $this->probe($action, $kind, $variant, $slot); + $process->start(); + + return $this->completedResult($process); + } + + private function completedResult(Process $process): array + { + $process->wait(); + self::assertSame(0, $process->getExitCode(), $process->getErrorOutput()); + + $result = json_decode($process->getOutput(), true, flags: JSON_THROW_ON_ERROR); + if (isset($result['body']['reference'])) { + // MySQL JSON may reorder object keys; preserve strict scalar identity. + ksort($result['body']['reference']); + } + + return $result; + } +} diff --git a/tests/Feature/RuntimePayloadCompletionUploadsTest.php b/tests/Feature/RuntimePayloadCompletionUploadsTest.php index 2720dd30..ba876afd 100644 --- a/tests/Feature/RuntimePayloadCompletionUploadsTest.php +++ b/tests/Feature/RuntimePayloadCompletionUploadsTest.php @@ -7,9 +7,11 @@ use App\Models\WorkerRegistration; use App\Models\WorkflowNamespace; use App\Support\RuntimeExternalPayloadCleanup; +use App\Support\RuntimeExternalPayloadRegistry; use App\Support\RuntimePayloadCompletionContext; use App\Support\WorkflowQueryTaskBroker; use Illuminate\Foundation\Testing\RefreshDatabase; +use Illuminate\Support\Facades\DB; use Illuminate\Support\Facades\File; use Illuminate\Support\Facades\Queue; use Illuminate\Testing\TestResponse; @@ -147,8 +149,8 @@ public function test_workflow_slots_are_bounded_even_when_the_bytes_are_deduplic $this->upload('same', $context)->assertCreated(); } $context['slot'] = ['commands', 128, 'arguments']; - $this->upload('same', $context)->assertStatus(429) - ->assertJsonPath('reason', 'external_payload_completion_budget_exhausted'); + $this->upload('same', $context)->assertStatus(503) + ->assertJsonPath('reason', 'storage_pressure')->assertJsonPath('request_admitted', false); $this->assertDatabaseCount('runtime_external_payloads', 1); $this->assertCount(128, RuntimePayloadCompletionBudget::query()->sole()->slots); } @@ -179,6 +181,39 @@ public static function wrongContexts(): array [['lease_owner' => 'another-worker'], 'default'], [[], 'other']]; } + public function test_operator_credentials_cannot_claim_the_worker_completion_exception(): void + { + config(['server.auth.driver' => 'token', 'server.auth.token' => null, + 'server.auth.role_tokens' => ['worker' => 'worker-fixture', 'operator' => 'operator-fixture']]); + $context = ['schema' => RuntimePayloadCompletionContext::SCHEMA, 'kind' => 'activity', + 'task_id' => 'missing', 'attempt' => 'missing', 'lease_owner' => 'worker', + 'operation' => 'complete', 'slot' => ['result']]; + $this->observeStoragePressure('draining'); + $this->upload('bytes', $context, token: 'operator-fixture')->assertForbidden() + ->assertJsonPath('reason', 'external_payload_unauthorized'); + $this->upload('bytes', $context, token: 'worker-fixture')->assertStatus(409) + ->assertJsonPath('reason', 'external_payload_completion_lease_rejected'); + $this->assertDatabaseCount('runtime_payload_completion_budgets', 0); + } + + public function test_registry_ownership_guards_precede_object_locks_in_each_write_transaction(): void + { + $order = []; + DB::listen(function ($query) use (&$order): void { + if (str_starts_with(strtolower($query->sql), 'select') + && str_contains($query->sql, 'runtime_external_payload_object_locks')) { + $order[] = 'object-lock'; + } + }); + app(RuntimeExternalPayloadRegistry::class)->upload('default', 'bytes', 'avro', hash('sha256', 'bytes'), + function () use (&$order): void { + $this->assertGreaterThan(0, DB::transactionLevel()); + $order[] = 'ownership'; + }); + $this->assertSame(['ownership', 'object-lock', 'ownership', + 'ownership', 'object-lock', 'ownership'], $order); + } + public function test_expired_lease_cannot_create_an_upload(): void { $context = $this->activity(); @@ -198,10 +233,13 @@ public function test_one_slot_cannot_rotate_content_and_complete_fail_share_the_ $this->upload('other', $context)->assertStatus(409)->assertJsonPath('reason', 'external_payload_completion_conflict'); $context['operation'] = 'fail'; $context['slot'] = ['failure', 'details']; - $this->upload('other', $context)->assertStatus(429)->assertJsonPath('reason', 'external_payload_completion_budget_exhausted'); + $this->upload('other', $context)->assertStatus(503)->assertJsonPath('reason', 'storage_pressure') + ->assertJsonPath('request_admitted', false)->assertJsonPath('retryable', true); $this->upload('first', $context)->assertCreated(); $this->assertDatabaseCount('runtime_external_payloads', 1); $this->assertDatabaseCount('runtime_payload_completion_budgets', 1); + $this->observeStoragePressure('normal'); + $this->upload('other', null)->assertCreated(); } #[DataProvider('closedAdmission')] @@ -294,13 +332,14 @@ private function workflow(): array 'lease_owner' => 'worker', 'operation' => 'complete', 'slot' => ['commands', 0, 'result']]; } - private function upload(string $bytes, array $context, string $namespace = 'default'): TestResponse + private function upload(string $bytes, ?array $context, string $namespace = 'default', ?string $token = null): TestResponse { return $this->call('POST', '/api/external-payloads/v1', [], [], [], [ 'CONTENT_TYPE' => 'application/octet-stream', 'HTTP_X_NAMESPACE' => $namespace, 'HTTP_X_DURABLE_WORKFLOW_PAYLOAD_CODEC' => 'avro', 'HTTP_X_DURABLE_WORKFLOW_PAYLOAD_SIZE' => (string) strlen($bytes), 'HTTP_X_DURABLE_WORKFLOW_PAYLOAD_SHA256' => hash('sha256', $bytes), - 'HTTP_X_DURABLE_WORKFLOW_PAYLOAD_COMPLETION' => json_encode($context, JSON_THROW_ON_ERROR), + ...($context === null ? [] : ['HTTP_X_DURABLE_WORKFLOW_PAYLOAD_COMPLETION' => json_encode($context, JSON_THROW_ON_ERROR)]), + ...($token === null ? [] : ['HTTP_AUTHORIZATION' => 'Bearer '.$token]), ], $bytes); } } diff --git a/tests/Support/RuntimePayloadCompletionProcess.php b/tests/Support/RuntimePayloadCompletionProcess.php new file mode 100644 index 00000000..393e07b9 --- /dev/null +++ b/tests/Support/RuntimePayloadCompletionProcess.php @@ -0,0 +1,122 @@ +make(Kernel::class)->bootstrap(); +Queue::fake(); + +try { + [$script, $action, $directory, $kind, $variant, $slot, $barrier] = $argv; + config(['server.storage_admission.file' => $directory.'/pressure.json', + 'server.storage_admission.source' => 'completion-process-test', + 'server.storage_admission.max_age_seconds' => 300, + 'server.external_payload_transport.max_payload_bytes' => 4096, + 'server.external_payload_transport.completion_max_bytes' => strlen(Serializer::serializeWithCodec('avro', str_repeat('alpha', 100)))]); + $request = static function (string $path, array|string $body, array $headers = []): array { + $request = Request::create($path, 'POST', [], [], [], array_replace([ + 'CONTENT_TYPE' => is_array($body) ? 'application/json' : 'application/octet-stream', + 'HTTP_ACCEPT' => 'application/json', 'HTTP_X_NAMESPACE' => 'default', + 'HTTP_X_DURABLE_WORKFLOW_CONTROL_PLANE_VERSION' => '2', + 'HTTP_'.str_replace('-', '_', strtoupper(WorkerProtocol::HEADER)) => WorkerProtocol::VERSION, + ], $headers), is_array($body) ? json_encode($body, JSON_THROW_ON_ERROR) : $body); + $kernel = app(Illuminate\Contracts\Http\Kernel::class); + $response = $kernel->handle($request); + $kernel->terminate($request, $response); + + return ['status' => $response->getStatusCode(), 'body' => json_decode($response->getContent(), true, flags: JSON_THROW_ON_ERROR)]; + }; + if ($action === 'init') { + file_put_contents($directory.'/pressure.json', json_encode(['schema' => StoragePressure::SCHEMA, + 'source' => 'completion-process-test', 'observed_at' => time(), 'state' => 'normal'], JSON_THROW_ON_ERROR)); + Artisan::call('migrate:fresh', ['--force' => true]); + WorkflowNamespace::query()->updateOrCreate(['name' => 'default'], ['status' => 'active', 'retention_days' => 30, + 'external_payload_storage' => ['driver' => 'local', 'enabled' => true, 'threshold_bytes' => 32, + 'config' => ['uri' => 'file://'.$directory.'/objects']]]); + WorkerRegistration::query()->create(['worker_id' => 'worker', 'namespace' => 'default', 'task_queue' => 'queue', + 'runtime' => 'php', 'supported_workflow_types' => ['test.workflow'], 'supported_activity_types' => ['test.activity'], + 'last_heartbeat_at' => now(), 'status' => 'active']); + $plural = $kind === 'activity' ? 'activities' : 'workflows'; + $started = $request('/api/'.$plural, [$kind.'_id' => 'test', $kind.'_type' => 'test.'.$kind, + 'task_queue' => 'queue', 'input' => []]); + if ($started['status'] !== 201) { + throw new RuntimeException(json_encode($started, JSON_THROW_ON_ERROR)); + } + $polled = $request('/api/worker/'.$kind.'-tasks/poll', ['worker_id' => 'worker', 'task_queue' => 'queue']); + $task = $polled['body']['task'] ?? null; + if (! is_array($task)) { + throw new RuntimeException(json_encode($polled, JSON_THROW_ON_ERROR)); + } + $context = ['schema' => RuntimePayloadCompletionContext::SCHEMA, 'kind' => $kind, + 'task_id' => $task['task_id'], 'attempt' => $task[$kind === 'activity' ? 'activity_attempt_id' : 'workflow_task_attempt'], + 'lease_owner' => 'worker', 'operation' => 'complete', + 'slot' => $kind === 'activity' ? ['result'] : ['commands', 0, 'result']]; + file_put_contents($directory.'/context.json', json_encode($context, JSON_THROW_ON_ERROR)); + file_put_contents($directory.'/pressure.json', json_encode(['schema' => StoragePressure::SCHEMA, + 'source' => 'completion-process-test', 'observed_at' => time(), 'state' => 'draining'], JSON_THROW_ON_ERROR)); + echo '{}'; + exit(0); + } + $context = json_decode(file_get_contents($directory.'/context.json'), true, flags: JSON_THROW_ON_ERROR); + if ($action === 'upload') { + if ($slot !== '0') { + if ($kind === 'activity') { + $context['operation'] = 'fail'; + $context['slot'] = ['failure', 'details']; + } else { + $context['slot'] = ['commands', (int) $slot, 'result']; + } + } + if ($barrier !== '') { + touch($directory.'/'.$barrier.'.ready'); + $deadline = microtime(true) + 10; + while (! is_file($directory.'/go')) { + if (microtime(true) > $deadline) { + throw new RuntimeException('Upload barrier timed out.'); + } + usleep(10000); + } + } + $bytes = Serializer::serializeWithCodec('avro', str_repeat($variant, 100)); + echo json_encode($request('/api/external-payloads/v1', $bytes, [ + 'HTTP_X_DURABLE_WORKFLOW_PAYLOAD_COMPLETION' => json_encode($context, JSON_THROW_ON_ERROR), + 'HTTP_X_DURABLE_WORKFLOW_PAYLOAD_CODEC' => 'avro', 'HTTP_X_DURABLE_WORKFLOW_PAYLOAD_SIZE' => (string) strlen($bytes), + 'HTTP_X_DURABLE_WORKFLOW_PAYLOAD_SHA256' => hash('sha256', $bytes), + ]), JSON_THROW_ON_ERROR); + } elseif ($action === 'complete') { + $row = RuntimeExternalPayload::query()->sole(); + $reference = app(RuntimeExternalPayloadRegistry::class)->referenceForUri('default', $row->storage_uri); + $envelope = ['codec' => 'avro', 'external_payload' => $reference]; + $body = ['lease_owner' => 'worker']; + if ($kind === 'activity') { + $body += ['activity_attempt_id' => $context['attempt'], 'result' => $envelope]; + } else { + $body += ['workflow_task_attempt' => $context['attempt'], + 'commands' => [['type' => 'complete_workflow', 'sequence' => 1, 'result' => $envelope]]]; + } + echo json_encode($request('/api/worker/'.$kind.'-tasks/'.$context['task_id'].'/complete', $body), JSON_THROW_ON_ERROR); + } elseif ($action === 'status') { + $budget = RuntimePayloadCompletionBudget::query()->sole(); + echo json_encode(['budgets' => RuntimePayloadCompletionBudget::query()->count(), + 'slots' => count($budget->slots), 'objects' => count($budget->objects), + 'bytes' => array_sum($budget->objects), 'rows' => RuntimeExternalPayload::query()->count()], JSON_THROW_ON_ERROR); + } else { + throw new RuntimeException('Unknown completion probe action.'); + } +} catch (Throwable $exception) { + fwrite(STDERR, $exception::class.': '.$exception->getMessage().PHP_EOL); + exit(1); +} From 6fa39ce53115e5877e7be45692c59e2d7be59c30 Mon Sep 17 00:00:00 2001 From: Durable Workflow Date: Tue, 8 Sep 2026 17:22:31 +0000 Subject: [PATCH 4/5] Run cold completion ownership tests on all supported SQL backends --- .github/workflows/phpunit-feature.yml | 2 ++ .../RuntimePayloadCompletionProcessTest.php | 19 +++++++++++-------- 2 files changed, 13 insertions(+), 8 deletions(-) diff --git a/.github/workflows/phpunit-feature.yml b/.github/workflows/phpunit-feature.yml index 30606976..1c244f01 100644 --- a/.github/workflows/phpunit-feature.yml +++ b/.github/workflows/phpunit-feature.yml @@ -215,7 +215,9 @@ jobs: -e DW_TEST_S3_ACCESS_KEY_ID=dw-minio-access \ -e DW_TEST_S3_SECRET_ACCESS_KEY=dw-minio-secret-key \ -e DW_TEST_BACKUP_MYSQL_HOST="dw-server-backup-mysql-${{ github.run_id }}-${{ github.run_attempt }}-${{ github.job }}" \ + -e DW_TEST_COMPLETION_MYSQL_HOST="dw-server-backup-mysql-${{ github.run_id }}-${{ github.run_attempt }}-${{ github.job }}" \ -e DW_TEST_BACKUP_PGSQL_HOST="dw-server-backup-pgsql-${{ github.run_id }}-${{ github.run_attempt }}-${{ github.job }}" \ + -e DW_TEST_COMPLETION_PGSQL_HOST="dw-server-backup-pgsql-${{ github.run_id }}-${{ github.run_attempt }}-${{ github.job }}" \ -w /app \ --entrypoint sh \ durable-workflow-server-corpus-validator \ diff --git a/tests/Feature/RuntimePayloadCompletionProcessTest.php b/tests/Feature/RuntimePayloadCompletionProcessTest.php index cfe22175..e5f0257a 100644 --- a/tests/Feature/RuntimePayloadCompletionProcessTest.php +++ b/tests/Feature/RuntimePayloadCompletionProcessTest.php @@ -22,7 +22,8 @@ class RuntimePayloadCompletionProcessTest extends TestCase public static function cases(): array { - return [['sqlite', 'activity'], ['sqlite', 'workflow'], ['mysql', 'activity'], ['mysql', 'workflow']]; + return [['sqlite', 'activity'], ['sqlite', 'workflow'], ['mysql', 'activity'], ['mysql', 'workflow'], + ['pgsql', 'activity'], ['pgsql', 'workflow']]; } #[DataProvider('cases')] @@ -91,9 +92,10 @@ protected function tearDown(): void private function initialize(string $driver): void { - $host = getenv('DW_TEST_COMPLETION_MYSQL_HOST'); - if ($driver === 'mysql' && ! $host) { - $this->markTestSkipped('Set DW_TEST_COMPLETION_MYSQL_HOST to a disposable MySQL test server.'); + $prefix = $driver === 'pgsql' ? 'DW_TEST_COMPLETION_PGSQL_' : 'DW_TEST_COMPLETION_MYSQL_'; + $host = getenv($prefix.'HOST'); + if ($driver !== 'sqlite' && ! $host) { + $this->markTestSkipped('Set '.$prefix.'HOST to a disposable database test server.'); } $this->directory = sys_get_temp_dir().'/dw-completion-process-'.bin2hex(random_bytes(6)); mkdir($this->directory, 0700); @@ -105,12 +107,13 @@ private function initialize(string $driver): void touch($this->environment['DB_DATABASE']); } else { $this->database = 'dw_completion_'.bin2hex(random_bytes(6)); - $user = getenv('DW_TEST_COMPLETION_MYSQL_USER') ?: 'root'; - $password = getenv('DW_TEST_COMPLETION_MYSQL_PASSWORD') ?: ''; - $this->databaseAdmin = new PDO('mysql:host='.$host.';port=3306', $user, $password, + $user = getenv($prefix.'USER') ?: ($driver === 'pgsql' ? 'postgres' : 'root'); + $password = getenv($prefix.'PASSWORD') ?: ''; + $port = $driver === 'pgsql' ? '5432' : '3306'; + $this->databaseAdmin = new PDO($driver.':host='.$host.';port='.$port.($driver === 'pgsql' ? ';dbname=postgres' : ''), $user, $password, [PDO::ATTR_ERRMODE => PDO::ERRMODE_EXCEPTION]); $this->databaseAdmin->exec('CREATE DATABASE '.$this->database); - $this->environment = array_replace($this->environment, ['DB_HOST' => $host, 'DB_PORT' => '3306', + $this->environment = array_replace($this->environment, ['DB_HOST' => $host, 'DB_PORT' => $port, 'DB_DATABASE' => $this->database, 'DB_USERNAME' => $user, 'DB_PASSWORD' => $password]); } } From fd28bd2503ec40e60c51d4ee23a2e78d3cd0525b Mon Sep 17 00:00:00 2001 From: Durable Workflow Date: Tue, 8 Sep 2026 17:37:04 +0000 Subject: [PATCH 5/5] Prepare Server 2.3.5 and chart 0.1.84 --- composer.json | 2 +- composer.lock | 2 +- docker-compose.dedicated-matching.yml | 4 ++-- docker-compose.memo-rolling.yml | 4 ++-- docker-compose.published.yml | 4 ++-- docker-compose.small-cluster.yml | 2 +- docker-compose.yml | 8 ++++---- k8s/README.md | 10 +++++----- k8s/helm/durable-workflow/Chart.yaml | 6 +++--- k8s/helm/durable-workflow/README.md | 2 +- .../durable-workflow/ci/existing-secrets-values.yaml | 2 +- .../durable-workflow/ci/ingress-and-hpa-values.yaml | 2 +- .../durable-workflow/ci/inline-secrets-values.yaml | 2 +- k8s/helm/durable-workflow/templates/_helpers.tpl | 2 +- k8s/helm/durable-workflow/values.yaml | 2 +- k8s/helm/examples/values-dev.yaml | 2 +- .../examples/values-external-secrets-operator.yaml | 2 +- .../examples/values-production-existing-secrets.yaml | 2 +- k8s/migration-job.yaml | 2 +- k8s/scheduler-cronjob.yaml | 2 +- k8s/secret.yaml | 2 +- k8s/server-deployment.yaml | 2 +- k8s/worker-deployment.yaml | 2 +- resources/release/source-release.json | 4 ++-- scripts/k8s-kind-smoke.sh | 2 +- 25 files changed, 38 insertions(+), 38 deletions(-) diff --git a/composer.json b/composer.json index 333e7122..5340112d 100644 --- a/composer.json +++ b/composer.json @@ -48,7 +48,7 @@ }, "extra": { "durable-workflow": { - "product-train": "2.3.4" + "product-train": "2.3.5" }, "laravel": { "dont-discover": [] diff --git a/composer.lock b/composer.lock index 908c17a6..7310934a 100644 --- a/composer.lock +++ b/composer.lock @@ -4,7 +4,7 @@ "Read more about it at https://getcomposer.org/doc/01-basic-usage.md#installing-dependencies", "This file is @generated automatically" ], - "content-hash": "3e5f896b0401b41e5a397549f446227c", + "content-hash": "113d9348b3c41acbd32018eb77476363", "packages": [ { "name": "apache/avro", diff --git a/docker-compose.dedicated-matching.yml b/docker-compose.dedicated-matching.yml index 7b252a5b..9d34c18c 100644 --- a/docker-compose.dedicated-matching.yml +++ b/docker-compose.dedicated-matching.yml @@ -32,13 +32,13 @@ name: durable-workflow-server # daemon reports `shape: dedicated`. # Generated by scripts/ci/sync-source-release.mjs. Do not edit the fallback. -x-server-image: &server-image ${DW_SERVER_IMAGE:-durableworkflow/server:${DW_SERVER_TAG:-2.3.4}} +x-server-image: &server-image ${DW_SERVER_IMAGE:-durableworkflow/server:${DW_SERVER_TAG:-2.3.5}} x-server-environment: &server-environment APP_NAME: "Durable Workflow Server" APP_ENV: ${APP_ENV:-local} DW_SERVER_KEY: ${DW_SERVER_KEY:-} - APP_VERSION: ${APP_VERSION:-${DW_SERVER_TAG:-2.3.4}} + APP_VERSION: ${APP_VERSION:-${DW_SERVER_TAG:-2.3.5}} APP_DEBUG: ${APP_DEBUG:-false} DB_CONNECTION: mysql DB_HOST: mysql diff --git a/docker-compose.memo-rolling.yml b/docker-compose.memo-rolling.yml index c171366f..158058a0 100644 --- a/docker-compose.memo-rolling.yml +++ b/docker-compose.memo-rolling.yml @@ -49,14 +49,14 @@ services: command: ["server-bootstrap"] environment: <<: *runtime-environment - APP_VERSION: ${APP_VERSION:-2.3.4} + APP_VERSION: ${APP_VERSION:-2.3.5} successor: image: ${DW_MEMO_SUCCESSOR_IMAGE:-durable-workflow/server-memo-rolling:local} ports: !override [] environment: <<: *runtime-environment - APP_VERSION: ${APP_VERSION:-2.3.4} + APP_VERSION: ${APP_VERSION:-2.3.5} DW_SERVER_ID: memo-successor DW_SERVER_TOPOLOGY_SHAPE: standalone_server DW_SERVER_PROCESS_CLASS: server_http_node diff --git a/docker-compose.published.yml b/docker-compose.published.yml index 90e7184a..2203209e 100644 --- a/docker-compose.published.yml +++ b/docker-compose.published.yml @@ -1,13 +1,13 @@ name: durable-workflow-server # Generated by scripts/ci/sync-source-release.mjs. Do not edit the fallback. -x-server-image: &server-image ${DW_SERVER_IMAGE:-durableworkflow/server:${DW_SERVER_TAG:-2.3.4}} +x-server-image: &server-image ${DW_SERVER_IMAGE:-durableworkflow/server:${DW_SERVER_TAG:-2.3.5}} x-server-environment: &server-environment APP_NAME: "Durable Workflow Server" APP_ENV: ${APP_ENV:-local} DW_SERVER_KEY: ${DW_SERVER_KEY:-} - APP_VERSION: ${APP_VERSION:-${DW_SERVER_TAG:-2.3.4}} + APP_VERSION: ${APP_VERSION:-${DW_SERVER_TAG:-2.3.5}} APP_DEBUG: ${APP_DEBUG:-false} LOG_CHANNEL: ${LOG_CHANNEL:-stderr} LOG_LEVEL: ${LOG_LEVEL:-info} diff --git a/docker-compose.small-cluster.yml b/docker-compose.small-cluster.yml index 476410c2..27fb87e9 100644 --- a/docker-compose.small-cluster.yml +++ b/docker-compose.small-cluster.yml @@ -12,7 +12,7 @@ x-server-build: &server-build x-server-environment: &server-environment APP_NAME: "Durable Workflow Server" APP_ENV: testing - APP_VERSION: ${APP_VERSION:-2.3.4} + APP_VERSION: ${APP_VERSION:-2.3.5} APP_DEBUG: "false" DW_SERVER_KEY: ${DW_SERVER_KEY:-base64:5Zt4nUhlCm3DD0nLXZJQdHiwPfb56yGo9gNV/g3jYbY=} DB_CONNECTION: ${DW_SMALL_CLUSTER_DB:-mysql} diff --git a/docker-compose.yml b/docker-compose.yml index feb61343..d22bea7b 100644 --- a/docker-compose.yml +++ b/docker-compose.yml @@ -15,7 +15,7 @@ services: DW_SERVER_KEY: "${DW_SERVER_KEY:-}" DW_SERVER_TOPOLOGY_SHAPE: standalone_server DW_SERVER_PROCESS_CLASS: server_http_node - APP_VERSION: "${APP_VERSION:-2.3.4}" + APP_VERSION: "${APP_VERSION:-2.3.5}" APP_DEBUG: "false" DB_CONNECTION: mysql DB_HOST: mysql @@ -62,7 +62,7 @@ services: APP_NAME: "Durable Workflow Server" APP_ENV: local DW_SERVER_KEY: "${DW_SERVER_KEY:-}" - APP_VERSION: "${APP_VERSION:-2.3.4}" + APP_VERSION: "${APP_VERSION:-2.3.5}" APP_DEBUG: "false" DB_CONNECTION: mysql DB_HOST: mysql @@ -123,7 +123,7 @@ services: DW_SERVER_KEY: "${DW_SERVER_KEY:-}" DW_SERVER_TOPOLOGY_SHAPE: standalone_server DW_SERVER_PROCESS_CLASS: worker_node - APP_VERSION: "${APP_VERSION:-2.3.4}" + APP_VERSION: "${APP_VERSION:-2.3.5}" DB_CONNECTION: mysql DB_HOST: mysql DB_PORT: 3306 @@ -176,7 +176,7 @@ services: DW_SERVER_KEY: "${DW_SERVER_KEY:-}" DW_SERVER_TOPOLOGY_SHAPE: standalone_server DW_SERVER_PROCESS_CLASS: scheduler_node - APP_VERSION: "${APP_VERSION:-2.3.4}" + APP_VERSION: "${APP_VERSION:-2.3.5}" DB_CONNECTION: mysql DB_HOST: mysql DB_PORT: 3306 diff --git a/k8s/README.md b/k8s/README.md index 4baf0110..3bf96497 100644 --- a/k8s/README.md +++ b/k8s/README.md @@ -13,7 +13,7 @@ The checked-in manifests are synchronized with the repository's stable source release and pin its Docker Hub tag: ```text -durableworkflow/server:2.3.4 +durableworkflow/server:2.3.5 ``` Before production use, patch every workload image to the exact published tag or @@ -21,15 +21,15 @@ digest you intend to run: ```bash kubectl set image -n durable-workflow deploy/durable-workflow-server \ - server=durableworkflow/server:2.3.4 + server=durableworkflow/server:2.3.5 kubectl set image -n durable-workflow deploy/durable-workflow-worker \ - worker=durableworkflow/server:2.3.4 + worker=durableworkflow/server:2.3.5 kubectl set image -n durable-workflow cronjob/durable-workflow-scheduler \ - scheduler=durableworkflow/server:2.3.4 + scheduler=durableworkflow/server:2.3.5 ``` GitHub Container Registry publishes the same release line at -`ghcr.io/durable-workflow/server:2.3.4`. Digest pinning is preferred for strict +`ghcr.io/durable-workflow/server:2.3.5`. Digest pinning is preferred for strict change control. The manifests expect you to provide: diff --git a/k8s/helm/durable-workflow/Chart.yaml b/k8s/helm/durable-workflow/Chart.yaml index d4312328..4fdf850a 100644 --- a/k8s/helm/durable-workflow/Chart.yaml +++ b/k8s/helm/durable-workflow/Chart.yaml @@ -5,11 +5,11 @@ type: application # The chart's own semver version. Bumped on every chart release; treated as # independent of the server image version (appVersion). Breaking-change rules # for this version live in docs/helm-upgrading.md alongside the chart. -version: 0.1.83 +version: 0.1.84 # The immutable Durable Workflow Server identity this chart release packages. # The onboarding default in values.yaml and appVersion are generated from the # checked-in source release record. -appVersion: "2.3.4" +appVersion: "2.3.5" kubeVersion: ">=1.27.0-0" home: https://durable-workflow.github.io/docs/2.0/deployment sources: @@ -30,7 +30,7 @@ annotations: # exact commit that most recently changed the packaged chart. org.opencontainers.image.source: https://github.com/durable-workflow/server dev.durable-workflow.source-revision: "unreleased" - dev.durable-workflow.image-reference: "docker.io/durableworkflow/server:2.3.4" + dev.durable-workflow.image-reference: "docker.io/durableworkflow/server:2.3.5" artifacthub.io/license: MIT artifacthub.io/category: integration-delivery # Free-form changelog for the current chart release shown by Artifact Hub. diff --git a/k8s/helm/durable-workflow/README.md b/k8s/helm/durable-workflow/README.md index d189b6a3..e39ca26c 100644 --- a/k8s/helm/durable-workflow/README.md +++ b/k8s/helm/durable-workflow/README.md @@ -63,7 +63,7 @@ helm install durable-workflow ./k8s/helm/durable-workflow \ ```yaml image: - tag: "2.3.4" + tag: "2.3.5" # Pin a digest in production: # digest: "sha256:abc123..." # memoPayloadStorage: "raw-json-v1" # Required for a digest or custom image. diff --git a/k8s/helm/durable-workflow/ci/existing-secrets-values.yaml b/k8s/helm/durable-workflow/ci/existing-secrets-values.yaml index df6f3395..3559eb19 100644 --- a/k8s/helm/durable-workflow/ci/existing-secrets-values.yaml +++ b/k8s/helm/durable-workflow/ci/existing-secrets-values.yaml @@ -1,7 +1,7 @@ # CI fixture: GitOps / externally-managed-secret path. The chart consumes # existing Secrets and renders no Secret resources of its own. image: - tag: "2.3.4" + tag: "2.3.5" externalDatabase: connection: pgsql diff --git a/k8s/helm/durable-workflow/ci/ingress-and-hpa-values.yaml b/k8s/helm/durable-workflow/ci/ingress-and-hpa-values.yaml index 18ccae0a..199d9de3 100644 --- a/k8s/helm/durable-workflow/ci/ingress-and-hpa-values.yaml +++ b/k8s/helm/durable-workflow/ci/ingress-and-hpa-values.yaml @@ -1,6 +1,6 @@ # CI fixture: ingress + autoscaling enabled. Exercises optional templates. image: - tag: "2.3.4" + tag: "2.3.5" externalDatabase: connection: mysql diff --git a/k8s/helm/durable-workflow/ci/inline-secrets-values.yaml b/k8s/helm/durable-workflow/ci/inline-secrets-values.yaml index 0d4c6c74..57fad99a 100644 --- a/k8s/helm/durable-workflow/ci/inline-secrets-values.yaml +++ b/k8s/helm/durable-workflow/ci/inline-secrets-values.yaml @@ -2,7 +2,7 @@ # chart's render path is exercised end-to-end. Real deployments should use # existingSecret instead. image: - tag: "2.3.4" + tag: "2.3.5" externalDatabase: connection: mysql diff --git a/k8s/helm/durable-workflow/templates/_helpers.tpl b/k8s/helm/durable-workflow/templates/_helpers.tpl index 5d6c25aa..f75df08b 100644 --- a/k8s/helm/durable-workflow/templates/_helpers.tpl +++ b/k8s/helm/durable-workflow/templates/_helpers.tpl @@ -88,7 +88,7 @@ resolved by an explicit capability declaration or an existing workload marker. {{- define "durable-workflow.memoPayloadStorageForImage" -}} {{- $image := toString . -}} {{- $normalized := regexReplaceAll "^index\\.docker\\.io/" $image "docker.io/" -}} -{{- if eq $normalized "docker.io/durableworkflow/server:2.3.4" -}} +{{- if eq $normalized "docker.io/durableworkflow/server:2.3.5" -}} dual-v1 {{- else if regexMatch "^docker\\.io/durableworkflow/server:2\\.0\\.0-rc\\.[0-9]+$" $normalized -}} {{- $releaseCandidate := atoi (regexFind "[0-9]+$" $normalized) -}} diff --git a/k8s/helm/durable-workflow/values.yaml b/k8s/helm/durable-workflow/values.yaml index 2f9a707a..acc8d6e9 100644 --- a/k8s/helm/durable-workflow/values.yaml +++ b/k8s/helm/durable-workflow/values.yaml @@ -21,7 +21,7 @@ image: registry: docker.io repository: durableworkflow/server # Generated by scripts/ci/sync-source-release.mjs. Do not edit this default. - tag: "2.3.4" + tag: "2.3.5" # Optional digest pin. When set, takes precedence over tag for change control. # Example: "sha256:abc123..." digest: "" diff --git a/k8s/helm/examples/values-dev.yaml b/k8s/helm/examples/values-dev.yaml index 5ee36577..f0fcc481 100644 --- a/k8s/helm/examples/values-dev.yaml +++ b/k8s/helm/examples/values-dev.yaml @@ -3,7 +3,7 @@ # shape in production. image: - tag: "2.3.4" + tag: "2.3.5" externalDatabase: connection: mysql diff --git a/k8s/helm/examples/values-external-secrets-operator.yaml b/k8s/helm/examples/values-external-secrets-operator.yaml index f2c3b3e9..f0558429 100644 --- a/k8s/helm/examples/values-external-secrets-operator.yaml +++ b/k8s/helm/examples/values-external-secrets-operator.yaml @@ -5,7 +5,7 @@ # concern. image: - tag: "2.3.4" + tag: "2.3.5" externalDatabase: connection: pgsql diff --git a/k8s/helm/examples/values-production-existing-secrets.yaml b/k8s/helm/examples/values-production-existing-secrets.yaml index 1ebd2e05..731483cf 100644 --- a/k8s/helm/examples/values-production-existing-secrets.yaml +++ b/k8s/helm/examples/values-production-existing-secrets.yaml @@ -10,7 +10,7 @@ image: repository: durable-workflow/server # Pin a digest in production for change-control auditability. digest: "" # e.g. "sha256:abc123..." - tag: "2.3.4" + tag: "2.3.5" externalDatabase: connection: pgsql diff --git a/k8s/migration-job.yaml b/k8s/migration-job.yaml index 51443aa4..82ba17c1 100644 --- a/k8s/migration-job.yaml +++ b/k8s/migration-job.yaml @@ -13,7 +13,7 @@ spec: restartPolicy: OnFailure containers: - name: migrate - image: durableworkflow/server:2.3.4 + image: durableworkflow/server:2.3.5 command: ["server-entrypoint"] args: ["server-bootstrap"] envFrom: diff --git a/k8s/scheduler-cronjob.yaml b/k8s/scheduler-cronjob.yaml index 5197075e..2cf08f79 100644 --- a/k8s/scheduler-cronjob.yaml +++ b/k8s/scheduler-cronjob.yaml @@ -24,7 +24,7 @@ spec: restartPolicy: Never containers: - name: scheduler - image: durableworkflow/server:2.3.4 + image: durableworkflow/server:2.3.5 command: ["server-entrypoint"] args: ["sh", "-c", "php artisan schedule:evaluate --limit=100 --json; php artisan activity:timeout-enforce --limit=100; if php artisan list --raw | grep -q '^external-payloads:cleanup '; then php artisan external-payloads:cleanup --limit=100 --json; fi; php artisan history:prune --limit=100"] envFrom: diff --git a/k8s/secret.yaml b/k8s/secret.yaml index 542baa30..43f6631b 100644 --- a/k8s/secret.yaml +++ b/k8s/secret.yaml @@ -12,7 +12,7 @@ metadata: app.kubernetes.io/name: durable-workflow data: APP_NAME: "Durable Workflow Server" - APP_VERSION: "2.3.4" + APP_VERSION: "2.3.5" APP_ENV: production APP_DEBUG: "false" DB_CONNECTION: mysql diff --git a/k8s/server-deployment.yaml b/k8s/server-deployment.yaml index 3cadef37..d665b18b 100644 --- a/k8s/server-deployment.yaml +++ b/k8s/server-deployment.yaml @@ -23,7 +23,7 @@ spec: spec: containers: - name: server - image: durableworkflow/server:2.3.4 + image: durableworkflow/server:2.3.5 ports: - containerPort: 8080 name: http diff --git a/k8s/worker-deployment.yaml b/k8s/worker-deployment.yaml index 2a5d6303..e518f591 100644 --- a/k8s/worker-deployment.yaml +++ b/k8s/worker-deployment.yaml @@ -19,7 +19,7 @@ spec: spec: containers: - name: worker - image: durableworkflow/server:2.3.4 + image: durableworkflow/server:2.3.5 command: ["server-entrypoint"] args: ["php", "artisan", "queue:work", "--sleep=1", "--tries=3", "--max-time=3600"] envFrom: diff --git a/resources/release/source-release.json b/resources/release/source-release.json index 5779b8db..f4765b34 100644 --- a/resources/release/source-release.json +++ b/resources/release/source-release.json @@ -1,9 +1,9 @@ { "schema": "durable-workflow.server.source-release/v1", "server": { - "version": "2.3.4" + "version": "2.3.5" }, "helm_chart": { - "version": "0.1.83" + "version": "0.1.84" } } diff --git a/scripts/k8s-kind-smoke.sh b/scripts/k8s-kind-smoke.sh index 825921f6..ffb6698d 100755 --- a/scripts/k8s-kind-smoke.sh +++ b/scripts/k8s-kind-smoke.sh @@ -7,7 +7,7 @@ cluster="${K8S_SMOKE_CLUSTER:-durable-workflow-server-smoke}" image="${K8S_SMOKE_IMAGE:-durableworkflow/server:k8s-smoke}" # Generated by scripts/ci/sync-source-release.mjs so the smoke replaces the # same default shipped by the public manifests. -manifest_image="durableworkflow/server:2.3.4" +manifest_image="durableworkflow/server:2.3.5" kind_node_image="${K8S_SMOKE_KIND_NODE_IMAGE:-kindest/node:v1.29.4}" artifact_dir="${K8S_SMOKE_ARTIFACT_DIR:-/tmp/durable-workflow-k8s-kind-smoke-artifacts}" rendered_dir="${artifact_dir}/rendered-manifests"