Take over a stopped PHP PostgreSQL database backup-first

.github/workflows/phpunit-feature.yml

@@ -220,6 +220,21 @@ jobs:
             done
             [ "$health" = healthy ]
           done
+          for mode in session transaction0 transaction200; do
+            pool_mode=transaction
+            prepared_statements=0
+            [ "$mode" != session ] || pool_mode=session
+            [ "$mode" != transaction200 ] || prepared_statements=200
+            docker run --detach \
+              --name "dw-server-pooler-${{ github.run_id }}-${{ github.run_attempt }}-${{ github.job }}-$mode" \
+              --network "dw-server-feature-${{ github.run_id }}-${{ github.run_attempt }}-${{ github.job }}" \
+              --memory 128m --memory-swap 128m \
+              -e DB_HOST="dw-server-backup-pgsql-${{ github.run_id }}-${{ github.run_attempt }}-${{ github.job }}" \
+              -e DB_NAME=postgres -e DB_USER=postgres -e DB_PASSWORD=pooler-disposable-password \
+              -e AUTH_TYPE=trust -e "POOL_MODE=$pool_mode" -e "MAX_PREPARED_STATEMENTS=$prepared_statements" \
+              -e DEFAULT_POOL_SIZE=2 -e MIN_POOL_SIZE=2 -e SERVER_ROUND_ROBIN=1 \
+              edoburu/pgbouncer@sha256:b17551c776ef7e5769ef80b956d20f85e2fd25dd8912d31d58f782aad495b711
+          done
           tar \
             --exclude=vendor \
             --exclude='*/vendor' \
@@ -236,6 +251,9 @@ 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_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 }}" \
                 -w /app \
                 --entrypoint sh \
                 durable-workflow-server-corpus-validator \
@@ -283,6 +301,9 @@ jobs:
       - name: Clean up test services
         if: always()
         run: |
+          for mode in session transaction0 transaction200; do
+            docker rm -f "dw-server-pooler-${{ github.run_id }}-${{ github.run_attempt }}-${{ github.job }}-$mode" >/dev/null 2>&1 || true
+          done
           docker rm -f \
             "dw-server-s3-${{ github.run_id }}-${{ github.run_attempt }}-${{ github.job }}" \
             "dw-server-backup-mysql-${{ github.run_id }}-${{ github.run_attempt }}-${{ github.job }}" \

.github/workflows/server-parity.yml

@@ -377,6 +377,11 @@ jobs:
       - name: Bootstrap independent published PHP and native MySQL databases
         run: |
           set -euo pipefail
+          # Failed Rust tests are printed from their evidence files into this step's log.
+          rust_test_failures() {
+            bash scripts/conformance/server-parity/rust-test-failures.sh
+          }
+          trap rust_test_failures ERR
           mkdir -p cargo-home rust-target parity-evidence mysql-state
           chmod 0777 parity-evidence
           docker network create "$RESOURCE_SCOPE"
@@ -663,14 +668,14 @@ jobs:
           cmp parity-evidence/catalog-before.json parity-evidence/catalog-after.json
       - name: Compare separate configured role-token cohort
         run: bash scripts/conformance/server-parity/role-profile.sh mysql
-      - name: Inspect separate published PHP upgrade source without takeover
+      - name: Back up, take over and resume a separate published PHP upgrade source
         timeout-minutes: 8
         run: bash scripts/conformance/server-parity/upgrade-source-profile.sh mysql
       - name: Retire upgrade source cohort
         if: always()
         timeout-minutes: 3
         run: bash scripts/conformance/server-parity/upgrade-source-cleanup.sh mysql
-      - name: Inspect published PHP 2.9.7 upgrade source without takeover
+      - name: Back up, take over and resume a published PHP 2.9.7 upgrade source
         timeout-minutes: 8
         env:
           DW_PARITY_UPGRADE_LABEL: php297
@@ -871,6 +876,11 @@ jobs:
           POSTGRES_IMAGE: ${{ matrix.image }}
         run: |
           set -euo pipefail
+          # Failed Rust tests are printed from their evidence files into this step's log.
+          rust_test_failures() {
+            bash scripts/conformance/server-parity/rust-test-failures.sh
+          }
+          trap rust_test_failures ERR
           mkdir -p cargo-home rust-target parity-evidence pg-state
           chmod 0777 parity-evidence
           docker network create "$RESOURCE_SCOPE"
