diff --git a/.github/workflows/phpunit-feature.yml b/.github/workflows/phpunit-feature.yml index 2898526d6..3929f9396 100644 --- a/.github/workflows/phpunit-feature.yml +++ b/.github/workflows/phpunit-feature.yml @@ -257,7 +257,18 @@ jobs: --health-cmd='pg_isready --host=127.0.0.1 --username=postgres' \ --health-interval=1s --health-start-period=5s --health-retries=30 \ mirror.gcr.io/library/postgres@sha256:f1c3376c26f2609ab9f29f71f824103fe2fcd8ee0346485cb6122a4f93df6f94 - for database in mysql pgsql; do + # MariaDB for the real-database contention proofs that must cover + # every supported engine (Server #774). + docker run --detach \ + --name "dw-server-backup-mariadb-${{ github.run_id }}-${{ github.run_attempt }}-${{ github.job }}" \ + --network "dw-server-feature-${{ github.run_id }}-${{ github.run_attempt }}-${{ github.job }}" \ + --memory 384m --memory-swap 384m \ + --env MARIADB_ALLOW_EMPTY_ROOT_PASSWORD=yes --env MARIADB_ROOT_HOST=% \ + --health-cmd='mariadb-admin --silent --host=127.0.0.1 ping' \ + --health-interval=1s --health-start-period=20s --health-retries=30 \ + mirror.gcr.io/library/mariadb@sha256:7db29378d4fdab73f8123bbc2b48905c90d1a4b00cf848b028f1e81e623257f2 \ + --innodb-buffer-pool-size=64M --max-connections=40 + for database in mysql pgsql mariadb; do for attempt in $(seq 1 60); do health="$(docker inspect --format '{{.State.Health.Status}}' "dw-server-backup-${database}-${{ github.run_id }}-${{ github.run_attempt }}-${{ github.job }}")" [ "$health" = healthy ] && break @@ -297,6 +308,7 @@ jobs: -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 }}" \ + -e DW_TEST_COMPLETION_MARIADB_HOST="dw-server-backup-mariadb-${{ github.run_id }}-${{ github.run_attempt }}-${{ github.job }}" \ -e DW_POOLER_TEST_HOST="dw-server-backup-pgsql-${{ github.run_id }}-${{ github.run_attempt }}-${{ github.job }}" \ -e DW_POOLER_TEST_PORT=5432 -e DW_POOLER_TEST_DATABASE=postgres -e DW_POOLER_TEST_USER=postgres \ -e DW_POOLER_TEST_PREFIX="dw-server-pooler-${{ github.run_id }}-${{ github.run_attempt }}-${{ github.job }}" \ @@ -365,6 +377,7 @@ jobs: "dw-server-s3-${{ github.run_id }}-${{ github.run_attempt }}-${{ github.job }}" \ "dw-server-backup-mysql-${{ github.run_id }}-${{ github.run_attempt }}-${{ github.job }}" \ "dw-server-backup-pgsql-${{ github.run_id }}-${{ github.run_attempt }}-${{ github.job }}" \ + "dw-server-backup-mariadb-${{ github.run_id }}-${{ github.run_attempt }}-${{ github.job }}" \ >/dev/null 2>&1 || true docker network rm \ "dw-server-feature-${{ github.run_id }}-${{ github.run_attempt }}-${{ github.job }}" \ diff --git a/.github/workflows/server-perf-soak.yml b/.github/workflows/server-perf-soak.yml index b642e8606..bf0e8548f 100644 --- a/.github/workflows/server-perf-soak.yml +++ b/.github/workflows/server-perf-soak.yml @@ -85,8 +85,8 @@ jobs: DW_PERF_MAX_SERVER_MEMORY_MB: "1024" DW_PERF_MAX_POLLING_KEYS: "2048" DW_PERF_MAX_FINAL_POLLING_KEYS: "0" - DW_PERF_MAX_SERVER_CACHE_KEYS_BY_POLICY: '{"worker_poll_diagnostics":0,"polling_cache_availability_probe":0,"workflow_task_poll_requests":2048,"activity_task_poll_requests":2048,"query_task_poll_requests":2048,"sqlite_worker_poll_claim_gate":1,"long_poll_signals":2048,"long_poll_wait_slots":64,"workflow_query_tasks":128,"task_queue_admission_locks":256,"task_queue_dispatch_counters":256,"namespace_request_admission":256,"shared_service_boundary_admission":256,"namespace_durable_state_quota_rejections":256,"runtime_external_payload_quota_rejections":256,"workflow_task_expired_lease_recovery":256,"history_retention_inline":128,"worker_compatibility_heartbeat":128,"readiness_probe":32}' - DW_PERF_MAX_FINAL_SERVER_CACHE_KEYS_BY_POLICY: '{"worker_poll_diagnostics":0,"polling_cache_availability_probe":0,"workflow_task_poll_requests":0,"activity_task_poll_requests":0,"query_task_poll_requests":0,"sqlite_worker_poll_claim_gate":0,"long_poll_signals":0,"long_poll_wait_slots":0,"workflow_query_tasks":0,"task_queue_admission_locks":0,"task_queue_dispatch_counters":0,"namespace_request_admission":0,"shared_service_boundary_admission":0,"namespace_durable_state_quota_rejections":0,"runtime_external_payload_quota_rejections":0,"workflow_task_expired_lease_recovery":0,"history_retention_inline":0,"worker_compatibility_heartbeat":0,"readiness_probe":0}' + DW_PERF_MAX_SERVER_CACHE_KEYS_BY_POLICY: '{"worker_poll_diagnostics":0,"polling_cache_availability_probe":0,"workflow_task_poll_requests":2048,"activity_task_poll_requests":2048,"query_task_poll_requests":2048,"sqlite_worker_poll_claim_gate":1,"long_poll_signals":2048,"long_poll_wait_slots":64,"workflow_query_tasks":128,"task_queue_admission_locks":256,"task_queue_dispatch_counters":256,"namespace_request_admission":256,"shared_service_boundary_admission":256,"namespace_durable_state_quota_rejections":256,"runtime_external_payload_quota_rejections":256,"workflow_task_expired_lease_recovery":256,"history_retention_inline":128,"stream_retention_inline":128,"worker_compatibility_heartbeat":128,"readiness_probe":32}' + DW_PERF_MAX_FINAL_SERVER_CACHE_KEYS_BY_POLICY: '{"worker_poll_diagnostics":0,"polling_cache_availability_probe":0,"workflow_task_poll_requests":0,"activity_task_poll_requests":0,"query_task_poll_requests":0,"sqlite_worker_poll_claim_gate":0,"long_poll_signals":0,"long_poll_wait_slots":0,"workflow_query_tasks":0,"task_queue_admission_locks":0,"task_queue_dispatch_counters":0,"namespace_request_admission":0,"shared_service_boundary_admission":0,"namespace_durable_state_quota_rejections":0,"runtime_external_payload_quota_rejections":0,"workflow_task_expired_lease_recovery":0,"history_retention_inline":0,"stream_retention_inline":0,"worker_compatibility_heartbeat":0,"readiness_probe":0}' DW_PERF_MAX_SERVER_MEMORY_SLOPE_MB_HOUR: "128" DW_PERF_REMOTE_WRITE_ENABLED: ${{ github.event_name != 'workflow_dispatch' || inputs.remote_write }} DW_PERF_REMOTE_WRITE_URL: ${{ vars.DW_PERF_REMOTE_WRITE_URL }} diff --git a/.github/workflows/server-perf.yml b/.github/workflows/server-perf.yml index deacfc226..4fc0b3e99 100644 --- a/.github/workflows/server-perf.yml +++ b/.github/workflows/server-perf.yml @@ -32,6 +32,7 @@ on: - "app/Support/RuntimeExternalPayloadQuota.php" - "app/Support/WorkerPollClaimGate.php" - "app/Support/WorkflowQueryTaskBroker.php" + - "app/Support/WorkflowStreamRetention.php" - "app/Support/WorkflowTaskFailureMetrics.php" - "app/Support/WorkflowTaskPoller.php" - "app/Support/WorkerCompatibilityHeartbeatRecorder.php" @@ -91,6 +92,7 @@ on: - "app/Support/RuntimeExternalPayloadQuota.php" - "app/Support/WorkerPollClaimGate.php" - "app/Support/WorkflowQueryTaskBroker.php" + - "app/Support/WorkflowStreamRetention.php" - "app/Support/WorkflowTaskFailureMetrics.php" - "app/Support/WorkflowTaskPoller.php" - "app/Support/WorkerCompatibilityHeartbeatRecorder.php" @@ -291,8 +293,8 @@ jobs: # limit per poll. Allow the 1,000 synthetic starts plus SDK canary; # total keys and zero-after-drain remain independently enforced. DW_PERF_MIN_SAMPLE_COVERAGE: "0.75" - DW_PERF_MAX_SERVER_CACHE_KEYS_BY_POLICY: '{"worker_poll_diagnostics":0,"polling_cache_availability_probe":0,"workflow_task_poll_requests":3072,"activity_task_poll_requests":512,"query_task_poll_requests":512,"sqlite_worker_poll_claim_gate":1,"long_poll_signals":2048,"workflow_query_tasks":64,"task_queue_admission_locks":128,"task_queue_dispatch_counters":128,"namespace_request_admission":128,"shared_service_boundary_admission":128,"namespace_durable_state_quota_rejections":128,"runtime_external_payload_quota_rejections":128,"workflow_task_expired_lease_recovery":1536,"history_retention_inline":64,"worker_compatibility_heartbeat":64,"readiness_probe":16}' - DW_PERF_MAX_FINAL_SERVER_CACHE_KEYS_BY_POLICY: '{"worker_poll_diagnostics":0,"polling_cache_availability_probe":0,"workflow_task_poll_requests":0,"activity_task_poll_requests":0,"query_task_poll_requests":0,"sqlite_worker_poll_claim_gate":0,"long_poll_signals":0,"workflow_query_tasks":0,"task_queue_admission_locks":0,"task_queue_dispatch_counters":0,"namespace_request_admission":0,"shared_service_boundary_admission":0,"namespace_durable_state_quota_rejections":0,"runtime_external_payload_quota_rejections":0,"workflow_task_expired_lease_recovery":0,"history_retention_inline":0,"worker_compatibility_heartbeat":0,"readiness_probe":0}' + DW_PERF_MAX_SERVER_CACHE_KEYS_BY_POLICY: '{"worker_poll_diagnostics":0,"polling_cache_availability_probe":0,"workflow_task_poll_requests":3072,"activity_task_poll_requests":512,"query_task_poll_requests":512,"sqlite_worker_poll_claim_gate":1,"long_poll_signals":2048,"workflow_query_tasks":64,"task_queue_admission_locks":128,"task_queue_dispatch_counters":128,"namespace_request_admission":128,"shared_service_boundary_admission":128,"namespace_durable_state_quota_rejections":128,"runtime_external_payload_quota_rejections":128,"workflow_task_expired_lease_recovery":1536,"history_retention_inline":64,"stream_retention_inline":64,"worker_compatibility_heartbeat":64,"readiness_probe":16}' + DW_PERF_MAX_FINAL_SERVER_CACHE_KEYS_BY_POLICY: '{"worker_poll_diagnostics":0,"polling_cache_availability_probe":0,"workflow_task_poll_requests":0,"activity_task_poll_requests":0,"query_task_poll_requests":0,"sqlite_worker_poll_claim_gate":0,"long_poll_signals":0,"workflow_query_tasks":0,"task_queue_admission_locks":0,"task_queue_dispatch_counters":0,"namespace_request_admission":0,"shared_service_boundary_admission":0,"namespace_durable_state_quota_rejections":0,"runtime_external_payload_quota_rejections":0,"workflow_task_expired_lease_recovery":0,"history_retention_inline":0,"stream_retention_inline":0,"worker_compatibility_heartbeat":0,"readiness_probe":0}' DW_PERF_COMPOSE_PROJECT: server-perf-${{ github.run_id }}-${{ github.run_attempt }}-${{ github.job }} RUNNER_ENVIRONMENT: "github-hosted" if: ${{ always() && success() }} diff --git a/app/Http/Controllers/Api/SystemController.php b/app/Http/Controllers/Api/SystemController.php index 7b778b0f3..6ba241280 100644 --- a/app/Http/Controllers/Api/SystemController.php +++ b/app/Http/Controllers/Api/SystemController.php @@ -16,9 +16,11 @@ use App\Support\RuntimeExternalPayloadQuota; use App\Support\TaskQueueBuildIdRolloutSnapshot; use App\Support\WorkerSessionRegistry; +use App\Support\WorkflowStreamRetention; use App\Support\WorkflowTaskFailureMetrics; use Illuminate\Http\JsonResponse; use Illuminate\Http\Request; +use Illuminate\Support\Facades\Log; use Illuminate\Validation\ValidationException; use Workflow\V2\Contracts\MatchingRole; use Workflow\V2\Contracts\OperatorObservabilityRepository; @@ -437,6 +439,22 @@ public function retentionEnforcePass(Request $request): JsonResponse $validated['run_ids'] ?? [], )); + // A general pass also reclaims closed workflow streams past their own + // retention (Server #774): at most $limit streams and $limit * 100 + // items. The report is the run pass's; a stream failure leaves its + // items for the next pass and does not stop the run pass. + if ($runIds === []) { + try { + WorkflowStreamRetention::pruneExpired($namespace, $limit, $limit * 100); + } catch (\Throwable $exception) { + Log::warning('workflow_stream_retention_failed', [ + 'namespace' => $namespace, + 'exception' => $exception::class, + 'message' => $exception->getMessage(), + ]); + } + } + $report = HistoryRetentionEnforcer::runPass($namespace, $limit, $runIds); $retention = HistoryRetentionEnforcer::retentionPolicy($namespace); diff --git a/app/Http/Controllers/Api/WorkflowStreamController.php b/app/Http/Controllers/Api/WorkflowStreamController.php index cd6ab406c..eb2a9af06 100644 --- a/app/Http/Controllers/Api/WorkflowStreamController.php +++ b/app/Http/Controllers/Api/WorkflowStreamController.php @@ -3,17 +3,20 @@ namespace App\Http\Controllers\Api; use App\Models\WorkflowDurableStream; +use App\Models\WorkflowDurableStreamItem; use App\Support\ControlPlaneProtocol; use App\Support\LegacyV1Projection; use App\Support\NamespaceExternalPayloadStorage; use App\Support\NamespaceWorkflowScope; use App\Support\PayloadCodecContract; use App\Support\RuntimeExternalPayloadException; +use App\Support\RuntimeTrackedExternalPayloadStorage; use App\Support\StreamClosedException; use App\Support\StreamErroredException; use App\Support\StreamFullException; use App\Support\StreamNotFoundException; use App\Support\WorkflowStreamService; +use Illuminate\Database\Eloquent\Collection; use Illuminate\Http\JsonResponse; use Illuminate\Http\Request; use Illuminate\Validation\ValidationException; @@ -177,9 +180,15 @@ public function append(Request $request, string $workflowId, string $runId, stri 'items.*.origin_reference' => ['nullable', 'string', 'max:191'], 'items.*.item_type' => ['nullable', 'string', 'max:64'], 'items.*.content_type' => ['nullable', 'string', 'max:191'], - 'max_pending_items' => ['nullable', 'integer', 'min:1'], + 'max_pending_items' => ['nullable', 'integer', 'min:1', 'max:'.WorkflowStreamService::MAX_PENDING_ITEMS_MAX], ]); + // References are fetched and verified before the append and claimed + // only inside its transaction, after the whole batch is admitted, so + // a refused append claims nothing (Server #774). + $driver = null; + $verified = []; + foreach ($payload['items'] as $index => $item) { if (array_key_exists('payload_codec', $item)) { try { @@ -194,9 +203,9 @@ public function append(Request $request, string $workflowId, string $runId, stri } $reference = $item['payload_reference'] ?? null; - if (is_string($reference) && $reference !== '') { - $driver = $this->externalPayloadStorage->driverFor($namespace); - if ($driver === null) { + if (is_string($reference) && $reference !== '' && ! array_key_exists($reference, $verified)) { + $driver ??= $this->externalPayloadStorage->driverFor($namespace); + if (! $driver instanceof RuntimeTrackedExternalPayloadStorage) { throw new RuntimeExternalPayloadException( 'external_payload_unavailable', 503, @@ -205,10 +214,16 @@ public function append(Request $request, string $workflowId, string $runId, stri ); } - $driver->get($reference); + $verified[$reference] = $driver->verify($reference); } } + $claimReferences = $verified === [] ? null : static function (array $references) use ($driver, $verified): void { + foreach ($references as $reference) { + $driver->claimVerified($reference, $verified[$reference]['size_bytes'], $verified[$reference]['sha256']); + } + }; + $maxPending = (int) ($payload['max_pending_items'] ?? WorkflowStreamService::DEFAULT_MAX_PENDING_ITEMS); @@ -219,7 +234,13 @@ public function append(Request $request, string $workflowId, string $runId, stri $streamName, $payload['items'], $maxPending, + $claimReferences, ); + } catch (StreamNotFoundException) { + return ControlPlaneProtocol::jsonForRequest($request, [ + 'message' => 'Stream not found.', + 'reason' => 'stream_not_found', + ], 404); } catch (StreamClosedException) { return ControlPlaneProtocol::jsonForRequest($request, [ 'message' => 'Stream is closed.', @@ -277,7 +298,7 @@ public function close(Request $request, string $workflowId, string $runId, strin $payload = $request->validate([ 'error_reason' => ['nullable', 'string', 'max:191'], - 'retention_seconds' => ['nullable', 'integer', 'min:1'], + 'retention_seconds' => ['nullable', 'integer', 'min:1', 'max:'.WorkflowStreamService::RETENTION_SECONDS_MAX], ]); try { @@ -328,7 +349,7 @@ private function notFound(Request $request): JsonResponse } /** - * @return array{stream: WorkflowDurableStream, items: \Illuminate\Database\Eloquent\Collection, next_offset: int, terminal: bool} + * @return array{stream: WorkflowDurableStream, items: Collection, next_offset: int, terminal: bool} */ private function readWithWait( WorkflowRun $run, diff --git a/app/Models/WorkflowDurableStream.php b/app/Models/WorkflowDurableStream.php index f8dccb6b8..3ea36b151 100644 --- a/app/Models/WorkflowDurableStream.php +++ b/app/Models/WorkflowDurableStream.php @@ -31,6 +31,7 @@ class WorkflowDurableStream extends Model 'opened_at' => 'datetime', 'last_appended_at' => 'datetime', 'closed_at' => 'datetime', + 'reclaim_at' => 'datetime', 'metadata' => 'array', ]; diff --git a/app/Support/ExternalPayloadRetentionCleanup.php b/app/Support/ExternalPayloadRetentionCleanup.php index 426673bd1..fd0036942 100644 --- a/app/Support/ExternalPayloadRetentionCleanup.php +++ b/app/Support/ExternalPayloadRetentionCleanup.php @@ -81,6 +81,37 @@ public function deleteForRuns( ); } + /** + * The external payloads of durable stream items the stream retention + * pass is about to delete (Server #774), except any another retained row + * still references. The items themselves count as released. + * + * @param list $itemIds + * @return array{found: int, deleted: int, blocked: bool, reason: string|null} + */ + public function deleteForDurableStreamItems(string $namespace, array $itemIds): array + { + $itemIds = array_values(array_unique(array_filter( + array_map(static fn (mixed $itemId): int => (int) $itemId, $itemIds), + static fn (int $itemId): bool => $itemId > 0, + ))); + $references = []; + + if ($itemIds !== []) { + $items = WorkflowDurableStreamItem::query() + ->where('namespace', $namespace) + ->whereIn('id', $itemIds); + $this->collectPayloadColumn(clone $items, 'payload', $references); + $this->collectReferenceColumn(clone $items, 'payload_reference', $references); + } + + return $this->deleteReferences( + $namespace, + array_values(array_unique($references)), + fn (string $uri): bool => $this->isReferencedByRetainedRun($uri, [], [], $itemIds), + ); + } + /** * @param list $runIds * @param list $instanceIds @@ -162,11 +193,19 @@ private function deleteReferencesForOwningNamespaces(array $references, callable } $deleted = 0; + $lockOrder = array_flip(RuntimeExternalPayloadObjectLock::inLockOrder(array_column($deletable, 'uri'))); + usort($deletable, static fn (array $left, array $right): int => [$lockOrder[$left['uri']], $left['namespace']] <=> [$lockOrder[$right['uri']], $right['namespace']]); foreach ($deletable as $reference) { try { - $drivers[$reference['namespace']]->delete($reference['uri']); - $deleted++; + $driver = $drivers[$reference['namespace']]; + $uri = $reference['uri']; + if (! $driver instanceof RuntimeTrackedExternalPayloadStorage) { + $driver->delete($uri); + $deleted++; + } elseif ($driver->deleteUnlessReferenced($uri, static fn (): bool => $isRetained($uri))) { + $deleted++; + } } catch (ExternalPayloadStorageUnavailable $e) { throw $e; } catch (\Throwable $e) { @@ -229,10 +268,18 @@ private function deleteReferences(string $namespace, array $references, callable $deleted = 0; - foreach ($deletable as $uri) { + // The retained-reference check is asked again under each object's + // lock, where writers claim, so a reference committed after the scan + // above keeps its object. Objects are locked in one canonical order + // because an enclosing transaction holds every lock until it commits. + foreach (RuntimeExternalPayloadObjectLock::inLockOrder($deletable) as $uri) { try { - $driver->delete($uri); - $deleted++; + if (! $driver instanceof RuntimeTrackedExternalPayloadStorage) { + $driver->delete($uri); + $deleted++; + } elseif ($driver->deleteUnlessReferenced($uri, static fn (): bool => $isRetained($uri))) { + $deleted++; + } } catch (ExternalPayloadStorageUnavailable $e) { throw $e; } catch (\Throwable $e) { @@ -551,6 +598,7 @@ private function isReferencedByRetainedRun( string $uri, array $deletedRunIds, array $releasedInboundStreamItemIds, + array $releasedDurableStreamItemIds = [], ): bool { $runColumns = [ 'arguments', @@ -580,13 +628,14 @@ private function isReferencedByRetainedRun( foreach ($this->retainedPayloadColumns( $deletedRunIds, $releasedInboundStreamItemIds, + $releasedDurableStreamItemIds, ) as [$query, $column]) { if ($this->payloadColumnReferencesUri($query, $column, $uri)) { return true; } } - foreach ($this->retainedReferenceColumns($deletedRunIds) as [$query, $column]) { + foreach ($this->retainedReferenceColumns($deletedRunIds, $releasedDurableStreamItemIds) as [$query, $column]) { if ($this->referenceColumnReferencesUri($query, $column, $uri)) { return true; } @@ -595,6 +644,22 @@ private function isReferencedByRetainedRun( return false; } + /** + * @param list $deletedRunIds + * @param list $releasedItemIds + * @return Builder + */ + private function retainedDurableStreamItems(array $deletedRunIds, array $releasedItemIds): Builder + { + $items = WorkflowDurableStreamItem::query()->whereIn('workflow_run_id', $this->retainedRunIdsQuery($deletedRunIds)); + + if ($releasedItemIds !== []) { + $items->whereNotIn('id', $releasedItemIds); + } + + return $items; + } + /** * @param list $deletedRunIds * @return Builder @@ -610,8 +675,11 @@ private function retainedRunIdsQuery(array $deletedRunIds): Builder * @param list $deletedRunIds * @return list, 1: string}> */ - private function retainedPayloadColumns(array $deletedRunIds, array $releasedInboundStreamItemIds): array - { + private function retainedPayloadColumns( + array $deletedRunIds, + array $releasedInboundStreamItemIds, + array $releasedDurableStreamItemIds = [], + ): array { $retainedInboundStreamItems = WorkflowInboundStreamItem::query(); if ($releasedInboundStreamItemIds !== []) { $retainedInboundStreamItems->whereNotIn('id', $releasedInboundStreamItemIds); @@ -649,7 +717,7 @@ private function retainedPayloadColumns(array $deletedRunIds, array $releasedInb [WorkflowServiceCall::query(), 'caller_principal_claims'], [WorkflowScheduleHistoryEvent::query()->whereIn('workflow_run_id', $this->retainedRunIdsQuery($deletedRunIds)), 'payload'], [WorkflowDurableStream::query()->whereIn('workflow_run_id', $this->retainedRunIdsQuery($deletedRunIds)), 'metadata'], - [WorkflowDurableStreamItem::query()->whereIn('workflow_run_id', $this->retainedRunIdsQuery($deletedRunIds)), 'payload'], + [$this->retainedDurableStreamItems($deletedRunIds, $releasedDurableStreamItemIds), 'payload'], [$retainedInboundStreamItems, 'payload_blob'], [WorkflowRunTimerEntry::query()->whereIn('workflow_run_id', $this->retainedRunIdsQuery($deletedRunIds)), 'payload'], [WorkflowRunWait::query()->whereIn('workflow_run_id', $this->retainedRunIdsQuery($deletedRunIds)), 'payload'], @@ -667,11 +735,11 @@ private function retainedPayloadColumns(array $deletedRunIds, array $releasedInb * @param list $deletedRunIds * @return list, 1: string}> */ - private function retainedReferenceColumns(array $deletedRunIds): array + private function retainedReferenceColumns(array $deletedRunIds, array $releasedDurableStreamItemIds = []): array { $columns = [ [ - WorkflowDurableStreamItem::query()->whereIn('workflow_run_id', $this->retainedRunIdsQuery($deletedRunIds)), + $this->retainedDurableStreamItems($deletedRunIds, $releasedDurableStreamItemIds), 'payload_reference', ], [ diff --git a/app/Support/HistoryRetentionEnforcer.php b/app/Support/HistoryRetentionEnforcer.php index 50ca7bf24..5b92954bb 100644 --- a/app/Support/HistoryRetentionEnforcer.php +++ b/app/Support/HistoryRetentionEnforcer.php @@ -2,10 +2,11 @@ namespace App\Support; +use App\Models\WorkflowDurableStream; +use App\Models\WorkflowDurableStreamItem; use App\Models\WorkflowNamespace; use Illuminate\Support\Carbon; use Illuminate\Support\Facades\Cache; -use Illuminate\Support\Facades\DB; use Illuminate\Support\Facades\Log; use Workflow\V2\Enums\RunStatus; use Workflow\V2\Models\WorkflowRunSummary; @@ -139,6 +140,20 @@ public static function runPass(string $namespace, int $limit = 100, array $runId */ public static function runInlinePass(string $namespace): array { + // Closed workflow streams carry their own retention, whatever the + // namespace's history retention mode (Server #774). The stream pass + // has its own throttle and is not part of this report; a failure + // leaves its items for the next pass instead of failing the caller. + try { + WorkflowStreamRetention::runInlinePass($namespace); + } catch (\Throwable $exception) { + Log::warning('workflow_stream_retention_failed', [ + 'namespace' => $namespace, + 'exception' => $exception::class, + 'message' => $exception->getMessage(), + ]); + } + if (self::retentionMode($namespace) === WorkflowNamespace::RETENTION_MODE_FOREVER) { return [ 'throttled' => false, @@ -212,19 +227,45 @@ public static function pruneRun(string $namespace, string $runId): array (string) $summary->workflow_instance_id, ); + // One transaction, fenced by the run row's lock before anything is + // deleted (Server #774). Stream writes take that lock first, so no + // append to this run can commit between the payload cleanup and the + // row deletes; a later one finds the run's details pruned. Each + // object's retained-reference check is asked again under its own + // lock, reading every reference committed before it. try { - $externalPayloads = app(ExternalPayloadRetentionCleanup::class)->deleteForRun( - $namespace, - $runId, - $streamPlan['releasable_item_ids'], - ); + $outcome = ReadCommittedTransaction::run(static function () use ($namespace, $runId, $messageStreams, $streamPlan): array { + WorkflowRunStreamLock::acquire($runId); + + $externalPayloads = app(ExternalPayloadRetentionCleanup::class)->deleteForRun( + $namespace, + $runId, + $streamPlan['releasable_item_ids'], + ); + + if ($externalPayloads['blocked']) { + return ['payloads' => $externalPayloads, 'report' => null]; + } + + return [ + 'payloads' => $externalPayloads, + 'report' => array_merge( + WorkflowRunRetentionCleanup::pruneRun($runId), + $messageStreams->apply($streamPlan), + self::pruneDurableStreams($runId), + ), + ]; + }); } catch (\Throwable $exception) { $messageStreams->markBlocked($streamPlan, 'external_payload_cleanup_failed'); throw $exception; } - if ($externalPayloads['blocked']) { + $externalPayloads = $outcome['payloads']; + $report = $outcome['report']; + + if ($report === null) { $reason = $externalPayloads['reason'] ?? 'external_payload_cleanup_blocked'; $messageStreams->markBlocked($streamPlan, $reason); @@ -240,13 +281,6 @@ public static function pruneRun(string $namespace, string $runId): array 'closed_at' => $summary->closed_at?->toIso8601String(), ]); - $report = DB::transaction(static function () use ($runId, $messageStreams, $streamPlan): array { - return array_merge( - WorkflowRunRetentionCleanup::pruneRun($runId), - $messageStreams->apply($streamPlan), - ); - }); - return [ 'pruned' => true, 'reason' => null, @@ -257,6 +291,20 @@ public static function pruneRun(string $namespace, string $runId): array ]; } + /** + * The run's durable workflow streams go with the run (Server #774). Their + * external payload references were settled by the payload cleanup. + * + * @return array{durable_stream_items_deleted: int, durable_streams_deleted: int} + */ + private static function pruneDurableStreams(string $runId): array + { + return [ + 'durable_stream_items_deleted' => WorkflowDurableStreamItem::query()->where('workflow_run_id', $runId)->delete(), + 'durable_streams_deleted' => WorkflowDurableStream::query()->where('workflow_run_id', $runId)->delete(), + ]; + } + /** * @return array{pruned: false, reason: string, history_events_deleted: 0, tasks_deleted: 0, external_payloads_deleted: 0, deleted: array} */ diff --git a/app/Support/ReadCommittedTransaction.php b/app/Support/ReadCommittedTransaction.php new file mode 100644 index 000000000..bf226a258 --- /dev/null +++ b/app/Support/ReadCommittedTransaction.php @@ -0,0 +1,55 @@ +getDriverName(); + + if ($connection->transactionLevel() > 0 || ! in_array($driver, ['mysql', 'mariadb', 'pgsql'], true)) { + return $connection->transaction($callback); + } + + if ($driver !== 'pgsql') { + // Applies to the next transaction the session starts. + $connection->unprepared('SET TRANSACTION ISOLATION LEVEL READ COMMITTED'); + + return $connection->transaction($callback); + } + + return $connection->transaction(static function () use ($connection, $callback): mixed { + // Must be the transaction's first statement. + $connection->unprepared('SET TRANSACTION ISOLATION LEVEL READ COMMITTED'); + + return $callback(); + }); + } +} diff --git a/app/Support/RuntimeExternalPayloadObjectLock.php b/app/Support/RuntimeExternalPayloadObjectLock.php index a9ad8701c..411b6c155 100644 --- a/app/Support/RuntimeExternalPayloadObjectLock.php +++ b/app/Support/RuntimeExternalPayloadObjectLock.php @@ -10,12 +10,37 @@ final class RuntimeExternalPayloadObjectLock { public const BUCKETS = 256; + public static function bucketFor(string $uri): int + { + return (int) (hexdec(substr(hash('sha256', $uri), 0, 4)) % self::BUCKETS); + } + + /** + * URIs in the order their lock rows must be taken when one transaction + * holds several: by bucket, then by URI. Every caller that locks more + * than one object uses it, so two transactions never wait on each other + * in opposite orders (Server #774). + * + * @param list $uris + * @return list + */ + public static function inLockOrder(array $uris): array + { + $uris = array_values(array_unique($uris)); + usort($uris, static fn (string $left, string $right): int => [self::bucketFor($left), $left] <=> [self::bucketFor($right), $right]); + + return $uris; + } + /** * Serialize registry and backing-object mutations for one provider URI. * * The fixed-size lock table avoids an unbounded lock-row registry. The * no-op update also obtains SQLite's writer lock, where lockForUpdate() is - * otherwise ignored. + * otherwise ignored. There it comes first: a SQLite transaction that has + * read cannot wait for another writer, it fails at once, so reading the + * lock row first made a concurrent payload operation fail instead of + * waiting its turn (Server #774). * * @template TReturn * @@ -24,12 +49,17 @@ final class RuntimeExternalPayloadObjectLock */ public function transaction(string $uri, Closure $callback, ?Closure $beforeLock = null): mixed { - $bucket = hexdec(substr(hash('sha256', $uri), 0, 4)) % self::BUCKETS; + $bucket = self::bucketFor($uri); 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(); + $sqlite = DB::connection()->getDriverName() === 'sqlite'; + if ($sqlite) { + $this->touch($bucket); + } + $lock = DB::table('runtime_external_payload_object_locks') ->where('bucket', $bucket) ->lockForUpdate() @@ -39,11 +69,18 @@ public function transaction(string $uri, Closure $callback, ?Closure $beforeLock throw new RuntimeException('Runtime external payload object lock table is not initialized.'); } - DB::table('runtime_external_payload_object_locks') - ->where('bucket', $bucket) - ->update(['bucket' => $bucket]); + if (! $sqlite) { + $this->touch($bucket); + } return $callback(); }, 3); } + + private function touch(int $bucket): void + { + DB::table('runtime_external_payload_object_locks') + ->where('bucket', $bucket) + ->update(['bucket' => $bucket]); + } } diff --git a/app/Support/RuntimeExternalPayloadRegistry.php b/app/Support/RuntimeExternalPayloadRegistry.php index 8dd2e2536..190757d3d 100644 --- a/app/Support/RuntimeExternalPayloadRegistry.php +++ b/app/Support/RuntimeExternalPayloadRegistry.php @@ -213,14 +213,57 @@ public function fetch(string $namespace, array $transportReference, bool $stream } public function verifyFetchedBytesAndClaim(string $namespace, string $uri, string|ExternalPayloadStream $data): void + { + [$observedSize, $observedHash] = $this->observedBytes($data); + $this->checkRegisteredBytes($namespace, $uri, $observedSize, $observedHash, claim: true); + } + + /** + * Checks fetched bytes against their registration without claiming the + * reference, so a caller can refuse early and claim later, inside the + * transaction that records the reference. + * + * @return array{size_bytes: int, sha256: string} + */ + public function verifyFetchedBytes(string $namespace, string $uri, string|ExternalPayloadStream $data): array + { + [$observedSize, $observedHash] = $this->observedBytes($data); + $this->checkRegisteredBytes($namespace, $uri, $observedSize, $observedHash, claim: false); + + return ['size_bytes' => $observedSize, 'sha256' => $observedHash]; + } + + /** + * Claims a reference whose bytes verifyFetchedBytes() checked, checking + * its registration again under the object lock. Inside a caller's + * transaction the claim rolls back with it. + */ + public function claimVerified(string $namespace, string $uri, int $sizeBytes, string $sha256): void + { + $this->checkRegisteredBytes($namespace, $uri, $sizeBytes, strtolower($sha256), claim: true); + } + + /** @return array{0: int, 1: string} */ + private function observedBytes(string|ExternalPayloadStream $data): array { $observedSize = $data instanceof ExternalPayloadStream ? $data->sizeBytes : strlen($data); $observedHash = $data instanceof ExternalPayloadStream ? $data->sha256 : hash('sha256', $data); $this->assertSize($observedSize); + + return [$observedSize, $observedHash]; + } + + private function checkRegisteredBytes( + string $namespace, + string $uri, + int $observedSize, + string $observedHash, + bool $claim, + ): void { $namespace = $this->namespace($namespace); try { - $this->objectLock->transaction($uri, function () use ($namespace, $uri, $observedSize, $observedHash): void { + $this->objectLock->transaction($uri, function () use ($namespace, $uri, $observedSize, $observedHash, $claim): void { $row = RuntimeExternalPayload::query() ->where('namespace', $namespace) ->where('storage_uri_sha256', hash('sha256', $uri)) @@ -246,6 +289,10 @@ public function verifyFetchedBytesAndClaim(string $namespace, string $uri, strin throw $this->integrityMismatch('Fetched external payload bytes failed runtime integrity verification.'); } + if (! $claim) { + return; + } + $row->forceFill([ 'retained_at' => $row->retained_at ?? now(), 'expires_at' => null, @@ -281,7 +328,7 @@ public function deleteForNamespace(string $namespace): int $deleted = 0; foreach ($rows as $row) { - $deleted += $this->deleteRegisteredUri($namespace, $row->storage_uri, $driver); + $deleted += (int) $this->deleteRegisteredUri($namespace, $row->storage_uri, $driver); } return $deleted; @@ -295,13 +342,91 @@ public function deleteUri( $this->deleteRegisteredUri($this->namespace($namespace), $uri, $driver); } + /** + * Deletes the object unless `$isReferenced` finds a reference to it, + * asked under the object's lock (Server #774). A writer claims under the + * same lock, so either the writer's reference commits first and the + * object stays, or the object goes first and the writer's claim fails. + * Returns false when a reference kept the object. + * + * @param callable(): bool $isReferenced + */ + public function deleteUriUnlessReferenced( + string $namespace, + string $uri, + RuntimeExternalPayloadStorageDriver $driver, + callable $isReferenced, + ): bool { + return $this->deleteRegisteredUri($this->namespace($namespace), $uri, $driver, $isReferenced) !== null; + } + + /** + * Claims a reference the runtime transport already resolved to this + * namespace's registration, without fetching its bytes again, checking + * the registration under the object lock. Inside a caller's transaction + * the claim rolls back with it. + */ + public function claimRegistered(string $namespace, string $uri): void + { + $namespace = $this->namespace($namespace); + + try { + $this->objectLock->transaction($uri, function () use ($namespace, $uri): void { + $row = RuntimeExternalPayload::query() + ->where('namespace', $namespace) + ->where('storage_uri_sha256', hash('sha256', $uri)) + ->where('storage_uri', $uri) + ->where('upload_status', RuntimeExternalPayload::UPLOAD_READY) + ->lockForUpdate() + ->first(); + + if ($row === null) { + throw new RuntimeExternalPayloadException( + 'external_payload_not_found', + 404, + false, + 'External payload reference was not found in this namespace.', + ); + } + + if ($row->retained_at === null && $row->expires_at !== null && $row->expires_at->isPast()) { + throw new RuntimeExternalPayloadException( + 'external_payload_expired', + 410, + false, + 'External payload reference has expired.', + ); + } + + $row->forceFill([ + 'retained_at' => $row->retained_at ?? now(), + 'expires_at' => null, + ])->save(); + }); + } catch (RuntimeExternalPayloadException $exception) { + throw $exception; + } catch (Throwable $exception) { + throw $this->unavailable('External payload reference claim failed.', $exception); + } + } + + /** + * @param (callable(): bool)|null $isReferenced + * @return int|null The namespace registrations removed, or null when a + * reference kept the object. + */ private function deleteRegisteredUri( string $namespace, string $uri, ?RuntimeExternalPayloadStorageDriver $driver, - ): int { + ?callable $isReferenced = null, + ): ?int { try { - return $this->objectLock->transaction($uri, function () use ($namespace, $uri, $driver): int { + return $this->objectLock->transaction($uri, function () use ($namespace, $uri, $driver, $isReferenced): ?int { + if ($isReferenced !== null && $isReferenced()) { + return null; + } + $owners = RuntimeExternalPayload::query() ->where('storage_uri_sha256', hash('sha256', $uri)) ->where('storage_uri', $uri) diff --git a/app/Support/RuntimeTrackedExternalPayloadStorage.php b/app/Support/RuntimeTrackedExternalPayloadStorage.php index 50d134a24..b0dc4d3bb 100644 --- a/app/Support/RuntimeTrackedExternalPayloadStorage.php +++ b/app/Support/RuntimeTrackedExternalPayloadStorage.php @@ -56,9 +56,58 @@ public function retainEnvelope(array $envelope, ?string $validationField = null) } public function get(string $uri): string + { + $data = $this->fetch($uri); + + app(RuntimeExternalPayloadRegistry::class)->verifyFetchedBytesAndClaim( + $this->namespace, + $uri, + $data, + ); + + return $data; + } + + /** + * Fetches and verifies the bytes behind $uri without claiming it. Claim + * it with claimVerified() in the transaction that records the reference. + * + * @return array{size_bytes: int, sha256: string} + */ + public function verify(string $uri): array + { + return app(RuntimeExternalPayloadRegistry::class)->verifyFetchedBytes( + $this->namespace, + $uri, + $this->fetch($uri), + ); + } + + public function claimVerified(string $uri, int $sizeBytes, string $sha256): void + { + app(RuntimeExternalPayloadRegistry::class)->claimVerified($this->namespace, $uri, $sizeBytes, $sha256); + } + + public function claimRegistered(string $uri): void + { + app(RuntimeExternalPayloadRegistry::class)->claimRegistered($this->namespace, $uri); + } + + /** + * Deletes $uri unless `$isReferenced`, asked under the object's lock, + * finds a reference to it. Returns false when one kept it. + * + * @param callable(): bool $isReferenced + */ + public function deleteUnlessReferenced(string $uri, callable $isReferenced): bool + { + return app(RuntimeExternalPayloadRegistry::class)->deleteUriUnlessReferenced($this->namespace, $uri, $this->inner, $isReferenced); + } + + private function fetch(string $uri): string { try { - $data = $this->inner->get($uri); + return $this->inner->get($uri); } catch (ExternalPayloadObjectOversized $exception) { throw new RuntimeExternalPayloadException( 'external_payload_oversized', @@ -84,14 +133,6 @@ public function get(string $uri): string $exception, ); } - - app(RuntimeExternalPayloadRegistry::class)->verifyFetchedBytesAndClaim( - $this->namespace, - $uri, - $data, - ); - - return $data; } public function delete(string $uri): void diff --git a/app/Support/WorkflowRunStreamLock.php b/app/Support/WorkflowRunStreamLock.php new file mode 100644 index 000000000..38ce4691d --- /dev/null +++ b/app/Support/WorkflowRunStreamLock.php @@ -0,0 +1,41 @@ +getModel()->getTable(); + + if ($connection->getDriverName() === 'sqlite') { + return $connection->update( + 'UPDATE '.$connection->getQueryGrammar()->wrapTable($table).' SET id = id WHERE id = ?', + [$runId], + ) > 0; + } + + return $connection->table($table)->where('id', $runId)->lockForUpdate()->first(['id']) !== null; + } +} diff --git a/app/Support/WorkflowStreamCommandProcessor.php b/app/Support/WorkflowStreamCommandProcessor.php index bf10a0c92..fa312af90 100644 --- a/app/Support/WorkflowStreamCommandProcessor.php +++ b/app/Support/WorkflowStreamCommandProcessor.php @@ -18,6 +18,7 @@ final class WorkflowStreamCommandProcessor { public function __construct( private readonly WorkflowStreamService $streams, + private readonly NamespaceExternalPayloadStorage $externalPayloadStorage, ) {} /** @@ -136,6 +137,7 @@ public function process(string $taskId, string $namespace, array $commands): arr isset($directive['max_pending_items']) ? (int) $directive['max_pending_items'] : WorkflowStreamService::DEFAULT_MAX_PENDING_ITEMS, + $this->claimReferences($namespace), ); } elseif ($operation === 'close' || $operation === 'error') { $errorReason = $operation === 'error' @@ -170,6 +172,35 @@ public function process(string $taskId, string $namespace, array $commands): arr return array_values($commands); } + /** + * Claims the references of the items an append will write, as the HTTP + * append does, under each object's lock inside the append's transaction + * (Server #774). The runtime transport already resolved each one to this + * namespace's registration; the claim checks it again and keeps it from + * expiring, and stream or run retention cannot delete it meanwhile. + * + * @return callable(list): void + */ + private function claimReferences(string $namespace): callable + { + return function (array $uris) use ($namespace): void { + $driver = $this->externalPayloadStorage->driverFor($namespace); + + if (! $driver instanceof RuntimeTrackedExternalPayloadStorage) { + throw new RuntimeExternalPayloadException( + 'external_payload_unavailable', + 503, + true, + 'External payload storage is unavailable for this namespace.', + ); + } + + foreach ($uris as $uri) { + $driver->claimRegistered($uri); + } + }; + } + private function nonEmptyString(mixed $value): ?string { if (! is_string($value) || trim($value) === '') { diff --git a/app/Support/WorkflowStreamRetention.php b/app/Support/WorkflowStreamRetention.php new file mode 100644 index 000000000..0088ae435 --- /dev/null +++ b/app/Support/WorkflowStreamRetention.php @@ -0,0 +1,191 @@ + 0, 'items_removed' => 0, 'budget_exhausted' => false, 'blocked_reason' => null]; + + foreach (self::dueStreams($namespace, max(1, $streamLimit), $now) as $stream) { + $step = self::reclaimStream($namespace, $stream, max(1, $itemBudget) - $report['items_removed'], $now); + $report['items_removed'] += $step['deleted']; + + if ($step['reclaimed']) { + $report['streams_reclaimed']++; + } + + if ($step['blocked_reason'] !== null || $step['budget_exhausted']) { + $report['blocked_reason'] = $step['blocked_reason']; + $report['budget_exhausted'] = $step['budget_exhausted']; + + break; + } + } + + return $report; + } + + /** + * A small pass from ordinary worker traffic, at most once a minute per + * namespace, so expired streams are reclaimed without a scheduler. + * + * @return array{streams_reclaimed: int, items_removed: int, budget_exhausted: bool, blocked_reason: string|null}|null + */ + public static function runInlinePass(string $namespace): ?array + { + if (! Cache::add(self::INLINE_CACHE_PREFIX.hash('sha256', $namespace), '1', now()->addSeconds(self::INLINE_THROTTLE_SECONDS))) { + return null; + } + + return self::pruneExpired($namespace, self::INLINE_STREAM_LIMIT, self::INLINE_ITEM_BUDGET); + } + + /** + * The namespace's due streams, oldest deadline first. + * + * @return list + */ + public static function dueStreams(string $namespace, int $limit, Carbon $now): array + { + return self::dueStreamsQuery($namespace, $limit, $now)->get()->all(); + } + + /** + * The predicate and order match the (namespace, reclaim_at, id) index, so + * the read is an index range scan that stops at the limit on every + * engine. + * + * @return Builder + */ + public static function dueStreamsQuery(string $namespace, int $limit, Carbon $now): Builder + { + return WorkflowDurableStream::query() + ->select(['id', 'workflow_run_id']) + ->where('namespace', $namespace) + ->where('reclaim_at', '<=', $now) + ->orderBy('reclaim_at') + ->orderBy('id') + ->limit($limit); + } + + /** + * @return array{deleted: int, reclaimed: bool, budget_exhausted: bool, blocked_reason: string|null} + */ + private static function reclaimStream(string $namespace, WorkflowDurableStream $stream, int $budget, Carbon $now): array + { + $deleted = 0; + + while ($deleted < $budget) { + $step = ReadCommittedTransaction::run(static fn (): array => self::reclaimBatch( + $namespace, + $stream, + min(self::ITEM_BATCH, $budget - $deleted), + $now, + )); + $deleted += $step['deleted']; + + if ($step['blocked_reason'] !== null) { + return ['deleted' => $deleted, 'reclaimed' => false, 'budget_exhausted' => false, 'blocked_reason' => $step['blocked_reason']]; + } + + if ($step['done'] || $step['deleted'] === 0) { + return ['deleted' => $deleted, 'reclaimed' => $step['reclaimed'], 'budget_exhausted' => false, 'blocked_reason' => null]; + } + } + + return ['deleted' => $deleted, 'reclaimed' => false, 'budget_exhausted' => true, 'blocked_reason' => null]; + } + + /** + * One transaction, under the run and stream locks taken before anything + * else: settle a batch's external payloads, delete its items, and clear + * the stream's deadline once none are left. A blocked or failed payload + * cleanup deletes no item; the failure rolls the transaction back. + * + * @return array{deleted: int, done: bool, reclaimed: bool, blocked_reason: string|null} + */ + private static function reclaimBatch(string $namespace, WorkflowDurableStream $due, int $take, Carbon $now): array + { + WorkflowRunStreamLock::acquire((string) $due->workflow_run_id); + $stream = WorkflowDurableStream::query()->whereKey($due->id)->lockForUpdate()->first(); + + if ($stream === null || $stream->reclaim_at === null || ! WorkflowStreamService::isExpired($stream, $now)) { + return ['deleted' => 0, 'done' => true, 'reclaimed' => false, 'blocked_reason' => null]; + } + + $ids = WorkflowDurableStreamItem::query() + ->where('stream_id', $stream->id) + ->orderBy('offset') + ->limit($take) + ->pluck('id') + ->map(static fn (mixed $id): int => (int) $id) + ->all(); + $deleted = 0; + + if ($ids !== []) { + $payloads = app(ExternalPayloadRetentionCleanup::class)->deleteForDurableStreamItems($namespace, $ids); + + if ($payloads['blocked']) { + return [ + 'deleted' => 0, + 'done' => false, + 'reclaimed' => false, + 'blocked_reason' => $payloads['reason'] ?? 'external_payload_cleanup_blocked', + ]; + } + + $deleted = WorkflowDurableStreamItem::query()->where('stream_id', $stream->id)->whereKey($ids)->delete(); + } + + if (WorkflowDurableStreamItem::query()->where('stream_id', $stream->id)->exists()) { + return ['deleted' => $deleted, 'done' => false, 'reclaimed' => false, 'blocked_reason' => null]; + } + + // The terminal row keeps the stream's identity until its run is + // pruned; without a deadline discovery no longer reads it. + $stream->forceFill(['reclaim_at' => null])->save(); + + return ['deleted' => $deleted, 'done' => true, 'reclaimed' => true, 'blocked_reason' => null]; + } +} diff --git a/app/Support/WorkflowStreamService.php b/app/Support/WorkflowStreamService.php index 674214e53..07b77b50d 100644 --- a/app/Support/WorkflowStreamService.php +++ b/app/Support/WorkflowStreamService.php @@ -31,6 +31,18 @@ final class WorkflowStreamService public const DEFAULT_MAX_PENDING_ITEMS = 10000; + /** + * The highest pending-items cap a producer may ask for (Server #774): a + * producer may lower its stream's backpressure threshold, not raise it. + */ + public const MAX_PENDING_ITEMS_MAX = self::DEFAULT_MAX_PENDING_ITEMS; + + /** + * The longest closed-stream retention a producer may ask for (30 days). + * It fits every engine's retention column. + */ + public const RETENTION_SECONDS_MAX = 2592000; + public const DEFAULT_MAX_ITEMS_PER_APPEND = 500; public const DEFAULT_MAX_ITEMS_PER_SUBSCRIBE = 100; @@ -42,7 +54,13 @@ final class WorkflowStreamService public const DEFAULT_CLOSED_RETENTION_SECONDS = 3600; /** + * `$claimReferences` receives the distinct `payload_reference` values of + * the items this append will write, after the whole batch is admitted and + * before any item is written, inside the append's transaction. A claim it + * records rolls back with a refused or failed append. + * * @param list> $items + * @param (callable(list): void)|null $claimReferences * @return array{stream: WorkflowDurableStream, items: list, accepted: int, deduped: int} */ public function append( @@ -51,6 +69,7 @@ public function append( string $streamName, array $items, int $maxPendingItems = self::DEFAULT_MAX_PENDING_ITEMS, + ?callable $claimReferences = null, ): array { $this->assertValidStreamName($streamName); @@ -65,13 +84,28 @@ public function append( )); } - $result = DB::transaction(function () use ($run, $namespace, $streamName, $items, $maxPendingItems) { + if ($maxPendingItems < 1 || $maxPendingItems > self::MAX_PENDING_ITEMS_MAX) { + throw new InvalidArgumentException(sprintf( + 'max_pending_items must be between 1 and %d', + self::MAX_PENDING_ITEMS_MAX, + )); + } + + return DB::transaction(function () use ($run, $namespace, $streamName, $items, $maxPendingItems, $claimReferences) { $stream = $this->lockStream($run, $namespace, $streamName, openIfMissing: true); if ($stream === null) { throw new RuntimeException('Stream lookup must return a row when openIfMissing is true'); } + // Past its retention a terminal stream reads as not found for + // every operation, append included, from the expiry instant on + // (Server #774). Its row stays until its run is pruned, so the + // name is never opened again inside the run. + if (self::isExpired($stream)) { + throw new StreamNotFoundException($run, $streamName); + } + if ($stream->status === WorkflowDurableStream::STATUS_CLOSED) { throw new StreamClosedException($stream); } @@ -80,11 +114,12 @@ public function append( throw new StreamErroredException($stream); } - $now = Carbon::now(); - $accepted = []; - $deduped = 0; - $offset = (int) $stream->last_offset; - $streamFull = false; + // The whole batch is admitted or refused before any item is + // written (Server #774). Items whose idempotency key is already + // in the stream, or earlier in this batch, take no capacity. + $plan = []; + $seen = []; + $fresh = 0; foreach ($items as $item) { if (! is_array($item)) { @@ -94,25 +129,77 @@ public function append( $idempotencyKey = isset($item['idempotency_key']) ? (string) $item['idempotency_key'] : null; $idempotencyKey = ($idempotencyKey === '' ? null : $idempotencyKey); + if ($idempotencyKey !== null && array_key_exists($idempotencyKey, $seen)) { + $plan[] = ['key' => $idempotencyKey, 'existing' => null, 'item' => $item, 'repeat' => true]; + + continue; + } + + $existing = null; + if ($idempotencyKey !== null) { + $seen[$idempotencyKey] = true; /** @var WorkflowDurableStreamItem|null $existing */ $existing = WorkflowDurableStreamItem::query() ->where('stream_id', $stream->id) ->where('idempotency_key', $idempotencyKey) ->first(); + } + + if ($existing === null) { + $fresh++; + } + + $plan[] = ['key' => $idempotencyKey, 'existing' => $existing, 'item' => $item, 'repeat' => false]; + } + + if ($fresh > 0 && (int) $stream->pending_items + $fresh > $maxPendingItems) { + // Rolls back the whole append, including a stream this + // request opened. + throw new StreamFullException($stream, $maxPendingItems); + } + + if ($claimReferences !== null) { + $references = []; - if ($existing !== null) { - $accepted[] = $existing; - $deduped++; + foreach ($plan as $entry) { + $reference = $entry['item']['payload_reference'] ?? null; - continue; + if (! $entry['repeat'] && $entry['existing'] === null && is_string($reference) && $reference !== '') { + $references[$reference] = true; } } - if ($stream->pending_items >= $maxPendingItems) { - $streamFull = true; + if ($references !== []) { + // Each claim holds its object's lock until the append + // commits; one canonical order keeps two appends with + // shared references from waiting on each other. + $claimReferences(RuntimeExternalPayloadObjectLock::inLockOrder(array_map('strval', array_keys($references)))); + } + } + + $now = Carbon::now(); + $accepted = []; + $byKey = []; + $deduped = 0; + $offset = (int) $stream->last_offset; + + foreach ($plan as $entry) { + $idempotencyKey = $entry['key']; + $item = $entry['item']; + + if ($entry['repeat']) { + $accepted[] = $byKey[$idempotencyKey]; + $deduped++; + + continue; + } + + if ($entry['existing'] !== null) { + $accepted[] = $byKey[$idempotencyKey] = $entry['existing']; + $deduped++; - break; + continue; } $offset++; @@ -140,6 +227,10 @@ public function append( $stream->last_appended_at = $now; $accepted[] = $created; + + if ($idempotencyKey !== null) { + $byKey[$idempotencyKey] = $created; + } } $stream->save(); @@ -149,17 +240,8 @@ public function append( 'items' => $accepted, 'accepted' => count($accepted) - $deduped, 'deduped' => $deduped, - 'stream_full' => $streamFull, ]; }); - - if (($result['stream_full'] ?? false) === true) { - throw new StreamFullException($result['stream'], $maxPendingItems); - } - - unset($result['stream_full']); - - return $result; } /** @@ -233,10 +315,17 @@ public function close( ): WorkflowDurableStream { $this->assertValidStreamName($streamName); + if ($retentionSeconds !== null && ($retentionSeconds < 1 || $retentionSeconds > self::RETENTION_SECONDS_MAX)) { + throw new InvalidArgumentException(sprintf( + 'retention_seconds must be between 1 and %d', + self::RETENTION_SECONDS_MAX, + )); + } + return DB::transaction(function () use ($run, $namespace, $streamName, $errorReason, $retentionSeconds) { $stream = $this->lockStream($run, $namespace, $streamName, openIfMissing: false); - if ($stream === null) { + if ($stream === null || self::isExpired($stream)) { throw new StreamNotFoundException($run, $streamName); } @@ -256,6 +345,11 @@ public function close( $stream->retention_seconds = self::DEFAULT_CLOSED_RETENTION_SECONDS; } + // The retention pass's deadline: the instant the stream expires, + // from the whole-second close time the row stores. + $stream->reclaim_at = $stream->closed_at->copy()->startOfSecond() + ->addSeconds((int) $stream->retention_seconds); + $stream->save(); return $stream->refresh(); @@ -267,11 +361,31 @@ public function close( */ public function listForRun(string $namespace, string $workflowRunId): Collection { + $now = Carbon::now(); + return WorkflowDurableStream::query() ->where('namespace', $namespace) ->where('workflow_run_id', $workflowRunId) ->orderBy('stream_name') - ->get(); + ->get() + ->reject(static fn (WorkflowDurableStream $stream): bool => self::isExpired($stream, $now)) + ->values(); + } + + /** + * Whether a closed or errored stream is past `closed_at` plus its + * retention (3600 seconds when none was recorded). An expired stream + * reads as not found; the retention pass reclaims its rows. + */ + public static function isExpired(WorkflowDurableStream $stream, ?Carbon $now = null): bool + { + if (! $stream->isTerminal() || $stream->closed_at === null) { + return false; + } + + $retention = $stream->retention_seconds ?? self::DEFAULT_CLOSED_RETENTION_SECONDS; + + return $stream->closed_at->copy()->addSeconds((int) $retention)->lessThanOrEqualTo($now ?? Carbon::now()); } public function describe( @@ -295,11 +409,13 @@ private function locateStream( string $namespace, string $streamName, ): ?WorkflowDurableStream { - return WorkflowDurableStream::query() + $stream = WorkflowDurableStream::query() ->where('namespace', $namespace) ->where('workflow_run_id', $run->id) ->where('stream_name', $streamName) ->first(); + + return $stream === null || self::isExpired($stream) ? null : $stream; } private function lockStream( @@ -308,6 +424,13 @@ private function lockStream( string $streamName, bool $openIfMissing, ): ?WorkflowDurableStream { + // A run pruned since it was located, or whose details retention has + // pruned, has no streams to write. + if (! WorkflowRunStreamLock::acquire($run->id) + || $run->newQuery()->whereKey($run->getKey())->whereNotNull('details_pruned_at')->exists()) { + throw new StreamNotFoundException($run, $streamName); + } + $stream = WorkflowDurableStream::query() ->where('namespace', $namespace) ->where('workflow_run_id', $run->id) @@ -315,14 +438,12 @@ private function lockStream( ->lockForUpdate() ->first(); - if ($stream !== null) { + if ($stream !== null || ! $openIfMissing) { return $stream; } - if (! $openIfMissing) { - return null; - } - + // The run lock admits one opener at a time, so the stream is created + // through its model and the namespace stream quota applies. return WorkflowDurableStream::query()->create([ 'namespace' => $namespace, 'workflow_instance_id' => $run->workflow_instance_id, diff --git a/app/Support/WorkflowStreamsContract.php b/app/Support/WorkflowStreamsContract.php index d18700d56..7feb485de 100644 --- a/app/Support/WorkflowStreamsContract.php +++ b/app/Support/WorkflowStreamsContract.php @@ -136,12 +136,17 @@ public static function manifest(): array 'max_items_per_subscribe_default' => 100, 'max_items_per_append_max' => 500, 'max_pending_items_per_stream_default' => 10000, + 'max_pending_items_per_stream_max' => WorkflowStreamService::MAX_PENDING_ITEMS_MAX, + 'capacity_refusal' => 'whole_batch_refused_before_any_item_is_written', ], 'retention' => [ 'open_stream' => 'kept_until_close', 'closed_stream_default_retention_seconds' => 3600, 'errored_stream_default_retention_seconds' => 3600, 'retention_override_field' => 'retention_seconds', + 'retention_seconds_max' => WorkflowStreamService::RETENTION_SECONDS_MAX, + 'expired_stream_reads' => 'not_found', + 'expired_stream_reclaim' => 'bounded_retention_pass', ], 'observability_surface' => [ 'list_streams_route' => 'GET /api/workflows/{workflowId}/runs/{runId}/streams', diff --git a/app/Support/WorkflowTaskCompletionValidation.php b/app/Support/WorkflowTaskCompletionValidation.php index 15067ca14..a256e19c9 100644 --- a/app/Support/WorkflowTaskCompletionValidation.php +++ b/app/Support/WorkflowTaskCompletionValidation.php @@ -144,9 +144,9 @@ public static function rules(): array 'commands.*.workflow_stream.items.*.idempotency_key' => ['nullable', 'string', 'max:191'], 'commands.*.workflow_stream.items.*.item_type' => ['nullable', 'string', 'max:64'], 'commands.*.workflow_stream.items.*.content_type' => ['nullable', 'string', 'max:191'], - 'commands.*.workflow_stream.max_pending_items' => ['nullable', 'integer', 'min:1'], + 'commands.*.workflow_stream.max_pending_items' => ['nullable', 'integer', 'min:1', 'max:'.WorkflowStreamService::MAX_PENDING_ITEMS_MAX], 'commands.*.workflow_stream.error_reason' => ['nullable', 'string', 'max:191'], - 'commands.*.workflow_stream.retention_seconds' => ['nullable', 'integer', 'min:1'], + 'commands.*.workflow_stream.retention_seconds' => ['nullable', 'integer', 'min:1', 'max:'.WorkflowStreamService::RETENTION_SECONDS_MAX], 'sticky_cache' => ['nullable', 'array'], 'sticky_cache.worker_id' => ['required_with:sticky_cache', 'string', 'max:255'], 'sticky_cache.workflow_id' => ['required_with:sticky_cache', 'string', 'max:255'], diff --git a/config/dw-bounded-growth.php b/config/dw-bounded-growth.php index c701b3888..575cf5128 100644 --- a/config/dw-bounded-growth.php +++ b/config/dw-bounded-growth.php @@ -18,6 +18,7 @@ use App\Support\WorkerPollClaimGate; use App\Support\WorkerPollDiagnostics; use App\Support\WorkflowQueryTaskBroker; +use App\Support\WorkflowStreamRetention; use App\Support\WorkflowTaskFailureMetrics; use App\Support\WorkflowTaskPoller; use App\Support\WorkflowTaskPollRequestStore; @@ -287,6 +288,18 @@ 'eviction' => 'Cache TTL only. Expired run discovery stays in SQL and no cache index is retained.', ], + 'stream_retention_inline' => [ + 'owner' => WorkflowStreamRetention::class, + 'prefix' => 'server:stream-retention-inline:', + 'dimensions' => [ + 'namespace_hash', + ], + 'ttl' => '60 seconds.', + 'bound' => 'One short-lived throttle key per namespace receiving worker heartbeats during the TTL window.', + 'admission' => 'Cache add elects at most one worker heartbeat per namespace per minute to run a bounded expired-stream pass.', + 'eviction' => 'Cache TTL only. Expired stream discovery stays in SQL and no cache index is retained.', + ], + 'worker_compatibility_heartbeat' => [ 'owner' => WorkerCompatibilityHeartbeatRecorder::class, 'prefix' => 'server:worker-compatibility-heartbeat:', diff --git a/database/migrations/2026_10_26_000100_add_workflow_durable_stream_reclaim_at.php b/database/migrations/2026_10_26_000100_add_workflow_durable_stream_reclaim_at.php new file mode 100644 index 000000000..ad962abe7 --- /dev/null +++ b/database/migrations/2026_10_26_000100_add_workflow_durable_stream_reclaim_at.php @@ -0,0 +1,45 @@ +timestamp('reclaim_at', 6)->nullable(); + // The retention pass reads one namespace's due streams in + // deadline order from this index. + $table->index(['namespace', 'reclaim_at', 'id'], 'wf_durable_streams_ns_reclaim_idx'); + }); + + DB::table('workflow_durable_streams') + ->whereIn('status', ['closed', 'errored']) + ->whereNotNull('closed_at') + ->orderBy('id') + ->chunkById(500, function ($streams): void { + foreach ($streams as $stream) { + DB::table('workflow_durable_streams') + ->where('id', $stream->id) + ->update(['reclaim_at' => Carbon::parse($stream->closed_at, 'UTC') + ->addSeconds((int) ($stream->retention_seconds ?? 3600))]); + } + }); + } + + public function down(): void + { + Schema::table('workflow_durable_streams', function (Blueprint $table) { + $table->dropIndex('wf_durable_streams_ns_reclaim_idx'); + $table->dropColumn('reclaim_at'); + }); + } +}; diff --git a/docs/bounded-growth.md b/docs/bounded-growth.md index 94f53fd87..45f0c2779 100644 --- a/docs/bounded-growth.md +++ b/docs/bounded-growth.md @@ -77,6 +77,7 @@ marked run without recreating its deleted summary. Check `/api/system/health`, | `namespace_durable_state_quota_rejections` | `server:namespace-durable-state:` | `App\Support\NamespaceDurableStateQuota` | One two-minute counter and one log-suppression key per namespace and fixed durable-state rejection reason; SQL rows remain the quota authority. | | `workflow_task_expired_lease_recovery` | `server:workflow-task-expired-lease-recovery:` | `App\Support\WorkflowTaskPoller` | Expired-task recovery scans are capped by `server.polling.expired_workflow_task_recovery_scan_limit`, default 5, and duplicate recovery attempts are TTL-suppressed per task. | | `history_retention_inline` | `server:history-retention-inline:` | `App\Support\HistoryRetentionEnforcer` | One short-lived throttle key per namespace receiving worker heartbeats; the elected heartbeat prunes at most one expired run. | +| `stream_retention_inline` | `server:stream-retention-inline:` | `App\Support\WorkflowStreamRetention` | One short-lived throttle key per namespace receiving worker heartbeats; the elected heartbeat reclaims at most 10 expired streams and 500 of their items. | | `worker_compatibility_heartbeat` | `server:worker-compatibility-heartbeat:` | `App\Support\WorkerCompatibilityHeartbeatRecorder` | One expiring throttle key per recently heartbeating namespace/worker elects a single compatibility-fleet refresh across standalone HTTP processes during each TTL/3 write interval, then drains without an index. | | `readiness_probe` | `server:readiness:` | `App\Support\ServerReadiness` | One temporary probe key per readiness check; deleted immediately and also protected by a 10-second TTL. | diff --git a/docs/contracts/workflow-streams.md b/docs/contracts/workflow-streams.md index 486934d3a..75ccdf165 100644 --- a/docs/contracts/workflow-streams.md +++ b/docs/contracts/workflow-streams.md @@ -87,11 +87,16 @@ The contract names two distinct flows: - **Slow consumer.** Each append increments `pending_items`. Each successful read with `next_offset > last_offset` (consumer caught up) advisory-decrements pending back toward zero. When - `pending_items >= max_pending_items`, the next append returns - `429` with `reason=stream_full`. The default - `max_pending_items` per stream is 10000; producers may pass an - explicit `max_pending_items` per request to negotiate a lower or - higher cap on a per-stream basis. + `pending_items` plus the new items of an append would exceed + `max_pending_items`, the append returns `429` with + `reason=stream_full`. The whole batch is refused before any item + is written, so a plain retry cannot duplicate part of it. Items + whose `idempotency_key` is already in the stream (or earlier in the + same batch) take no capacity, and a batch made only of such items + still answers with their existing offsets. The default and the + maximum `max_pending_items` per stream are both 10000. A producer + may pass a lower `max_pending_items` per request; a value above + 10000 is refused with `422`. - **Fast producer / idle consumer.** A subscriber may pass `wait_seconds` (default 0, max 60) to long-poll. When no items are available within the window the request returns an empty @@ -103,6 +108,8 @@ Per-request limits: - `max_items` on subscribe: default 100, max 500. - `items` on append: max 500 per request. +- `max_pending_items` on append: 1 to 10000. +- `retention_seconds` on close: 1 to 2592000 (30 days). - `wait_seconds` on subscribe: max 60. ## Observability @@ -134,8 +141,44 @@ responses. - An `open` stream is retained until close. - A `closed` or `errored` stream is retained for `retention_seconds` after `closed_at`. Default is 3600 seconds. - The producer may override on close. The namespace retention - sweep removes expired closed streams. + The producer may override on close, up to 2592000 seconds (30 + days); a larger value is refused with `422`. +- Once `closed_at + retention_seconds` has passed, the stream reads + as not found from that instant: list omits it, and describe, + subscribe, append and close answer `404 stream_not_found`. +- The expired stream's name stays taken for the rest of its run: an + append never opens it again at offset 0. Its row, without items, + is removed when the run itself is pruned. +- A bounded retention pass removes the expired stream's items, + whatever the namespace's history retention mode. It runs from the + throttled inline retention pass on worker heartbeats, from + `history:prune`, and from a general operator retention pass + (`POST /api/system/retention/pass` without `run_ids`). + - Each pass reads a capped number of a namespace's due streams in + deadline order from an index on the stream's reclaim deadline, + so its reads stop at that cap. + - It deletes a capped number of items in small transactions, and + the next pass resumes where it stopped. + - Before deleting a batch it settles the batch's external + payloads as run retention does: an object no other retained row + references is deleted (honoring backup holds), and an object + another row still references is kept. + - If the namespace's external payload storage is unavailable, or a + deletion fails, the batch's items stay and a later pass retries + them. +- When a run is pruned, its streams and their items are removed with + it, after the same external payload handling. Pruning takes the + run's lock before anything else, so an append to the run that + arrives meanwhile waits, then answers `404 stream_not_found` and + writes nothing. +- Appends and cleanup decide each external payload object under that + object's lock. An admitted append claims its items' references in + its own transaction. Cleanup asks again, under the lock, whether + any retained row references the object before deleting it. If + cleanup deleted the object first, the append is refused and writes + nothing; over HTTP it answers `422 external_payload_unsupported`, + and the producer uploads the payload again. Otherwise the object + stays. ## SDK implementation notes diff --git a/docs/perf-runner.md b/docs/perf-runner.md index d81f9d0a7..34d5840fb 100644 --- a/docs/perf-runner.md +++ b/docs/perf-runner.md @@ -184,7 +184,7 @@ omitted or incomplete. The workflow file contains the canonical smoke and long-soak threshold maps, for example: ```bash -DW_PERF_MAX_FINAL_SERVER_CACHE_KEYS_BY_POLICY='{"workflow_task_poll_requests":0,"activity_task_poll_requests":0,"query_task_poll_requests":0,"sqlite_worker_poll_claim_gate":0,"long_poll_signals":0,"long_poll_wait_slots":0,"workflow_query_tasks":0,"task_queue_admission_locks":0,"task_queue_dispatch_counters":0,"workflow_task_expired_lease_recovery":0,"history_retention_inline":0,"readiness_probe":0}' +DW_PERF_MAX_FINAL_SERVER_CACHE_KEYS_BY_POLICY='{"workflow_task_poll_requests":0,"activity_task_poll_requests":0,"query_task_poll_requests":0,"sqlite_worker_poll_claim_gate":0,"long_poll_signals":0,"long_poll_wait_slots":0,"workflow_query_tasks":0,"task_queue_admission_locks":0,"task_queue_dispatch_counters":0,"workflow_task_expired_lease_recovery":0,"history_retention_inline":0,"stream_retention_inline":0,"readiness_probe":0}' ``` ## Safety Rules diff --git a/routes/console.php b/routes/console.php index 9888de778..79ba0e2d3 100644 --- a/routes/console.php +++ b/routes/console.php @@ -10,6 +10,7 @@ use App\Support\PayloadCodecDeploymentPreflight; use App\Support\RuntimeExternalPayloadCleanup; use App\Support\StoragePressure; +use App\Support\WorkflowStreamRetention; use Illuminate\Database\Migrations\Migrator; use Illuminate\Support\Facades\Artisan; use Workflow\V2\Enums\RunStatus; @@ -342,6 +343,31 @@ return 0; } + // Closed workflow streams past their own retention, in any retention + // mode (Server #774): at most $limit streams and $limit * 100 items per + // namespace per pass. A namespace whose payload cleanup is blocked or + // fails keeps its items for the next pass. + foreach ($namespaces as $ns) { + try { + $streams = WorkflowStreamRetention::pruneExpired($ns->name, $limit, $limit * 100); + } catch (Throwable $e) { + $this->components->error(sprintf('Namespace [%s]: stream retention failed: %s', $ns->name, $e->getMessage())); + + continue; + } + + if ($streams['streams_reclaimed'] > 0 || $streams['items_removed'] > 0 || $streams['blocked_reason'] !== null) { + $this->components->info(sprintf( + 'Namespace [%s]: %d expired stream(s) reclaimed, %d stream item(s) deleted%s%s.', + $ns->name, + $streams['streams_reclaimed'], + $streams['items_removed'], + $streams['budget_exhausted'] ? '; more remain for the next pass' : '', + $streams['blocked_reason'] !== null ? '; blocked: '.$streams['blocked_reason'] : '', + )); + } + } + $totalPruned = 0; $totalSkipped = 0; $totalFailed = 0; diff --git a/rust/src/runtime/completion_rules.rs b/rust/src/runtime/completion_rules.rs index 43907f873..6c8382191 100644 --- a/rust/src/runtime/completion_rules.rs +++ b/rust/src/runtime/completion_rules.rs @@ -212,9 +212,11 @@ const COMMAND: &[(&str, &[Check])] = &[ ("workflow_stream.items.*.idempotency_key", &[NUL, STR, max(191)]), ("workflow_stream.items.*.item_type", &[NUL, STR, max(64)]), ("workflow_stream.items.*.content_type", &[NUL, STR, max(191)]), - ("workflow_stream.max_pending_items", &[NUL, INT, min(1)]), + // `WorkflowStreamService::MAX_PENDING_ITEMS_MAX` and + // `RETENTION_SECONDS_MAX` (Server #774). + ("workflow_stream.max_pending_items", &[NUL, INT, min(1), max(10_000)]), ("workflow_stream.error_reason", &[NUL, STR, max(191)]), - ("workflow_stream.retention_seconds", &[NUL, INT, min(1)]), + ("workflow_stream.retention_seconds", &[NUL, INT, min(1), max(2_592_000)]), ]; /// `WORKFLOW_TASK_COMMAND_VALIDATION_BATCH_SIZE`. @@ -764,6 +766,15 @@ mod tests { assert_eq!(refusal(base(json!(["x"])))["message"], "The commands.0.type field is required."); assert_eq!(refusal(json!({"workflow_task_attempt":"x","commands":[{"type":"a"}]}))["message"], "The lease owner field is required. (and 1 more error)"); + // Server #774: a workflow stream's producer controls are bounded. + let stream = |field: &str, value: i64| base(json!([{"type":"record_side_effect","workflow_stream":{"operation":"append", + "stream_name":"tokens","command_identity":"c","command_ordinal":0,field:value}}])); + assert_eq!(refusal(stream("max_pending_items", 10_001))["errors"]["commands.0.workflow_stream.max_pending_items"], + json!(["The commands.0.workflow_stream.max_pending_items field must not be greater than 10000."])); + assert_eq!(refusal(stream("retention_seconds", 2_592_001))["errors"]["commands.0.workflow_stream.retention_seconds"], + json!(["The commands.0.workflow_stream.retention_seconds field must not be greater than 2592000."])); + assert!(checked(stream("max_pending_items", 10_000)).is_ok()); + assert!(checked(stream("retention_seconds", 2_592_000)).is_ok()); } #[test] diff --git a/scripts/perf/server_soak.py b/scripts/perf/server_soak.py index bc0de0cad..6e581bc2e 100755 --- a/scripts/perf/server_soak.py +++ b/scripts/perf/server_soak.py @@ -44,6 +44,7 @@ "runtime_external_payload_quota_rejections": "*server:external-payload-quota:*", "workflow_task_expired_lease_recovery": "*server:workflow-task-expired-lease-recovery:*", "history_retention_inline": "*server:history-retention-inline:*", + "stream_retention_inline": "*server:stream-retention-inline:*", "worker_compatibility_heartbeat": "*server:worker-compatibility-heartbeat:*", "readiness_probe": "*server:readiness:*", } diff --git a/tests/Feature/Concerns/ForkedDatabaseActors.php b/tests/Feature/Concerns/ForkedDatabaseActors.php new file mode 100644 index 000000000..2ed712dd2 --- /dev/null +++ b/tests/Feature/Concerns/ForkedDatabaseActors.php @@ -0,0 +1,408 @@ +, tables: list, seeds: array>>}> + */ + private static array $actorDatabases = []; + + /** @var array */ + private array $actors = []; + + /** @return array */ + public static function drivers(): array + { + $drivers = ['mysql', 'mariadb', 'pgsql', 'sqlite']; + + return array_combine($drivers, array_map(static fn (string $driver): array => [$driver], $drivers)); + } + + private function initializeActorDatabase(string $driver): void + { + if (! function_exists('pcntl_fork') || ! function_exists('posix_kill') || ! function_exists('stream_socket_pair')) { + $this->markTestSkipped('Process control and local sockets are required.'); + } + + if (! isset(self::$actorDatabases[$driver])) { + self::$actorDatabases[$driver] = $this->migrateActorDatabase($driver); + } else { + config(self::$actorDatabases[$driver]['config']); + DB::purge($driver); + $this->resetActorDatabase($driver, self::$actorDatabases[$driver]); + } + $this->createNamespace('default'); + } + + /** + * Creates and migrates the process's database for $driver, records its + * migrated rows, and drops it when the test process exits. + * + * @return array{config: array, tables: list, seeds: array>>} + */ + private function migrateActorDatabase(string $driver): array + { + $owner = getmypid(); + + if ($driver === 'sqlite') { + $directory = is_dir('/dev/shm') && is_writable('/dev/shm') ? '/dev/shm' : sys_get_temp_dir(); + $path = tempnam($directory, 'dw-stream-race-'); + $this->assertIsString($path); + // The server's own SQLite settings: WAL and a five-second busy + // timeout. PHP 8.3 begins transactions DEFERRED whatever the + // configured mode, so the stream lock must hold without it. + $config = ['database.default' => 'sqlite', 'database.connections.sqlite.database' => $path, + 'database.connections.sqlite.busy_timeout' => 5000, 'database.connections.sqlite.journal_mode' => 'WAL']; + register_shutdown_function(static function () use ($owner, $path): void { + if (getmypid() === $owner) { + foreach ([$path, $path.'-wal', $path.'-shm'] as $file) { + if (is_file($file)) { + @unlink($file); + } + } + } + }); + } else { + $prefix = [ + 'mysql' => 'DW_TEST_COMPLETION_MYSQL_', + 'mariadb' => 'DW_TEST_COMPLETION_MARIADB_', + 'pgsql' => 'DW_TEST_COMPLETION_PGSQL_', + ][$driver]; + $host = getenv($prefix.'HOST'); + if (! $host) { + $this->markTestSkipped('Set '.$prefix.'HOST to a disposable database test server.'); + } + $port = getenv($prefix.'PORT') ?: ($driver === 'pgsql' ? '5432' : '3306'); + $user = getenv($prefix.'USER') ?: ($driver === 'pgsql' ? 'postgres' : 'root'); + $password = getenv($prefix.'PASSWORD') ?: ''; + $admin = [($driver === 'pgsql' ? 'pgsql' : 'mysql').':host='.$host.';port='.$port.($driver === 'pgsql' ? ';dbname=postgres' : ''), $user, $password]; + $name = 'dw_stream_race_'.bin2hex(random_bytes(6)); + (new PDO(...$admin))->exec('CREATE DATABASE '.$name); + $config = ['database.default' => $driver, 'database.connections.'.$driver.'.host' => $host, + 'database.connections.'.$driver.'.port' => $port, 'database.connections.'.$driver.'.database' => $name, + 'database.connections.'.$driver.'.username' => $user, 'database.connections.'.$driver.'.password' => $password, + 'database.connections.'.$driver.'.url' => null]; + register_shutdown_function(static function () use ($owner, $admin, $name, $driver): void { + if (getmypid() === $owner) { + try { + (new PDO(...$admin))->exec('DROP DATABASE '.$name.($driver === 'pgsql' ? ' WITH (FORCE)' : '')); + } catch (Throwable) { + // A disposable test server; nothing else to clean. + } + } + }); + } + + config($config); + DB::purge($driver); + $this->artisan('migrate:fresh', ['--force' => true])->assertExitCode(0); + + $schema = match ($driver) { + 'sqlite' => 'main', + 'pgsql' => 'public', + default => DB::connection()->getDatabaseName(), + }; + $tables = array_values(array_diff(Schema::getTableListing($schema, schemaQualified: false), ['migrations'])); + $seeds = []; + foreach ($tables as $table) { + $rows = DB::table($table)->get()->map(static fn (object $row): array => (array) $row)->all(); + if ($rows !== []) { + $seeds[$table] = $rows; + } + } + + return ['config' => $config, 'tables' => $tables, 'seeds' => $seeds]; + } + + /** + * Deletes every row a test wrote and restores the rows the migrations + * wrote, with foreign keys unchecked while the tables are inconsistent. + * + * @param array{tables: list, seeds: array>>} $database + */ + private function resetActorDatabase(string $driver, array $database): void + { + $connection = DB::connection(); + $grammar = $connection->getQueryGrammar(); + $wrapped = array_map(static fn (string $table): string => $grammar->wrapTable($table), $database['tables']); + if ($driver === 'pgsql') { + $connection->unprepared('TRUNCATE '.implode(', ', $wrapped).' CASCADE'); + } else { + $connection->unprepared($driver === 'sqlite' ? 'PRAGMA foreign_keys = OFF' : 'SET FOREIGN_KEY_CHECKS = 0'); + foreach ($wrapped as $table) { + $connection->unprepared('DELETE FROM '.$table); + } + } + foreach ($database['seeds'] as $table => $rows) { + foreach (array_chunk($rows, 200) as $chunk) { + DB::table($table)->insert($chunk); + } + } + if ($driver !== 'pgsql') { + $connection->unprepared($driver === 'sqlite' ? 'PRAGMA foreign_keys = ON' : 'SET FOREIGN_KEY_CHECKS = 1'); + } + } + + protected function tearDown(): void + { + $this->tearDownActors(); + parent::tearDown(); + } + + /** Stops any actor still running. */ + private function tearDownActors(): void + { + foreach ($this->actors as $actor) { + posix_kill($actor['pid'], SIGKILL); + pcntl_waitpid($actor['pid'], $status); + if (is_resource($actor['socket'])) { + fclose($actor['socket']); + } + } + $this->actors = []; + DB::disconnect(); + } + + private function createRun(string $workflowId, RunStatus $status = RunStatus::Running): string + { + $instance = WorkflowInstance::query()->create([ + 'id' => $workflowId, + 'workflow_class' => 'Tests\\Fixtures\\StreamingWorkflow', + 'workflow_type' => 'tests.streaming-workflow', + 'namespace' => 'default', + 'run_count' => 1, + 'started_at' => now(), + ]); + $run = WorkflowRun::query()->create([ + 'id' => (string) Str::ulid(), + 'workflow_instance_id' => $instance->id, + 'run_number' => 1, + 'workflow_class' => 'Tests\\Fixtures\\StreamingWorkflow', + 'workflow_type' => 'tests.streaming-workflow', + 'namespace' => 'default', + 'status' => $status->value, + 'started_at' => now(), + 'closed_at' => $status->isTerminal() ? now() : null, + 'last_progress_at' => now(), + ]); + $instance->forceFill(['current_run_id' => $run->id])->save(); + + return $run->id; + } + + /** + * Starts $work in a child process on its own database connection and + * returns once the child is connected. The child stops at each pause in + * turn, until resumed: right after the first statement, or top-level + * commit, for which the pause returns true. A pause is asked the + * statement's lowercase SQL, or 'commit', with the driver and the + * transaction level the statement ran at (0 after a top-level commit). + * + * @param Closure(string $sql, string $driver, int $level): bool|list $pauses + * @return array{pid: int, socket: resource, connection_id: int} + */ + private function startActor(callable $work, Closure|array $pauses = []): array + { + $pauses = $pauses instanceof Closure ? [$pauses] : $pauses; + // The child must not share this process's database connection. + DB::purge(); + $sockets = stream_socket_pair(STREAM_PF_UNIX, STREAM_SOCK_STREAM, STREAM_IPPROTO_IP); + $this->assertIsArray($sockets); + $pid = pcntl_fork(); + $this->assertNotSame(-1, $pid); + if ($pid === 0) { + fclose($sockets[0]); + stream_set_timeout($sockets[1], 15); + $ok = false; + try { + DB::reconnect(); + $driver = DB::connection()->getDriverName(); + $connectionId = match ($driver) { + 'mysql', 'mariadb' => (int) DB::selectOne('SELECT CONNECTION_ID() AS id')->id, + 'pgsql' => (int) DB::selectOne('SELECT pg_backend_pid() AS id')->id, + default => getmypid(), + }; + fwrite($sockets[1], json_encode(['connection_id' => $connectionId]).PHP_EOL); + if (fgets($sockets[1]) !== "go\n") { + throw new RuntimeException('Actor was not released.'); + } + if ($pauses !== []) { + $check = static function (string $sql, int $level) use (&$pauses, $sockets, $driver): void { + if ($pauses !== [] && $pauses[0]($sql, $driver, $level)) { + array_shift($pauses); + fwrite($sockets[1], json_encode(['paused' => true]).PHP_EOL); + if (fgets($sockets[1]) !== "continue\n") { + throw new RuntimeException('Actor was not resumed.'); + } + } + }; + DB::listen(static fn (QueryExecuted $query) => $check(strtolower(ltrim($query->sql)), $query->connection->transactionLevel())); + Event::listen(TransactionCommitted::class, static fn (TransactionCommitted $event) => $check('commit', $event->connection->transactionLevel())); + } + fwrite($sockets[1], json_encode(['result' => $work()], JSON_THROW_ON_ERROR).PHP_EOL); + $ok = true; + } catch (Throwable $error) { + fwrite($sockets[1], json_encode(['error' => $error::class.': '.$error->getMessage()]).PHP_EOL); + } finally { + DB::disconnect(); + fclose($sockets[1]); + } + exit($ok ? 0 : 1); + } + fclose($sockets[1]); + stream_set_timeout($sockets[0], 15); + $actor = ['pid' => $pid, 'socket' => $sockets[0]]; + $this->actors[$pid] = $actor; + $actor['connection_id'] = (int) $this->readMessage($actor)['connection_id']; + + return $actor; + } + + /** Releases an actor and waits until it reaches its first pause. */ + private function runUntilPaused(array $actor): void + { + fwrite($actor['socket'], "go\n"); + $this->assertTrue($this->readMessage($actor)['paused'] ?? false, 'Actor did not reach its pause.'); + } + + /** Resumes a paused actor and waits until it reaches its next pause. */ + private function resumeUntilPaused(array $actor): void + { + fwrite($actor['socket'], "continue\n"); + $this->assertTrue($this->readMessage($actor)['paused'] ?? false, 'Actor did not reach its next pause.'); + } + + private function release(array $actor): void + { + fwrite($actor['socket'], "go\n"); + } + + private function resume(array $actor): void + { + fwrite($actor['socket'], "continue\n"); + } + + /** + * The actor is waiting on another actor's row lock in $table. MySQL + * reports the lock wait and its table; PostgreSQL reports the actor + * waiting on a lock in its locking read of $table. MariaDB has no + * lock-wait table that shows a wait taken while the optimizer reads a + * constant row, so there the actor must be executing its locking read of + * $table and still be there a moment later. SQLite has no row locks: the + * actor must not finish while the other holds the database write lock. + */ + private function awaitWait(array $actor, string $table): void + { + $driver = DB::connection()->getDriverName(); + + if ($driver === 'sqlite') { + $this->assertStillRunning($actor); + + return; + } + + [$query, $bindings] = match ($driver) { + 'mysql' => ['SELECT COUNT(*) AS waiting FROM performance_schema.data_lock_waits w ' + .'JOIN performance_schema.threads t ON t.THREAD_ID = w.REQUESTING_THREAD_ID ' + .'JOIN performance_schema.data_locks l ON l.ENGINE_LOCK_ID = w.REQUESTING_ENGINE_LOCK_ID ' + .'WHERE t.PROCESSLIST_ID = ? AND l.OBJECT_NAME = ?', [$actor['connection_id'], $table]], + 'mariadb' => ["SELECT COUNT(*) AS waiting FROM information_schema.PROCESSLIST WHERE ID = ? AND COMMAND <> 'Sleep' AND LOWER(INFO) LIKE ?", + [$actor['connection_id'], '%'.$table.'%for update%']], + default => ["SELECT COUNT(*) AS waiting FROM pg_stat_activity WHERE pid = ? AND wait_event_type = 'Lock' AND LOWER(query) LIKE ?", + [$actor['connection_id'], '%'.$table.'%for update%']], + }; + $until = microtime(true) + 5; + do { + if ((int) DB::selectOne($query, $bindings)->waiting > 0) { + if ($driver === 'mariadb') { + $this->assertStillRunning($actor); + $this->assertSame(1, (int) DB::selectOne($query, $bindings)->waiting, + 'The actor left its locking read of '.$table.' while the other actor held the lock.'); + } + $this->addToAssertionCount(1); + + return; + } + usleep(10_000); + } while (microtime(true) < $until); + $this->fail('The actor did not wait on the other actor\'s lock on '.$table.', connection '.$actor['connection_id'] + .': '.json_encode($this->waitDiagnostics($driver, $actor['connection_id']))); + } + + /** What the actor's connection is doing, for a failure message. */ + private function waitDiagnostics(string $driver, int $connectionId): array + { + return match ($driver) { + 'mysql' => DB::select('SELECT l.OBJECT_NAME, l.INDEX_NAME, l.LOCK_TYPE, l.LOCK_MODE, l.LOCK_DATA, t.PROCESSLIST_INFO ' + .'FROM performance_schema.data_lock_waits w ' + .'JOIN performance_schema.threads t ON t.THREAD_ID = w.REQUESTING_THREAD_ID ' + .'JOIN performance_schema.data_locks l ON l.ENGINE_LOCK_ID = w.REQUESTING_ENGINE_LOCK_ID ' + .'WHERE t.PROCESSLIST_ID = ?', [$connectionId]), + 'mariadb' => DB::select('SELECT COMMAND, STATE, INFO FROM information_schema.PROCESSLIST WHERE ID = ?', [$connectionId]), + default => DB::select('SELECT state, wait_event_type, wait_event, query FROM pg_stat_activity WHERE pid = ?', [$connectionId]), + }; + } + + private function assertStillRunning(array $actor): void + { + $read = [$actor['socket']]; + $write = $except = null; + $this->assertSame(0, stream_select($read, $write, $except, 0, 300_000), + 'The actor finished while the other actor held its locks.'); + } + + private function readMessage(array $actor): array + { + $line = fgets($actor['socket']); + $this->assertIsString($line, 'Actor did not respond.'); + $message = json_decode($line, true, flags: JSON_THROW_ON_ERROR); + $this->assertArrayNotHasKey('error', $message, $message['error'] ?? ''); + + return $message; + } + + /** @return mixed The actor's result. */ + private function finishActor(array $actor): mixed + { + $message = $this->readMessage($actor); + pcntl_waitpid($actor['pid'], $status); + unset($this->actors[$actor['pid']]); + $this->assertTrue(pcntl_wifexited($status)); + $this->assertSame(0, pcntl_wexitstatus($status)); + fclose($actor['socket']); + + return $message['result']; + } +} diff --git a/tests/Feature/HistoryRetentionTest.php b/tests/Feature/HistoryRetentionTest.php index 4401ac39e..7d7f79f36 100644 --- a/tests/Feature/HistoryRetentionTest.php +++ b/tests/Feature/HistoryRetentionTest.php @@ -5,13 +5,19 @@ namespace Tests\Feature; use App\Models\RuntimeExternalPayload; +use App\Models\WorkflowDurableStream; +use App\Models\WorkflowDurableStreamItem; use App\Models\WorkflowInboundStream; use App\Models\WorkflowInboundStreamItem; use App\Models\WorkflowNamespace; use App\Support\ExternalPayloadRetentionCleanup; use App\Support\MessageStreamService; +use App\Support\NamespaceExternalPayloadStorage; use App\Support\NamespaceWorkflowScope; use App\Support\RuntimeExternalPayloadRegistry; +use App\Support\RuntimeExternalPayloadStorageDriver; +use App\Support\RuntimeTrackedExternalPayloadStorage; +use App\Support\WorkflowStreamRetention; use Illuminate\Foundation\Testing\RefreshDatabase; use Illuminate\Support\Facades\Cache; use Illuminate\Support\Facades\DB; @@ -1284,6 +1290,320 @@ private function createNamespaceWithRetention(string $name, int $retentionDays): ); } + public function test_retention_pass_deletes_durable_streams_with_their_stream_only_payloads(): void + { + Queue::fake(); + $storageDirectory = $this->localExternalPayloadStorage('retention-durable-streams'); + $expiredRunId = $this->createExpiredClosedRun('default', 'wf-prune-durable-stream'); + $retainedRunId = $this->createExpiredClosedRun('default', 'wf-retain-durable-stream', daysAgo: 1); + $path = $storageDirectory.'/payloads/stream-item.bin'; + File::ensureDirectoryExists(dirname($path)); + file_put_contents($path, 'stream item bytes'); + + $this->addDurableStream($expiredRunId, 'tokens', ['file://'.$path, null]); + $kept = $this->addDurableStream($retainedRunId, 'kept', [null]); + + $this->withHeaders($this->apiHeaders()) + ->postJson('/api/system/retention/pass') + ->assertOk() + ->assertJsonPath('processed', 1) + ->assertJsonPath('pruned', 1) + ->assertJsonPath('results.0.external_payloads_deleted', 1) + ->assertJsonPath('results.0.deleted.durable_streams_deleted', 1) + ->assertJsonPath('results.0.deleted.durable_stream_items_deleted', 2); + + $this->assertFileDoesNotExist($path); + $this->assertSame(0, WorkflowDurableStream::query()->where('workflow_run_id', $expiredRunId)->count()); + $this->assertSame(0, WorkflowDurableStreamItem::query()->where('workflow_run_id', $expiredRunId)->count()); + $this->assertNotNull($kept->fresh()); + $this->assertSame(1, WorkflowDurableStreamItem::query()->where('workflow_run_id', $retainedRunId)->count()); + + File::deleteDirectory($storageDirectory); + } + + public function test_retention_pass_keeps_a_stream_payload_shared_with_a_retained_run(): void + { + Queue::fake(); + $storageDirectory = $this->localExternalPayloadStorage('retention-shared-durable-streams'); + $expiredRunId = $this->createExpiredClosedRun('default', 'wf-prune-shared-durable-stream'); + $retainedRunId = $this->createExpiredClosedRun('default', 'wf-retain-shared-durable-stream', daysAgo: 1); + $path = $storageDirectory.'/payloads/shared-stream-item.bin'; + File::ensureDirectoryExists(dirname($path)); + file_put_contents($path, 'shared stream item bytes'); + + $this->addDurableStream($expiredRunId, 'tokens', ['file://'.$path]); + $this->addDurableStream($retainedRunId, 'tokens', ['file://'.$path]); + + $this->withHeaders($this->apiHeaders()) + ->postJson('/api/system/retention/pass') + ->assertOk() + ->assertJsonPath('pruned', 1) + ->assertJsonPath('results.0.external_payloads_deleted', 0) + ->assertJsonPath('results.0.deleted.durable_streams_deleted', 1); + + $this->assertFileExists($path); + $this->assertSame(0, WorkflowDurableStream::query()->where('workflow_run_id', $expiredRunId)->count()); + $this->assertSame(1, WorkflowDurableStreamItem::query()->where('workflow_run_id', $retainedRunId)->count()); + + File::deleteDirectory($storageDirectory); + } + + public function test_retention_pass_keeps_durable_streams_when_their_payload_cleanup_is_blocked(): void + { + Queue::fake(); + $this->createNamespace('default'); + WorkflowNamespace::where('name', 'default')->update([ + 'external_payload_storage' => [ + 'driver' => 's3', + 'enabled' => true, + 'config' => ['bucket' => 'dw-payloads', 'prefix' => 'retention/'], + ], + ]); + $runId = $this->createExpiredClosedRun('default', 'wf-prune-durable-stream-blocked'); + $stream = $this->addDurableStream($runId, 'tokens', ['s3://dw-payloads/retention/stream-item.bin']); + + $this->withHeaders($this->apiHeaders()) + ->postJson('/api/system/retention/pass') + ->assertOk() + ->assertJsonPath('pruned', 0) + ->assertJsonPath('skipped', 1) + ->assertJsonPath('results.0.reason', 'external_payload_storage_driver_unavailable'); + + $this->assertNotNull(WorkflowRunSummary::find($runId)); + $this->assertNotNull($stream->fresh()); + $this->assertSame(1, WorkflowDurableStreamItem::query()->where('workflow_run_id', $runId)->count()); + } + + public function test_retention_pass_keeps_durable_streams_when_payload_deletion_fails_and_retries(): void + { + Queue::fake(); + $this->createNamespace('default'); + $runId = $this->createExpiredClosedRun('default', 'wf-prune-durable-stream-storage-failure'); + $stream = $this->addDurableStream($runId, 'tokens', ['file:///unreachable/stream-item.bin']); + $failing = $this->failingExternalPayloadStorage(); + + $this->withHeaders($this->apiHeaders()) + ->postJson('/api/system/retention/pass') + ->assertStatus(207) + ->assertJsonPath('pruned', 0) + ->assertJsonPath('failed', 1) + ->assertJsonPath('results.0.outcome', 'error'); + + $this->assertNotNull(WorkflowRunSummary::find($runId)); + $this->assertNotNull($stream->fresh()); + $this->assertSame(1, WorkflowDurableStreamItem::query()->where('workflow_run_id', $runId)->count()); + + // The next pass, with storage back, settles the reference and prunes. + $failing->fail = false; + $this->withHeaders($this->apiHeaders()) + ->postJson('/api/system/retention/pass') + ->assertOk() + ->assertJsonPath('pruned', 1) + ->assertJsonPath('results.0.deleted.durable_streams_deleted', 1); + + $this->assertSame(['file:///unreachable/stream-item.bin'], $failing->deleted); + $this->assertNull($stream->fresh()); + } + + public function test_stream_retention_reclaims_a_stream_only_payload_with_its_items(): void + { + Queue::fake(); + $storageDirectory = $this->localExternalPayloadStorage('stream-retention-payloads'); + $runId = $this->createExpiredClosedRun('default', 'wf-stream-retention-payload', daysAgo: 1); + $path = $storageDirectory.'/payloads/expired-stream-item.bin'; + File::ensureDirectoryExists(dirname($path)); + file_put_contents($path, 'expired stream item bytes'); + $stream = $this->addDurableStream($runId, 'tokens', ['file://'.$path, null], due: true); + + $this->assertSame( + ['streams_reclaimed' => 1, 'items_removed' => 2, 'budget_exhausted' => false, 'blocked_reason' => null], + WorkflowStreamRetention::pruneExpired('default', 10, 1000), + ); + + $this->assertFileDoesNotExist($path); + $this->assertSame(0, WorkflowDurableStreamItem::query()->where('stream_id', $stream->id)->count()); + $this->assertNull($stream->fresh()->reclaim_at); + $this->assertNotNull(WorkflowRunSummary::find($runId)); + + File::deleteDirectory($storageDirectory); + } + + public function test_stream_retention_keeps_a_payload_another_retained_row_references(): void + { + Queue::fake(); + $storageDirectory = $this->localExternalPayloadStorage('stream-retention-shared-payloads'); + $runId = $this->createExpiredClosedRun('default', 'wf-stream-retention-shared', daysAgo: 1); + $path = $storageDirectory.'/payloads/shared-stream-item.bin'; + File::ensureDirectoryExists(dirname($path)); + file_put_contents($path, 'shared stream item bytes'); + $this->addDurableStream($runId, 'expired', ['file://'.$path], due: true); + // A stream that is not due references the same object. + $kept = $this->addDurableStream($runId, 'kept', ['file://'.$path]); + + $this->assertSame( + ['streams_reclaimed' => 1, 'items_removed' => 1, 'budget_exhausted' => false, 'blocked_reason' => null], + WorkflowStreamRetention::pruneExpired('default', 10, 1000), + ); + + $this->assertFileExists($path); + $this->assertSame(1, WorkflowDurableStreamItem::query()->where('stream_id', $kept->id)->count()); + + File::deleteDirectory($storageDirectory); + } + + public function test_stream_retention_keeps_items_when_payload_cleanup_is_blocked(): void + { + Queue::fake(); + $this->createNamespace('default'); + WorkflowNamespace::where('name', 'default')->update([ + 'external_payload_storage' => [ + 'driver' => 's3', + 'enabled' => true, + 'config' => ['bucket' => 'dw-payloads', 'prefix' => 'retention/'], + ], + ]); + $runId = $this->createExpiredClosedRun('default', 'wf-stream-retention-blocked', daysAgo: 1); + $stream = $this->addDurableStream($runId, 'tokens', ['s3://dw-payloads/retention/stream-item.bin', null], due: true); + + $this->assertSame( + ['streams_reclaimed' => 0, 'items_removed' => 0, 'budget_exhausted' => false, 'blocked_reason' => 'external_payload_storage_driver_unavailable'], + WorkflowStreamRetention::pruneExpired('default', 10, 1000), + ); + + $this->assertSame(2, WorkflowDurableStreamItem::query()->where('stream_id', $stream->id)->count()); + $this->assertNotNull($stream->fresh()->reclaim_at); + } + + public function test_stream_retention_keeps_items_when_payload_deletion_fails_and_retries(): void + { + Queue::fake(); + $this->createNamespace('default'); + $runId = $this->createExpiredClosedRun('default', 'wf-stream-retention-failure', daysAgo: 1); + $stream = $this->addDurableStream($runId, 'tokens', ['file:///unreachable/expired-stream-item.bin'], due: true); + $failing = $this->failingExternalPayloadStorage(); + + try { + WorkflowStreamRetention::pruneExpired('default', 10, 1000); + $this->fail('Expected the payload deletion failure to stop the pass.'); + } catch (\RuntimeException $exception) { + $this->assertStringContainsString('external payload', strtolower($exception->getMessage())); + } + + $this->assertSame(1, WorkflowDurableStreamItem::query()->where('stream_id', $stream->id)->count()); + $this->assertNotNull($stream->fresh()->reclaim_at); + + // The next pass, with storage back, settles the reference and reclaims. + $failing->fail = false; + $this->assertSame( + ['streams_reclaimed' => 1, 'items_removed' => 1, 'budget_exhausted' => false, 'blocked_reason' => null], + WorkflowStreamRetention::pruneExpired('default', 10, 1000), + ); + $this->assertSame(['file:///unreachable/expired-stream-item.bin'], $failing->deleted); + } + + /** + * Binds a namespace storage whose provider fails deletes until `fail` is + * cleared, and records the deletes it then performs. + */ + private function failingExternalPayloadStorage(): object + { + $failing = new class implements RuntimeExternalPayloadStorageDriver + { + public bool $fail = true; + + /** @var list */ + public array $deleted = []; + + public function put(string $data, string $sha256, string $codec): string + { + throw new \LogicException('not used'); + } + + public function get(string $uri): string + { + throw new \LogicException('not used'); + } + + public function delete(string $uri): void + { + if ($this->fail) { + throw new \RuntimeException('object storage is down'); + } + $this->deleted[] = $uri; + } + + public function uriFor(string $sha256, string $codec): string + { + throw new \LogicException('not used'); + } + }; + $this->partialMock(NamespaceExternalPayloadStorage::class, function ($mock) use ($failing): void { + $mock->shouldReceive('driverFor')->andReturn(new RuntimeTrackedExternalPayloadStorage('default', $failing)); + }); + + return $failing; + } + + private function localExternalPayloadStorage(string $name): string + { + $storageDirectory = storage_path('framework/testing/'.$name); + File::deleteDirectory($storageDirectory); + $this->createNamespace('default'); + WorkflowNamespace::where('name', 'default')->update([ + 'external_payload_storage' => [ + 'driver' => 'local', + 'enabled' => true, + 'config' => ['uri' => 'file://'.$storageDirectory], + ], + ]); + + return $storageDirectory; + } + + /** + * A closed durable stream on $runId with one item per entry of + * $references: a provider reference, or null for an inline payload. + * + * @param list $references + */ + private function addDurableStream(string $runId, string $streamName, array $references, bool $due = false): WorkflowDurableStream + { + $run = WorkflowRun::findOrFail($runId); + + return DB::transaction(function () use ($run, $streamName, $references, $due): WorkflowDurableStream { + $stream = WorkflowDurableStream::query()->create([ + 'namespace' => $run->namespace, + 'workflow_instance_id' => $run->workflow_instance_id, + 'workflow_run_id' => $run->id, + 'stream_name' => $streamName, + 'status' => WorkflowDurableStream::STATUS_CLOSED, + 'last_offset' => count($references) - 1, + 'total_items' => count($references), + 'opened_at' => now()->subDays(61), + 'closed_at' => now()->subDays(61), + 'retention_seconds' => 3600, + // Not due unless asked, so run retention tests see every item. + 'reclaim_at' => $due ? now()->subDays(61)->addHour() : null, + ]); + + foreach ($references as $offset => $reference) { + WorkflowDurableStreamItem::query()->create([ + 'stream_id' => $stream->id, + 'namespace' => $run->namespace, + 'workflow_run_id' => $run->id, + 'stream_name' => $streamName, + 'offset' => $offset, + 'payload' => $reference === null ? ['inline' => $offset] : null, + 'payload_reference' => $reference, + 'payload_codec' => $reference === null ? null : 'avro', + 'emitted_at' => now()->subDays(61), + ]); + } + + return $stream; + }); + } + private function createExpiredClosedRun(string $namespace, string $workflowId, int $daysAgo = 60): string { $workflow = WorkflowStub::make(ExternalGreetingWorkflow::class, $workflowId); diff --git a/tests/Feature/RuntimeExternalPayloadTransportTest.php b/tests/Feature/RuntimeExternalPayloadTransportTest.php index baacb66b3..eb83a8928 100644 --- a/tests/Feature/RuntimeExternalPayloadTransportTest.php +++ b/tests/Feature/RuntimeExternalPayloadTransportTest.php @@ -1177,6 +1177,64 @@ public function test_worker_memo_command_resolves_opaque_reference_before_record ->count()); } + public function test_stream_reference_is_claimed_only_by_an_admitted_append(): void + { + Queue::fake(); + $start = $this->withHeaders($this->controlHeaders()) + ->postJson('/api/workflows', [ + 'workflow_id' => 'runtime-reference-stream-claim', + 'workflow_type' => 'tests.external-greeting-workflow', + 'input' => ['stream'], + ]) + ->assertCreated(); + $route = fn (string $stream): string => sprintf( + '/api/workflows/runtime-reference-stream-claim/runs/%s/streams/%s/items', + $start->json('run_id'), + $stream, + ); + $reference = $this->upload(Serializer::serializeWithCodec('avro', ['claimed item']))->json('reference'); + $item = ['payload_reference' => $reference, 'payload_codec' => 'avro']; + $unclaimed = function () use ($reference): void { + $row = RuntimeExternalPayload::query()->whereKey($reference['reference_id'])->firstOrFail(); + $this->assertNull($row->retained_at); + $this->assertNotNull($row->expires_at); + }; + $unclaimed(); + + // A batch refused at capacity writes nothing and claims nothing. + $this->withHeaders($this->controlHeaders()) + ->postJson($route('full'), ['items' => [['payload' => 'first']], 'max_pending_items' => 1]) + ->assertOk(); + $this->withHeaders($this->controlHeaders()) + ->postJson($route('full'), ['items' => [$item], 'max_pending_items' => 1]) + ->assertStatus(429) + ->assertJsonPath('reason', 'stream_full'); + $unclaimed(); + + // So does an append refused by a closed stream. + $this->withHeaders($this->controlHeaders()) + ->postJson($route('closed'), ['items' => [['payload' => 'first']]]) + ->assertOk(); + $this->withHeaders($this->controlHeaders()) + ->postJson(sprintf('/api/workflows/runtime-reference-stream-claim/runs/%s/streams/closed/close', $start->json('run_id')), []) + ->assertOk(); + $this->withHeaders($this->controlHeaders()) + ->postJson($route('closed'), ['items' => [$item]]) + ->assertStatus(409) + ->assertJsonPath('reason', 'stream_closed'); + $unclaimed(); + $this->assertDatabaseCount('workflow_durable_stream_items', 2); + + // An admitted append claims it with the item it writes. + $this->withHeaders($this->controlHeaders()) + ->postJson($route('admitted'), ['items' => [$item]]) + ->assertOk(); + $row = RuntimeExternalPayload::query()->whereKey($reference['reference_id'])->firstOrFail(); + $this->assertNotNull($row->retained_at); + $this->assertNull($row->expires_at); + $this->assertDatabaseCount('workflow_durable_stream_items', 3); + } + public function test_stream_reference_is_verified_before_commit_and_returns_opaque(): void { Queue::fake(); diff --git a/tests/Feature/WorkflowStreamCapacityRaceTest.php b/tests/Feature/WorkflowStreamCapacityRaceTest.php new file mode 100644 index 000000000..08192dee3 --- /dev/null +++ b/tests/Feature/WorkflowStreamCapacityRaceTest.php @@ -0,0 +1,231 @@ + ['mysql'], 'mariadb' => ['mariadb'], 'pgsql' => ['pgsql']]; + } + + #[DataProvider('drivers')] + public function test_concurrent_first_appends_open_one_stream_and_never_exceed_the_cap(string $driver): void + { + $this->initializeActorDatabase($driver); + $runId = $this->createRun('wf-stream-race-open'); + + // The first append pauses after inserting the stream row, before it + // commits; the second append's insert waits on that row. + [$first, $second] = $this->race($runId, pause: 'insert', firstItems: 3, secondItems: 3); + + $this->assertSame(['outcome' => 'accepted', 'accepted' => 3], $first); + $this->assertSame(['outcome' => 'full'], $second); + $this->assertSame(1, WorkflowDurableStream::query()->where('workflow_run_id', $runId)->count()); + $this->assertSame(3, WorkflowDurableStreamItem::query()->where('workflow_run_id', $runId)->count()); + $this->assertSame(3, (int) WorkflowDurableStream::query()->where('workflow_run_id', $runId)->value('pending_items')); + } + + public static function contended(): array + { + $cases = []; + foreach (array_keys(self::drivers()) as $driver) { + // 1 stored + 3 + 3 would pass the cap: the second batch is refused whole. + $cases[$driver.' over'] = [$driver, 3, ['outcome' => 'full'], 4]; + // 1 stored + 3 + 1 meets the cap exactly: the second batch is admitted. + $cases[$driver.' at'] = [$driver, 1, ['outcome' => 'accepted', 'accepted' => 1], 5]; + } + + return $cases; + } + + #[DataProvider('contended')] + public function test_contended_appends_decide_capacity_against_the_committed_stream(string $driver, int $secondItems, array $expected, int $total): void + { + $this->initializeActorDatabase($driver); + $runId = $this->createRun('wf-stream-race-locked'); + $run = WorkflowRun::query()->findOrFail($runId); + app(WorkflowStreamService::class)->append($run, 'default', 'tokens', [['payload' => 0]], self::CAP); + + // The first append pauses holding the stream row's lock; the second + // waits for it. + [$first, $second] = $this->race($runId, pause: 'lock', firstItems: 3, secondItems: $secondItems); + + $this->assertSame(['outcome' => 'accepted', 'accepted' => 3], $first); + $this->assertSame($expected, $second); + $this->assertSame($total, WorkflowDurableStreamItem::query()->where('workflow_run_id', $runId)->count()); + $this->assertSame($total, (int) WorkflowDurableStream::query()->where('workflow_run_id', $runId)->value('pending_items')); + $this->assertSame(range(0, $total - 1), WorkflowDurableStreamItem::query()->where('workflow_run_id', $runId) + ->orderBy('offset')->pluck('offset')->map(static fn ($offset): int => (int) $offset)->all()); + } + + /** + * The retention pass reads a namespace's due streams from the + * (namespace, reclaim_at, id) index: a range scan in deadline order with + * no sort, which stops at the limit. Each namespace here mostly holds + * rows the pass must not read: open streams and reclaimed terminal rows + * (no deadline) and streams not yet due. MySQL and MariaDB show the rows + * read in their handler counters. PostgreSQL's planner prefers a + * sequential scan of a small table, so its plan is read with sequential + * scans disabled, which shows the index serves predicate and order. + */ + #[DataProvider('serverDrivers')] + public function test_due_stream_discovery_reads_the_deadline_index(string $driver): void + { + $this->initializeActorDatabase($driver); + $now = now()->startOfSecond(); + foreach (['default', 'elsewhere'] as $namespace) { + $rows = []; + for ($n = 0; $n < 2400; $n++) { + // 1800 without a deadline, 570 not yet due, 30 due. + $reclaimAt = match (true) { + $n < 1800 => null, + $n < 2370 => $now->copy()->addSeconds($n), + default => $now->copy()->subSeconds($n), + }; + $rows[] = [ + 'namespace' => $namespace, + 'workflow_instance_id' => 'plan-'.$namespace, + 'workflow_run_id' => sprintf('%s%08d', $namespace === 'default' ? 'D' : 'E', $n), + 'stream_name' => 'stream', + 'status' => $reclaimAt === null && $n % 2 === 0 ? 'open' : 'closed', + 'closed_at' => $reclaimAt === null && $n % 2 === 0 ? null : $now, + 'reclaim_at' => $reclaimAt, + 'created_at' => $now, + 'updated_at' => $now, + ]; + } + foreach (array_chunk($rows, 400) as $chunk) { + DB::table('workflow_durable_streams')->insert($chunk); + } + } + + $query = WorkflowStreamRetention::dueStreamsQuery('default', 10, $now)->toBase(); + $sql = $query->toSql(); + + if ($driver === 'pgsql') { + DB::statement('ANALYZE workflow_durable_streams'); + $plan = DB::transaction(static function () use ($sql, $query): string { + DB::statement('SET LOCAL enable_seqscan = off'); + + return collect(DB::select('EXPLAIN (ANALYZE, COSTS OFF, TIMING OFF, SUMMARY OFF) '.$sql, $query->getBindings())) + ->map(static fn (object $row): string => (string) $row->{'QUERY PLAN'})->implode("\n"); + }); + $this->assertMatchesRegularExpression('/Index (Only )?Scan using wf_durable_streams_ns_reclaim_idx .*\(actual rows=10 loops=1\)/', $plan, $plan); + $this->assertStringNotContainsString('Sort', $plan, $plan); + $this->assertStringNotContainsString('Rows Removed', $plan, $plan); + } else { + DB::statement('ANALYZE TABLE workflow_durable_streams'); + $plan = DB::select('EXPLAIN '.$sql, $query->getBindings()); + $this->assertCount(1, $plan, json_encode($plan)); + $this->assertSame('wf_durable_streams_ns_reclaim_idx', $plan[0]->key, json_encode($plan)); + $this->assertSame('range', $plan[0]->type, json_encode($plan)); + $this->assertStringNotContainsString('filesort', (string) ($plan[0]->Extra ?? ''), json_encode($plan)); + // One PDO handle, so the counters are this session's own. MySQL + // also counts the data-dictionary reads of a table's or a status + // view's first use, so both run once before the counted read. + $pdo = DB::connection()->getPdo(); + $statement = $pdo->prepare($sql); + $statement->execute(DB::connection()->prepareBindings($query->getBindings())); + $statement->fetchAll(); + $pdo->query("SHOW SESSION STATUS LIKE 'Handler_read%'")->fetchAll(); + $pdo->exec('FLUSH STATUS'); + $statement->execute(DB::connection()->prepareBindings($query->getBindings())); + $this->assertCount(10, $statement->fetchAll()); + $reads = collect($pdo->query("SHOW SESSION STATUS LIKE 'Handler_read%'")->fetchAll(PDO::FETCH_KEY_PAIR)) + ->map(static fn (mixed $value): int => (int) $value); + // One seek and nine index steps: the limit, not the namespace. + $this->assertLessThanOrEqual(11, $reads['Handler_read_key'] + $reads['Handler_read_next'], json_encode($reads)); + $this->assertSame(0, $reads['Handler_read_rnd_next'], json_encode($reads)); + $plan = json_encode(['explain' => $plan, 'handler_reads' => $reads], JSON_PRETTY_PRINT); + } + fwrite(STDERR, sprintf("[%s due-stream plan]\n%s\n", $driver, $plan)); + + $this->assertCount(10, $query->get()); + } + + /** + * @return array{0: array, 1: array} + */ + private function race(string $runId, string $pause, int $firstItems, int $secondItems): array + { + try { + $first = $this->startActor(fn (): array => $this->append($runId, $firstItems), self::streamPause($pause)); + $this->runUntilPaused($first); + + $second = $this->startActor(fn (): array => $this->append($runId, $secondItems)); + $this->release($second); + $this->awaitWait($second, 'workflow_runs'); + + $this->resume($first); + + return [$this->finishActor($first), $this->finishActor($second)]; + } finally { + DB::purge(); + DB::reconnect(); + } + } + + /** + * Pauses an append right after it inserts the stream row ('insert') or + * locks it ('lock'). SQLite drops FOR UPDATE; its append holds the write + * lock from the run lock's no-op write on. + */ + private static function streamPause(string $pause): Closure + { + return static function (string $sql, string $driver, int $level) use ($pause): bool { + $stream = str_contains($sql, 'workflow_durable_streams') && ! str_contains($sql, 'workflow_durable_stream_items'); + $matches = $pause === 'insert' + ? str_starts_with($sql, 'insert') + : ($driver === 'sqlite' + ? str_starts_with($sql, 'select') + : str_contains($sql, 'for update') || str_contains($sql, 'for no key update')); + + return $stream && $matches; + }; + } + + /** @return array */ + private function append(string $runId, int $count): array + { + $run = WorkflowRun::query()->findOrFail($runId); + $items = array_map(static fn (int $n): array => ['payload' => $n], range(1, $count)); + + try { + $result = app(WorkflowStreamService::class)->append($run, 'default', 'tokens', $items, self::CAP); + + return ['outcome' => 'accepted', 'accepted' => $result['accepted']]; + } catch (StreamFullException) { + return ['outcome' => 'full']; + } + } +} diff --git a/tests/Feature/WorkflowStreamPayloadRaceTest.php b/tests/Feature/WorkflowStreamPayloadRaceTest.php new file mode 100644 index 000000000..bf80a2bac --- /dev/null +++ b/tests/Feature/WorkflowStreamPayloadRaceTest.php @@ -0,0 +1,478 @@ + ['reclaim', 'http', 'cleanup'], + 'http append first, reclamation' => ['reclaim', 'http', 'writer'], + 'command append, reclamation first' => ['reclaim', 'command', 'cleanup'], + 'command append first, reclamation' => ['reclaim', 'command', 'writer'], + 'http append, run prune first' => ['prune', 'http', 'cleanup'], + 'http append first, run prune' => ['prune', 'http', 'writer'], + ] as $name => $case) { + $cases[$driver.': '.$name] = [$driver, ...$case]; + } + } + + return $cases; + } + + /** + * Another run's expired stream (reclaim), or another expired run (prune), + * holds the only reference to an object a writer references again. + */ + #[DataProvider('interleavings')] + public function test_a_writer_and_payload_cleanup_decide_a_shared_object_under_its_lock( + string $driver, + string $cleanup, + string $writer, + string $first, + ): void { + $this->initializeDatabase($driver); + $object = $this->registerObject('shared'); + $oldRunId = $this->summarizedRun('wf-payload-race-old', $cleanup === 'prune' ? RunStatus::Completed : RunStatus::Running); + $this->addStream($oldRunId, 'old', [$object['uri']], expired: $cleanup === 'reclaim'); + $runId = $this->summarizedRun('wf-payload-race-writer'); + $taskId = $writer === 'command' ? $this->createWorkflowTask($runId) : null; + + $cleanupWork = static fn (): array => $cleanup === 'reclaim' + ? WorkflowStreamRetention::pruneExpired('default', 10, 1000) + : HistoryRetentionEnforcer::pruneRun('default', $oldRunId); + $writerWork = fn (): array => $writer === 'http' + ? $this->httpAppend('wf-payload-race-writer', $runId, [$object['reference']]) + : $this->commandAppend($runId, $taskId, [$object['uri']]); + // Cleanup pauses holding the object's lock, before it decides. + $cleanupHoldsTheObject = self::after('update', self::OBJECT_LOCKS, minLevel: 2); + + if ($first === 'cleanup') { + // The HTTP append has fetched and verified the object, outside + // its transaction; the command append's runtime transport + // resolved it before the completion. + $writerActor = $this->startActor($writerWork, $writer === 'http' ? [self::afterVerification()] : []); + if ($writer === 'http') { + $this->runUntilPaused($writerActor); + } + $cleanupActor = $this->startActor($cleanupWork, $cleanupHoldsTheObject); + $this->runUntilPaused($cleanupActor); + if ($writer === 'http') { + $this->resume($writerActor); + } else { + $this->release($writerActor); + } + $this->awaitWait($writerActor, self::OBJECT_LOCKS); + $this->resume($cleanupActor); + + $cleaned = $this->finishActor($cleanupActor); + $written = $this->finishActor($writerActor); + + // The object went first; the writer's claim found no + // registration, and its append wrote nothing. + $this->assertSame($writer === 'http' + ? ['status' => 422, 'reason' => 'external_payload_unsupported'] + : ['outcome' => 'refused', 'reason' => 'external_payload_not_found'], $written); + $this->assertCleanup($cleanup, $cleaned, objectDeleted: true); + $this->assertObjectDeleted($object); + $this->assertSame(0, WorkflowDurableStreamItem::query()->where('workflow_run_id', $runId)->count()); + $this->assertSame(0, WorkflowDurableStreamItem::query()->where('payload_reference', $object['uri'])->count()); + + return; + } + + // The writer has claimed the object and written its item, holding + // the object's lock until it commits. + $writerActor = $this->startActor($writerWork, self::after('insert', 'workflow_durable_stream_items')); + $this->runUntilPaused($writerActor); + $cleanupActor = $this->startActor($cleanupWork); + $this->release($cleanupActor); + $this->awaitWait($cleanupActor, self::OBJECT_LOCKS); + $this->resume($writerActor); + + $written = $this->finishActor($writerActor); + $cleaned = $this->finishActor($cleanupActor); + + // The writer's item committed first; cleanup found it under the + // object's lock and kept the object. + $this->assertSame($writer === 'http' ? ['status' => 200, 'accepted' => 1] : ['outcome' => 'accepted'], $written); + $this->assertCleanup($cleanup, $cleaned, objectDeleted: false); + $this->assertObjectKept($object); + $this->assertSame([$object['uri']], WorkflowDurableStreamItem::query()->where('workflow_run_id', $runId) + ->pluck('payload_reference')->all()); + $this->assertSame(0, WorkflowDurableStreamItem::query()->where('workflow_run_id', $oldRunId)->count()); + } + + /** + * Run retention takes the run's lock before it deletes anything, so an + * append to the same run that comes while it deletes the run's objects + * waits, then finds the run's details pruned and writes nothing. Were + * the append admitted then, retention would delete its item with the + * run's rows and leave the object it claimed unreferenced. + */ + #[DataProvider('drivers')] + public function test_a_late_append_to_a_run_being_pruned_writes_nothing(string $driver): void + { + $this->initializeDatabase($driver); + $old = $this->registerObject('pruned'); + $late = $this->registerObject('late', claimed: false); + $runId = $this->summarizedRun('wf-payload-race-pruned', RunStatus::Completed); + $this->addStream($runId, 'old', [$old['uri']]); + + // The append has verified its object; then retention deletes the + // run's object and pauses before it deletes the run's rows. + $append = $this->startActor( + fn (): array => $this->httpAppend('wf-payload-race-pruned', $runId, [$late['reference']]), + self::afterVerification(), + ); + $this->runUntilPaused($append); + $prune = $this->startActor( + static fn (): array => HistoryRetentionEnforcer::pruneRun('default', $runId), + self::after('delete', 'runtime_external_payloads', minLevel: 2), + ); + $this->runUntilPaused($prune); + $this->resume($append); + $this->awaitWait($append, 'workflow_runs'); + $this->resume($prune); + + $this->assertCleanup('prune', $this->finishActor($prune), objectDeleted: true); + $this->assertSame(['status' => 404, 'reason' => 'stream_not_found'], $this->finishActor($append)); + $this->assertSame(0, WorkflowDurableStream::query()->where('workflow_run_id', $runId)->count()); + $this->assertSame(0, WorkflowDurableStreamItem::query()->where('workflow_run_id', $runId)->count()); + $this->assertObjectDeleted($old); + // The refused append claimed nothing; its upload still expires. + $row = RuntimeExternalPayload::query()->where('storage_uri', $late['uri'])->sole(); + $this->assertNull($row->retained_at); + $this->assertNotNull($row->expires_at); + } + + /** + * Two appends that reference the same two objects, listed in opposite + * orders, claim them in one canonical order, so the second waits for the + * first instead of each holding the lock the other needs. Both commit. + * Each run's stream is already open, so the appends share no lock but + * the objects'. + */ + #[DataProvider('drivers')] + public function test_appends_listing_shared_objects_in_opposite_orders_both_commit(string $driver): void + { + $this->initializeDatabase($driver); + [$a, $b] = $this->objectsInTwoLockBuckets(); + $abRunId = $this->summarizedRun('wf-payload-race-ab'); + $baRunId = $this->summarizedRun('wf-payload-race-ba'); + $this->addStream($abRunId, 'live', []); + $this->addStream($baRunId, 'live', []); + $claimsFirst = self::after('update', self::OBJECT_LOCKS, minLevel: 2); + + $ab = fn (): array => $this->httpAppend('wf-payload-race-ab', $abRunId, [$a['reference'], $b['reference']]); + $ba = fn (): array => $this->httpAppend('wf-payload-race-ba', $baRunId, [$b['reference'], $a['reference']]); + + if ($driver === 'sqlite') { + // One writer at a time: the second append waits for the first's + // write lock. + $first = $this->startActor($ab, $claimsFirst); + $this->runUntilPaused($first); + $second = $this->startActor($ba); + $this->release($second); + $this->awaitWait($second, self::OBJECT_LOCKS); + } else { + // Both appends verify their objects and lock their runs; then the + // first claims its first object and holds it. + $first = $this->startActor($ab, [self::afterRunLock(), $claimsFirst]); + $this->runUntilPaused($first); + $second = $this->startActor($ba, [self::afterRunLock()]); + $this->runUntilPaused($second); + $this->resumeUntilPaused($first); + $this->resume($second); + $this->awaitWait($second, self::OBJECT_LOCKS); + } + $this->resume($first); + + $this->assertSame(['status' => 200, 'accepted' => 2], $this->finishActor($first)); + $this->assertSame(['status' => 200, 'accepted' => 2], $this->finishActor($second)); + foreach ([$abRunId, $baRunId] as $runId) { + $this->assertEqualsCanonicalizing([$a['uri'], $b['uri']], WorkflowDurableStreamItem::query() + ->where('workflow_run_id', $runId)->pluck('payload_reference')->all()); + } + $this->assertObjectKept($a); + $this->assertObjectKept($b); + } + + private function initializeDatabase(string $driver): void + { + $this->initializeActorDatabase($driver); + $this->storageDirectory = storage_path('framework/testing/stream-payload-race-'.bin2hex(random_bytes(4))); + File::deleteDirectory($this->storageDirectory); + WorkflowNamespace::query()->where('name', 'default')->update([ + 'external_payload_storage' => [ + 'driver' => 'local', + 'enabled' => true, + 'config' => ['uri' => 'file://'.$this->storageDirectory], + ], + ]); + } + + protected function tearDown(): void + { + $this->tearDownActors(); + if ($this->storageDirectory !== null) { + File::deleteDirectory($this->storageDirectory); + } + parent::tearDown(); + } + + /** + * A run with its summary, as every started run has. Payload cleanup + * counts a reference as retained while its run's summary remains. + */ + private function summarizedRun(string $workflowId, RunStatus $status = RunStatus::Running): string + { + $runId = $this->createRun($workflowId, $status); + RunSummaryProjector::project(WorkflowRun::query()->findOrFail($runId)); + + return $runId; + } + + /** + * An uploaded object; claimed, as one a stored item references is, unless + * the test uploaded it for an append still to come. + */ + private function registerObject(string $value, bool $claimed = true): array + { + $bytes = Serializer::serializeWithCodec('avro', [$value]); + $registry = app(RuntimeExternalPayloadRegistry::class); + $reference = $registry->upload('default', $bytes, 'avro', hash('sha256', $bytes)); + $uri = (string) RuntimeExternalPayload::query()->whereKey($reference['reference_id'])->value('storage_uri'); + if ($claimed) { + $registry->claimRegistered('default', $uri); + } + + return ['uri' => $uri, 'reference' => $reference]; + } + + /** @return array{0: array, 1: array} Two objects in different lock buckets, in lock order. */ + private function objectsInTwoLockBuckets(): array + { + $first = $this->registerObject('object-0'); + for ($n = 1; $n < 64; $n++) { + $second = $this->registerObject('object-'.$n); + if (RuntimeExternalPayloadObjectLock::bucketFor($second['uri']) !== RuntimeExternalPayloadObjectLock::bucketFor($first['uri'])) { + $ordered = RuntimeExternalPayloadObjectLock::inLockOrder([$first['uri'], $second['uri']]); + + return $ordered[0] === $first['uri'] ? [$first, $second] : [$second, $first]; + } + } + $this->fail('No two objects in different lock buckets.'); + } + + /** @param list $uris */ + private function addStream(string $runId, string $name, array $uris, bool $expired = false): void + { + $run = WorkflowRun::query()->findOrFail($runId); + $at = now()->subDay(); + DB::transaction(static function () use ($run, $name, $uris, $expired, $at): void { + $stream = WorkflowDurableStream::query()->create([ + 'namespace' => 'default', + 'workflow_instance_id' => $run->workflow_instance_id, + 'workflow_run_id' => $run->id, + 'stream_name' => $name, + 'status' => $expired ? WorkflowDurableStream::STATUS_CLOSED : WorkflowDurableStream::STATUS_OPEN, + 'last_offset' => count($uris) - 1, + 'total_items' => count($uris), + 'pending_items' => count($uris), + 'opened_at' => $at, + 'closed_at' => $expired ? $at : null, + 'retention_seconds' => 3600, + 'reclaim_at' => $expired ? $at->copy()->addHour() : null, + ]); + foreach ($uris as $offset => $uri) { + WorkflowDurableStreamItem::query()->create([ + 'stream_id' => $stream->id, + 'namespace' => 'default', + 'workflow_run_id' => $run->id, + 'stream_name' => $name, + 'offset' => $offset, + 'payload_reference' => $uri, + 'payload_codec' => 'avro', + 'emitted_at' => $at, + ]); + } + }); + } + + private function createWorkflowTask(string $runId): string + { + return (string) WorkflowTask::query()->create([ + 'id' => (string) Str::ulid(), + 'workflow_run_id' => $runId, + 'namespace' => 'default', + 'task_type' => TaskType::Workflow->value, + 'status' => TaskStatus::Ready->value, + 'queue' => 'default', + 'payload' => ['workflow_command_id' => 'race-command'], + 'available_at' => now(), + ])->id; + } + + /** @param list> $references Opaque runtime references. */ + private function httpAppend(string $workflowId, string $runId, array $references): array + { + $response = $this->withHeaders($this->apiHeaders())->postJson( + sprintf('/api/workflows/%s/runs/%s/streams/live/items', $workflowId, $runId), + ['items' => array_map(static fn (array $reference): array => [ + 'payload_reference' => $reference, + 'payload_codec' => 'avro', + ], $references)], + ); + + return $response->status() === 200 + ? ['status' => 200, 'accepted' => $response->json('accepted')] + : ['status' => $response->status(), 'reason' => $response->json('reason')]; + } + + /** + * A workflow command's append, inside a transaction that already holds + * the run, as a worker completion's does. The runtime transport has + * resolved each reference to its provider URI. + * + * @param list $uris + */ + private function commandAppend(string $runId, string $taskId, array $uris): array + { + try { + DB::transaction(static function () use ($runId, $taskId, $uris): void { + WorkflowRunStreamLock::acquire($runId); + app(WorkflowStreamCommandProcessor::class)->process($taskId, 'default', [[ + 'type' => 'record_side_effect', + 'result' => ['codec' => 'avro', 'blob' => 'recorded-null'], + 'workflow_stream' => [ + 'operation' => 'append', + 'stream_name' => 'live', + 'command_identity' => 'race-command', + 'command_ordinal' => 0, + 'items' => array_map(static fn (string $uri, int $index): array => [ + 'payload_reference' => $uri, + 'payload_codec' => 'avro', + 'idempotency_key' => 'dw-stream:race-command:0:'.$index, + ], $uris, array_keys($uris)), + ], + ]]); + }); + + return ['outcome' => 'accepted']; + } catch (RuntimeExternalPayloadException $exception) { + return ['outcome' => 'refused', 'reason' => $exception->reason]; + } + } + + private function assertCleanup(string $cleanup, array $result, bool $objectDeleted): void + { + if ($cleanup === 'reclaim') { + $this->assertSame(['streams_reclaimed' => 1, 'items_removed' => 1, 'budget_exhausted' => false, 'blocked_reason' => null], $result); + + return; + } + $this->assertTrue($result['pruned'], json_encode($result)); + $this->assertSame($objectDeleted ? 1 : 0, $result['external_payloads_deleted'], json_encode($result)); + } + + private function assertObjectDeleted(array $object): void + { + $this->assertFileDoesNotExist(rawurldecode((string) parse_url($object['uri'], PHP_URL_PATH))); + $this->assertSame(0, RuntimeExternalPayload::query()->where('storage_uri', $object['uri'])->count()); + } + + private function assertObjectKept(array $object): void + { + $this->assertFileExists(rawurldecode((string) parse_url($object['uri'], PHP_URL_PATH))); + $row = RuntimeExternalPayload::query()->where('storage_uri', $object['uri'])->sole(); + $this->assertNotNull($row->retained_at); + $this->assertNull($row->expires_at); + } + + /** Right after a $verb statement on $table at transaction level $minLevel or deeper. */ + private static function after(string $verb, string $table, int $minLevel = 0): Closure + { + return static fn (string $sql, string $driver, int $level): bool => $level >= $minLevel + && str_starts_with($sql, $verb) + && str_contains($sql, $table); + } + + /** + * Right after an append takes its run's stream lock: a locking read of + * the run row, or on SQLite the no-op write that takes the write lock. + */ + private static function afterRunLock(): Closure + { + return static fn (string $sql, string $driver, int $level): bool => $level >= 1 + && str_contains($sql, 'workflow_runs') + && ($driver === 'sqlite' ? str_starts_with($sql, 'update') : str_contains($sql, 'for update')); + } + + /** + * Right after an HTTP append's verification commits: it has read the + * object's registration under the object's lock and released it, and + * holds no lock. + */ + private static function afterVerification(): Closure + { + $verified = false; + + return static function (string $sql, string $driver, int $level) use (&$verified): bool { + if ($level >= 1 && str_starts_with($sql, 'select') && str_contains($sql, 'runtime_external_payloads')) { + $verified = true; + } + + return $verified && $sql === 'commit' && $level === 0; + }; + } +} diff --git a/tests/Feature/WorkflowStreamsTest.php b/tests/Feature/WorkflowStreamsTest.php index 3411be2e3..19c6489a1 100644 --- a/tests/Feature/WorkflowStreamsTest.php +++ b/tests/Feature/WorkflowStreamsTest.php @@ -7,13 +7,18 @@ use App\Models\WorkflowDurableStream; use App\Models\WorkflowDurableStreamItem; use App\Models\WorkflowNamespace; +use App\Support\HistoryRetentionEnforcer; +use App\Support\NamespaceDurableStateException; use App\Support\RuntimeExternalPayloadRegistry; use App\Support\WorkerProtocol; use App\Support\WorkflowStreamCommandProcessor; +use App\Support\WorkflowStreamRetention; use App\Support\WorkflowStreamService; use Illuminate\Foundation\Testing\RefreshDatabase; +use Illuminate\Support\Facades\DB; use Illuminate\Support\Facades\File; use Illuminate\Support\Str; +use PHPUnit\Framework\Attributes\DataProvider; use Tests\Feature\Concerns\ServerTestHelpers; use Tests\Fixtures\ExternalGreetingWorkflow; use Tests\TestCase; @@ -225,14 +230,319 @@ public function test_pending_items_cap_returns_429_stream_full(): void ->assertJsonPath('reason', 'stream_full') ->assertJsonPath('max_pending_items', 2); - // The first two items did land before the cap kicked in; - // ordering is preserved. - $this->assertSame( - 2, - WorkflowDurableStreamItem::query() - ->where('workflow_run_id', $run->id) - ->count(), - ); + // Server #774: the whole batch is refused before any item is + // written, and the stream this append would have opened is not + // created, so a plain retry duplicates nothing. + $this->assertSame(0, WorkflowDurableStreamItem::query()->where('workflow_run_id', $run->id)->count()); + $this->assertSame(0, WorkflowDurableStream::query()->where('workflow_run_id', $run->id)->count()); + } + + public function test_capacity_is_decided_for_the_whole_batch_and_duplicates_take_none(): void + { + $run = $this->createRun('default', 'wf-streams-batch-cap'); + $route = $this->itemsRoute('wf-streams-batch-cap', $run->id, 'tokens'); + $append = fn (array $items) => $this->withHeaders($this->apiHeaders()) + ->postJson($route, ['max_pending_items' => 3, 'items' => $items]); + + $append([['payload' => ['n' => 0], 'idempotency_key' => 'k0']])->assertOk()->assertJsonPath('accepted', 1); + $append([['payload' => ['n' => 1]], ['payload' => ['n' => 2]]])->assertOk()->assertJsonPath('accepted', 2); + + // At the cap: one more fresh item is refused with nothing written. + $append([['payload' => ['n' => 3]]])->assertStatus(429)->assertJsonPath('pending_items', 3); + // A batch that repeats a stored key and adds a fresh one is refused + // whole: the duplicate is not answered and the fresh item not written. + $append([['payload' => ['n' => 0], 'idempotency_key' => 'k0'], ['payload' => ['n' => 4]]])->assertStatus(429); + // A batch of duplicates only takes no capacity and answers its offsets. + $append([['payload' => ['n' => 0], 'idempotency_key' => 'k0'], ['payload' => ['n' => 0], 'idempotency_key' => 'k0']]) + ->assertOk() + ->assertJsonPath('accepted', 0) + ->assertJsonPath('deduped', 2) + ->assertJsonPath('accepted_offsets', [0, 0]); + + $this->assertSame([0, 1, 2], WorkflowDurableStreamItem::query()->where('workflow_run_id', $run->id) + ->orderBy('offset')->pluck('offset')->map(static fn ($offset): int => (int) $offset)->all()); + $this->assertSame(3, (int) WorkflowDurableStream::query()->where('workflow_run_id', $run->id)->value('pending_items')); + } + + public function test_producer_controls_are_bounded_at_every_ingress(): void + { + $run = $this->createRun('default', 'wf-streams-bounds'); + $items = $this->itemsRoute('wf-streams-bounds', $run->id, 'tokens'); + + $this->withHeaders($this->apiHeaders()) + ->postJson($items, ['max_pending_items' => WorkflowStreamService::MAX_PENDING_ITEMS_MAX + 1, 'items' => [['payload' => 1]]]) + ->assertStatus(422) + ->assertJsonPath('errors.max_pending_items.0', 'The max pending items field must not be greater than 10000.'); + $this->assertSame(0, WorkflowDurableStream::query()->where('workflow_run_id', $run->id)->count()); + + $this->withHeaders($this->apiHeaders()) + ->postJson($items, ['max_pending_items' => WorkflowStreamService::MAX_PENDING_ITEMS_MAX, 'items' => [['payload' => 1]]]) + ->assertOk(); + + $close = $this->closeRoute('wf-streams-bounds', $run->id, 'tokens'); + $this->withHeaders($this->apiHeaders()) + ->postJson($close, ['retention_seconds' => WorkflowStreamService::RETENTION_SECONDS_MAX + 1]) + ->assertStatus(422) + ->assertJsonPath('errors.retention_seconds.0', 'The retention seconds field must not be greater than 2592000.'); + $this->assertSame('open', WorkflowDurableStream::query()->where('workflow_run_id', $run->id)->value('status')); + + $this->withHeaders($this->apiHeaders()) + ->postJson($close, ['retention_seconds' => WorkflowStreamService::RETENTION_SECONDS_MAX]) + ->assertOk() + ->assertJsonPath('stream.retention_seconds', WorkflowStreamService::RETENTION_SECONDS_MAX); + + // The service refuses the same values for callers that skip request + // validation. + $streams = app(WorkflowStreamService::class); + foreach ([0, WorkflowStreamService::MAX_PENDING_ITEMS_MAX + 1] as $cap) { + try { + $streams->append($run, 'default', 'other', [['payload' => 1]], $cap); + $this->fail("cap {$cap} accepted"); + } catch (\InvalidArgumentException) { + $this->addToAssertionCount(1); + } + } + $this->expectException(\InvalidArgumentException::class); + $streams->close($run, 'default', 'tokens', null, WorkflowStreamService::RETENTION_SECONDS_MAX + 1); + } + + public static function oversizedStreamCommands(): array + { + return [ + 'max_pending_items' => ['max_pending_items', WorkflowStreamService::MAX_PENDING_ITEMS_MAX + 1, 'append'], + 'retention_seconds' => ['retention_seconds', WorkflowStreamService::RETENTION_SECONDS_MAX + 1, 'close'], + ]; + } + + #[DataProvider('oversizedStreamCommands')] + public function test_workflow_stream_commands_bound_producer_controls_before_admission(string $field, int $value, string $operation): void + { + $task = $this->claimStreamCompletionTask(); + $command = $this->streamCompletionCommand($task); + $command['workflow_stream'][$field] = $value; + if ($operation === 'close') { + $command['workflow_stream']['operation'] = 'close'; + unset($command['workflow_stream']['items']); + } + + $this->withHeaders($this->workerHeaders())->postJson("/api/worker/workflow-tasks/{$task['task_id']}/complete", [ + 'lease_owner' => $task['lease_owner'], 'workflow_task_attempt' => $task['workflow_task_attempt'], + 'commands' => [$command], + ])->assertStatus(422); + $this->assertDatabaseCount('workflow_durable_streams', 0); + $this->assertDatabaseMissing('workflow_history_events', ['workflow_run_id' => $task['run_id'], 'event_type' => 'SideEffectRecorded']); + } + + public function test_an_expired_stream_reads_as_not_found(): void + { + $run = $this->createRun('default', 'wf-streams-expiry'); + $this->postItems($run->id, 'tokens', [['payload' => 1]], 'wf-streams-expiry'); + $this->postItems($run->id, 'kept', [['payload' => 1]], 'wf-streams-expiry'); + $this->withHeaders($this->apiHeaders()) + ->postJson($this->closeRoute('wf-streams-expiry', $run->id, 'tokens'), ['retention_seconds' => 60]) + ->assertOk(); + + $describe = sprintf('/api/workflows/%s/runs/%s/streams/%s', 'wf-streams-expiry', $run->id, 'tokens'); + $this->withHeaders($this->apiHeaders())->getJson($describe)->assertOk(); + + $this->travel(61)->seconds(); + + $this->withHeaders($this->apiHeaders())->getJson($describe)->assertNotFound()->assertJsonPath('reason', 'stream_not_found'); + $this->withHeaders($this->apiHeaders())->getJson($this->itemsRoute('wf-streams-expiry', $run->id, 'tokens')) + ->assertNotFound()->assertJsonPath('reason', 'stream_not_found'); + $this->withHeaders($this->apiHeaders())->postJson($this->closeRoute('wf-streams-expiry', $run->id, 'tokens'), []) + ->assertNotFound()->assertJsonPath('reason', 'stream_not_found'); + $this->withHeaders($this->apiHeaders()) + ->getJson(sprintf('/api/workflows/%s/runs/%s/streams', 'wf-streams-expiry', $run->id)) + ->assertOk() + ->assertJsonCount(1, 'streams') + ->assertJsonPath('streams.0.stream_name', 'kept'); + // Append is not found too, and writes nothing: the terminal row stays + // until the run is pruned, so the name is never opened again. + $this->withHeaders($this->apiHeaders()) + ->postJson($this->itemsRoute('wf-streams-expiry', $run->id, 'tokens'), ['items' => [['payload' => 2]]]) + ->assertNotFound()->assertJsonPath('reason', 'stream_not_found'); + $stream = WorkflowDurableStream::query()->where('workflow_run_id', $run->id)->where('stream_name', 'tokens')->sole(); + $this->assertSame(WorkflowDurableStream::STATUS_CLOSED, $stream->status); + $this->assertSame(1, (int) $stream->total_items); + $this->assertSame(1, WorkflowDurableStreamItem::query()->where('stream_id', $stream->id)->count()); + } + + public function test_a_stream_expires_at_the_instant_its_retention_ends(): void + { + $run = $this->createRun('default', 'wf-streams-expiry-instant'); + $this->postItems($run->id, 'tokens', [['payload' => 1]], 'wf-streams-expiry-instant'); + $this->withHeaders($this->apiHeaders()) + ->postJson($this->closeRoute('wf-streams-expiry-instant', $run->id, 'tokens'), ['retention_seconds' => 60]) + ->assertOk(); + $append = fn () => $this->withHeaders($this->apiHeaders()) + ->postJson($this->itemsRoute('wf-streams-expiry-instant', $run->id, 'tokens'), ['items' => [['payload' => 2]]]); + + $this->travel(59)->seconds(); + $append()->assertStatus(409)->assertJsonPath('reason', 'stream_closed'); + + $this->travel(1)->seconds(); + $append()->assertNotFound()->assertJsonPath('reason', 'stream_not_found'); + $this->assertSame(1, WorkflowDurableStreamItem::query()->where('workflow_run_id', $run->id)->count()); + } + + public function test_an_append_that_opens_a_stream_counts_against_the_namespace_stream_quota(): void + { + $run = $this->createRun('default', 'wf-streams-quota'); + $streams = app(WorkflowStreamService::class); + config(['server.namespace_durable_state.overrides' => [ + 'default' => ['max_workflow_streams' => 1], + ]]); + + $streams->append($run, 'default', 'first', [['payload' => 1]]); + // An existing stream takes no more of the quota. + $streams->append($run, 'default', 'first', [['payload' => 2]]); + + try { + $streams->append($run, 'default', 'second', [['payload' => 1]]); + $this->fail('Expected the namespace stream quota to refuse a second stream.'); + } catch (NamespaceDurableStateException $exception) { + $this->assertSame('namespace_workflow_streams_exhausted', $exception->reason); + } + + $this->assertSame(['first'], WorkflowDurableStream::query()->pluck('stream_name')->all()); + $this->assertSame(2, WorkflowDurableStreamItem::query()->count()); + } + + public function test_the_retention_pass_reclaims_expired_streams_in_bounded_resumable_steps(): void + { + $run = $this->createRun('default', 'wf-streams-sweep'); + $streams = app(WorkflowStreamService::class); + $streams->append($run, 'default', 'first', array_map(static fn (int $n): array => ['payload' => $n], range(1, 5))); + $streams->close($run, 'default', 'first', null, 60); + $this->travel(1)->seconds(); + $streams->append($run, 'default', 'second', [['payload' => 1], ['payload' => 2]]); + $streams->close($run, 'default', 'second', 'failed', 60); + $streams->append($run, 'default', 'long', [['payload' => 1]]); + $streams->close($run, 'default', 'long', null, 3600); + $streams->append($run, 'default', 'open', [['payload' => 1]]); + $this->createNamespace('elsewhere'); + $other = $this->createRun('elsewhere', 'wf-streams-sweep-other'); + $streams->append($other, 'elsewhere', 'foreign', [['payload' => 1]]); + $streams->close($other, 'elsewhere', 'foreign', null, 60); + + // Nothing is due yet. + $this->assertSame(['streams_reclaimed' => 0, 'items_removed' => 0, 'budget_exhausted' => false, 'blocked_reason' => null], + WorkflowStreamRetention::pruneExpired('default', 10, 1000)); + + $this->travel(120)->seconds(); + + // A pass deletes at most its item budget, earliest deadline first. + $this->assertSame(['streams_reclaimed' => 0, 'items_removed' => 3, 'budget_exhausted' => true, 'blocked_reason' => null], + WorkflowStreamRetention::pruneExpired('default', 10, 3)); + $this->assertSame(2, WorkflowDurableStreamItem::query()->where('stream_name', 'first')->count()); + $this->assertSame(2, WorkflowDurableStreamItem::query()->where('stream_name', 'second')->count()); + + // The next pass resumes: the rest of `first`, then `second`. + $this->assertSame(['streams_reclaimed' => 2, 'items_removed' => 4, 'budget_exhausted' => false, 'blocked_reason' => null], + WorkflowStreamRetention::pruneExpired('default', 10, 1000)); + $this->assertSame(['long' => 1, 'open' => 1], WorkflowDurableStreamItem::query()->where('workflow_run_id', $run->id) + ->orderBy('stream_name')->pluck('stream_name')->countBy()->all()); + // Reclaimed streams keep their rows, with no deadline left. + $this->assertSame(['first', 'long', 'open', 'second'], WorkflowDurableStream::query() + ->where('workflow_run_id', $run->id)->orderBy('stream_name')->pluck('stream_name')->all()); + $this->assertSame(['first', 'open', 'second'], WorkflowDurableStream::query() + ->where('workflow_run_id', $run->id)->whereNull('reclaim_at')->orderBy('stream_name')->pluck('stream_name')->all()); + // Nothing is left to read for `default`; another namespace's stream waits for its own pass. + $this->assertSame([], WorkflowStreamRetention::dueStreams('default', 10, now())); + $this->assertSame(1, WorkflowDurableStreamItem::query()->where('stream_name', 'foreign')->count()); + + // The stream-count bound: one stream per pass. + $this->assertSame(['streams_reclaimed' => 1, 'items_removed' => 1, 'budget_exhausted' => false, 'blocked_reason' => null], + WorkflowStreamRetention::pruneExpired('elsewhere', 1, 1000)); + } + + public function test_a_reclaimed_stream_keeps_its_identity_until_its_run_is_pruned(): void + { + $run = $this->createRun('default', 'wf-streams-tombstone'); + $this->postItems($run->id, 'tokens', [['payload' => 1], ['payload' => 2]], 'wf-streams-tombstone'); + $this->withHeaders($this->apiHeaders()) + ->postJson($this->closeRoute('wf-streams-tombstone', $run->id, 'tokens'), ['retention_seconds' => 60]) + ->assertOk(); + $this->travel(60)->seconds(); + + WorkflowStreamRetention::pruneExpired('default', 10, 1000); + + $this->assertSame(0, WorkflowDurableStreamItem::query()->where('workflow_run_id', $run->id)->count()); + $stream = WorkflowDurableStream::query()->where('workflow_run_id', $run->id)->sole(); + $this->assertNull($stream->reclaim_at); + // The name is not opened again: append, describe and close stay not found. + $this->withHeaders($this->apiHeaders()) + ->postJson($this->itemsRoute('wf-streams-tombstone', $run->id, 'tokens'), ['items' => [['payload' => 3]]]) + ->assertNotFound()->assertJsonPath('reason', 'stream_not_found'); + $this->withHeaders($this->apiHeaders()) + ->getJson(sprintf('/api/workflows/%s/runs/%s/streams/%s', 'wf-streams-tombstone', $run->id, 'tokens')) + ->assertNotFound(); + $this->withHeaders($this->apiHeaders()) + ->postJson($this->closeRoute('wf-streams-tombstone', $run->id, 'tokens'), []) + ->assertNotFound(); + $this->assertSame(1, WorkflowDurableStream::query()->where('workflow_run_id', $run->id)->count()); + $this->assertSame(0, WorkflowDurableStreamItem::query()->where('workflow_run_id', $run->id)->count()); + } + + public function test_worker_heartbeat_and_history_prune_run_the_stream_retention_pass(): void + { + $run = $this->createRun('default', 'wf-streams-inline'); + $streams = app(WorkflowStreamService::class); + $streams->append($run, 'default', 'inline', [['payload' => 1]]); + $streams->close($run, 'default', 'inline', null, 60); + $streams->append($run, 'default', 'command', [['payload' => 1]]); + $streams->close($run, 'default', 'command', null, 120); + WorkflowNamespace::query()->where('name', 'default')->update(['retention_mode' => WorkflowNamespace::RETENTION_MODE_FOREVER]); + + $this->travel(61)->seconds(); + HistoryRetentionEnforcer::runInlinePass('default'); + $this->assertSame(['command'], WorkflowDurableStreamItem::query()->pluck('stream_name')->all()); + + $this->travel(60)->seconds(); + $this->artisan('history:prune', ['--namespace' => 'default'])->assertExitCode(0); + $this->assertSame(0, WorkflowDurableStreamItem::query()->count()); + $this->assertSame(2, WorkflowDurableStream::query()->whereNull('reclaim_at')->count()); + } + + public function test_the_operator_retention_pass_reclaims_expired_streams_within_its_limit(): void + { + $run = $this->createRun('default', 'wf-streams-operator-pass'); + $streams = app(WorkflowStreamService::class); + $streams->append($run, 'default', 'large', array_map(static fn (int $n): array => ['payload' => $n], range(1, 150))); + $streams->close($run, 'default', 'large', null, 60); + $streams->append($run, 'default', 'small', [['payload' => 1]]); + $streams->close($run, 'default', 'small', null, 60); + $this->travel(61)->seconds(); + + // limit 1: one stream and at most 100 items per pass, resumed by the next. + $pass = fn () => $this->withHeaders($this->apiHeaders())->postJson('/api/system/retention/pass', ['limit' => 1])->assertOk(); + $pass(); + $this->assertSame(50, WorkflowDurableStreamItem::query()->where('stream_name', 'large')->count()); + $pass(); + $this->assertSame(['small'], WorkflowDurableStreamItem::query()->pluck('stream_name')->all()); + $pass(); + $this->assertSame(0, WorkflowDurableStreamItem::query()->count()); + $this->assertSame(2, WorkflowDurableStream::query()->whereNull('reclaim_at')->count()); + } + + public function test_due_stream_discovery_reads_the_deadline_index(): void + { + if (DB::connection()->getDriverName() !== 'sqlite') { + // The query runs here; WorkflowStreamCapacityRaceTest reads the + // MySQL, MariaDB and PostgreSQL plans over realistic data. + $this->assertSame([], WorkflowStreamRetention::dueStreamsQuery('default', 10, now())->pluck('id')->all()); + + return; + } + + $plan = collect(DB::select('EXPLAIN QUERY PLAN '.WorkflowStreamRetention::dueStreamsQuery('default', 10, now())->toRawSql())) + ->pluck('detail')->implode("\n"); + + // A search on both index columns, namespace equal and deadline + // bounded, read in index order. + $this->assertStringContainsString('SEARCH workflow_durable_streams USING INDEX wf_durable_streams_ns_reclaim_idx (namespace=? AND reclaim_atassertStringNotContainsString('TEMP B-TREE', $plan); } public function test_close_marks_stream_terminal_and_rejects_further_appends(): void diff --git a/tests/Unit/Support/composer-sinks.json b/tests/Unit/Support/composer-sinks.json index b81a9fc9e..3e6063bec 100644 --- a/tests/Unit/Support/composer-sinks.json +++ b/tests/Unit/Support/composer-sinks.json @@ -13,12 +13,12 @@ { "type": "server-verify", "line": "&& server-verify-trust final", - "distance": 75 + "distance": 87 }, { "type": "server-verify", "line": "&& server-verify-trust final", - "distance": 75 + "distance": 87 } ] },