Bound workflow stream producers, refuse whole batches at capacity, and expire closed streams (#774)

.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 }}" \

.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 }}

.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() }}

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);
 

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<int, \App\Models\WorkflowDurableStreamItem>, next_offset: int, terminal: bool}
+     * @return array{stream: WorkflowDurableStream, items: Collection<int, WorkflowDurableStreamItem>, next_offset: int, terminal: bool}
      */
     private function readWithWait(
         WorkflowRun $run,

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',
     ];