@@ -1157,14 +1167,14 @@ jobs:
         env:
           POSTGRES_IMAGE: ${{ matrix.image }}
         run: bash scripts/conformance/server-parity/role-profile.sh pgsql
-      - name: Inspect separate published PHP upgrade source without takeover
+      - name: Back up, take over and resume a separate published PHP upgrade source
         timeout-minutes: 8
         run: bash scripts/conformance/server-parity/upgrade-source-profile.sh pgsql
       - name: Retire upgrade source cohort
         if: always()
         timeout-minutes: 3
         run: bash scripts/conformance/server-parity/upgrade-source-cleanup.sh pgsql
-      - name: Inspect published PHP 2.9.7 upgrade source without takeover
+      - name: Back up, take over and resume a published PHP 2.9.7 upgrade source
         timeout-minutes: 8
         env:
           DW_PARITY_UPGRADE_LABEL: php297
@@ -1394,6 +1404,11 @@ jobs:
       - name: Execute official Rust codec and cross-decode with published PHP
         run: |
           set -euo pipefail
+          # Failed Rust tests are printed from their evidence files into this step's log.
+          rust_test_failures() {
+            bash scripts/conformance/server-parity/rust-test-failures.sh
+          }
+          trap rust_test_failures ERR
           mkdir -p cargo-home rust-target parity-evidence
           # Compiling the library tests peaks near 2.5 GB; a 2 GB limit OOM-killed rustc (exit 101).
           docker run --rm --user="$(id -u):$(id -g)" --cpus=2 --memory=4g --memory-swap=4g \
@@ -1642,7 +1657,7 @@ jobs:
         env:
           RESOURCE_SCOPE: dw-parity-${{ github.run_id }}-${{ github.run_attempt }}-${{ github.job }}
         run: bash scripts/conformance/server-parity/role-profile.sh sqlite
-      - name: Inspect separate published PHP upgrade source without takeover
+      - name: Back up, take over and resume a separate published PHP upgrade source
         timeout-minutes: 8
         env:
           RESOURCE_SCOPE: dw-parity-${{ github.run_id }}-${{ github.run_attempt }}-${{ github.job }}
@@ -1653,9 +1668,10 @@ jobs:
         env:
           RESOURCE_SCOPE: dw-parity-${{ github.run_id }}-${{ github.run_attempt }}-${{ github.job }}
         run: bash scripts/conformance/server-parity/upgrade-source-cleanup.sh sqlite
-      - name: Inspect published PHP 2.9.7 upgrade source without takeover
+      - name: Back up, take over and resume a published PHP 2.9.7 upgrade source
         timeout-minutes: 8
         env:
+          RESOURCE_SCOPE: dw-parity-${{ github.run_id }}-${{ github.run_attempt }}-${{ github.job }}
           DW_PARITY_UPGRADE_LABEL: php297
           DW_PARITY_UPGRADE_EXPECTED_LAYOUT: php-2.9.7
           DW_PARITY_UPGRADE_PHP_IMAGE: ghcr.io/durable-workflow/server@sha256:30c34c5f3256a812fbccbe505a068309b58b0c14d01ffff0465d88604cb36d50
@@ -1664,6 +1680,7 @@ jobs:
         if: always()
         timeout-minutes: 3
         env:
+          RESOURCE_SCOPE: dw-parity-${{ github.run_id }}-${{ github.run_attempt }}-${{ github.job }}
           DW_PARITY_UPGRADE_LABEL: php297
           DW_PARITY_UPGRADE_EXPECTED_LAYOUT: php-2.9.7
           DW_PARITY_UPGRADE_PHP_IMAGE: ghcr.io/durable-workflow/server@sha256:30c34c5f3256a812fbccbe505a068309b58b0c14d01ffff0465d88604cb36d50

app/Http/Controllers/Api/TaskQueueController.php

@@ -10,6 +10,7 @@
 use App\Support\TaskQueueAdmission;
 use App\Support\TaskQueueBuildIdRolloutSnapshot;
 use App\Support\TaskQueuePriorityFairnessSurface;
+use App\Support\WorkflowDefinitionRouting;
 use App\Support\WorkflowQueryTaskBroker;
 use Illuminate\Http\JsonResponse;
 use Illuminate\Http\Request;
@@ -28,6 +29,7 @@ public function __construct(
         private readonly TaskQueueBuildIdRolloutSnapshot $buildIdRollouts,
         private readonly TaskQueuePriorityFairnessSurface $priorityFairness,
         private readonly DeploymentLifecycleService $deployments,
+        private readonly WorkflowDefinitionRouting $definitionRouting,
     ) {}
 
     public function index(Request $request): JsonResponse
@@ -67,22 +69,26 @@ public function show(Request $request, string $taskQueue): JsonResponse
 
         $namespace = (string) $request->attributes->get('namespace');
 
-        return ControlPlaneProtocol::json(
-            $this->withAdmission(
+        $payload = $this->withAdmission(
+            $namespace,
+            $this->withRecentTaskFlow(
                 $namespace,
-                $this->withRecentTaskFlow(
+                $taskQueue,
+                StandaloneWorkerVisibility::queueDetail(
                     $namespace,
                     $taskQueue,
-                    StandaloneWorkerVisibility::queueDetail(
-                        $namespace,
-                        $taskQueue,
-                        WorkerRegistration::class,
-                        now(),
-                        $this->workerStaleAfterSeconds(),
-                    )->toArray(),
-                ),
+                    WorkerRegistration::class,
+                    now(),
+                    $this->workerStaleAfterSeconds(),
+                )->toArray(),
             ),
         );
+        $payload['definition_routing'] = $this->definitionRouting->forQueue($namespace, $taskQueue);
+        if ($payload['definition_routing']['blocked_sample_count'] > 0) {
+            $payload['repair']['needs_attention'] = true;
+        }
+
+        return ControlPlaneProtocol::json($payload);
     }
 
     /**
@@ -263,7 +269,7 @@ private function setBuildIdDrainIntent(Request $request, string $taskQueue, stri
     }
 
     /**
-     * @param array<string, mixed> $deployment
+     * @param  array<string, mixed>  $deployment
      * @return array<string, mixed>
      */
     private function buildIdLifecyclePayload(
@@ -351,7 +357,7 @@ private function normalizeDeploymentLifecycleTimestamps(array $deployment): arra
     }
 
     /**
-     * @param list<array<string, mixed>> $blockages
+     * @param  list<array<string, mixed>>  $blockages
      */
     private function orderBlockages(array &$blockages): void
     {
@@ -630,6 +636,7 @@ private function activityAttemptStateCounts(string $namespace, string $taskQueue
             if ($attemptStatus === ActivityAttemptStatus::Running->value) {
                 $counts['running_count'] += $count;
                 $counts['open_count'] += $count;
+
                 continue;
             }
 
@@ -639,13 +646,15 @@ private function activityAttemptStateCounts(string $namespace, string $taskQueue
             ) {
                 $counts['retrying_count'] += $count;
                 $counts['open_count'] += $count;
+
                 continue;
             }
 
             if ($attemptStatus === ActivityAttemptStatus::Expired->value) {
                 $counts['timed_out_count'] += $count;
                 $counts['expired_count'] += $count;
                 $counts['terminal_count'] += $count;
+
                 continue;
             }
 
@@ -656,18 +665,21 @@ private function activityAttemptStateCounts(string $namespace, string $taskQueue
             ) {
                 $counts['timed_out_count'] += $count;
                 $counts['terminal_count'] += $count;
+
                 continue;
             }
 
             if ($attemptStatus === ActivityAttemptStatus::Failed->value) {
                 $counts['failed_count'] += $count;
                 $counts['terminal_count'] += $count;
+
                 continue;
             }
 
             if ($attemptStatus === ActivityAttemptStatus::Completed->value) {
                 $counts['completed_count'] += $count;
                 $counts['terminal_count'] += $count;
+
                 continue;
             }
 

app/Http/Controllers/Api/WorkerController.php

@@ -46,6 +46,7 @@
 use App\Support\WorkflowHistoryPageToken;
 use App\Support\WorkflowMetadataCapabilityPolicy;
 use App\Support\WorkflowQueryTaskBroker;
+use App\Support\WorkflowReplayEligibility;
 use App\Support\WorkflowStreamCommandProcessor;
 use App\Support\WorkflowStreamService;
 use App\Support\WorkflowTaskLeaseRecovery;
@@ -160,6 +161,7 @@ class WorkerController
         'version_marker_kind_mismatch',
         'version_marker_sequence_invalid',
         'version_range_invalid',
+        'workflow_entry_arguments_mismatch',
         'workflow_nondeterministic',
     ];
 
@@ -407,7 +409,10 @@ public function register(Request $request): JsonResponse
         $this->validateWorkerTaskCapacity($maxWorkflowTasks, $maxActivityTasks);
         $maxWorkerSessions = $validated['max_concurrent_worker_sessions'] ?? 10;
         $taskSlots = is_array($validated['task_slots'] ?? null) ? $validated['task_slots'] : [];
-        $processMetrics = $this->normalizeProcessMetrics($validated['process_metrics'] ?? null);
+        $processMetrics = [
+            ...($this->normalizeProcessMetrics($validated['process_metrics'] ?? null) ?? []),
+            WorkflowReplayEligibility::PROTOCOL_METRIC => WorkerProtocol::requestVersion($request),
+        ];
         $releaseLeasesForRegistration = $this->shouldReleaseLeasesForWorkerRegistration($existing, $processMetrics);
 
         try {
@@ -935,6 +940,10 @@ public function heartbeat(Request $request): JsonResponse
         if (array_key_exists('process_metrics', $validated)) {
             $update['process_metrics'] = $this->normalizeProcessMetrics($validated['process_metrics']);
         }
+        $update['process_metrics'] = [
+            ...(array_key_exists('process_metrics', $update) ? ($update['process_metrics'] ?? []) : ($worker->process_metrics ?? [])),
+            WorkflowReplayEligibility::PROTOCOL_METRIC => WorkerProtocol::requestVersion($request),
+        ];
 
         if (array_key_exists('heartbeat_interval_seconds', $validated)
             && $validated['heartbeat_interval_seconds'] !== null) {

app/Http/Controllers/Api/WorkflowController.php

@@ -1250,6 +1250,7 @@ private function formatRun(WorkflowRun $run, string $namespace, array $descripti
             'compatibility_supported_in_fleet' => $this->compatibilitySupportedInFleet($namespace, $run),
             'compatibility_fleet_reason' => $this->compatibilityFleetReason($namespace, $run),
             'payload_codec' => $run->payload_codec,
+            'definition_routing' => $description['definition_routing'] ?? null,
             'execution_timeout_seconds' => $description['execution_timeout_seconds'] ?? null,
             'run_timeout_seconds' => $runDescription['run_timeout_seconds'] ?? null,
             'execution_deadline_at' => $runDescription['execution_deadline_at'] ?? null,

app/Support/ServerReadiness.php

@@ -40,7 +40,7 @@ public function snapshot(): array
                 : $this->cacheCheck();
 
         $checks = [
-            'database' => $this->databaseCheck(),
+            'database' => $this->databaseCheck(probeQuery: true),
             'migrations' => $this->migrationCheck(),
             'queue' => $queue,
             'default_namespace' => $this->defaultNamespaceCheck(),
@@ -93,22 +93,38 @@ public function bootstrapStatus(): array
         ]));
     }
 
-    private function databaseCheck(): array
+    private function databaseCheck(bool $probeQuery = false): array
     {
+        $queryAttempted = false;
         try {
-            DB::connection()->getPdo();
+            $connection = DB::connection();
+            $connection->getPdo();
+            if ($probeQuery) {
+                $queryAttempted = true;
+                $row = $connection->selectOne('SELECT ? AS readiness_probe', [1], useReadPdo: false);
+                if ((string) ($row->readiness_probe ?? '') !== '1') {
+                    throw new \UnexpectedValueException('The database did not return the readiness probe value.');
+                }
+            }
 
             return [
                 'status' => 'ok',
-                'check_scope' => 'connection_only',
+                'check_scope' => $probeQuery ? 'bound_read_query' : 'connection_only',
                 'write_capacity_verified' => false,
-                'message' => 'The database connection is available. Write capacity has not been verified.',
+                'message' => $probeQuery
+                    ? 'The primary database connection can execute a bound read query. Write capacity has not been verified.'
+                    : 'The database connection is available. Write capacity has not been verified.',
             ];
         } catch (\Throwable $exception) {
+            $reason = $queryAttempted ? 'database_query_failed' : null;
+            if (BackendUnavailable::is($exception)) {
+                $reason = 'backend_unavailable';
+            }
+
             return [
                 'status' => 'unavailable',
                 'message' => $exception->getMessage(),
-                'reason' => BackendUnavailable::is($exception) ? 'backend_unavailable' : null,
+                'reason' => $reason,
             ];
         }
     }