Align server repair passes with matching-role scopes

durable-workflow-ops · durable-workflow-ops · commit cd14691786bb · 2026-04-28T21:19:14.000Z
diff --git a/README.md b/README.md
@@ -506,6 +506,11 @@ workflow-task command payload.
 - `GET /api/system/activity-timeouts` — Expired activity execution diagnostics
 - `POST /api/system/activity-timeouts/pass` — Enforce activity timeouts
 
+`POST /api/system/repair/pass` accepts optional `connection`, `queue`,
+`run_ids`, and `instance_id` filters. Set `respect_throttle=true` when a
+dedicated matching-role loop should skip a pass rather than duplicate work
+already covered by another matching-role process holding the repair throttle.
+
 ### Namespaces
 - `GET /api/namespaces` — List namespaces
 - `POST /api/namespaces` — Create namespace
diff --git a/app/Http/Controllers/Api/SystemController.php b/app/Http/Controllers/Api/SystemController.php
@@ -9,17 +9,18 @@
 use App\Support\WorkflowTaskFailureMetrics;
 use Illuminate\Http\JsonResponse;
 use Illuminate\Http\Request;
+use Workflow\V2\Contracts\MatchingRole;
 use Workflow\V2\Support\ActivityTimeoutEnforcer;
 use Workflow\V2\Support\HealthCheck;
 use Workflow\V2\Support\OperatorMetrics;
 use Workflow\V2\Support\TaskRepairCandidates;
 use Workflow\V2\Support\TaskRepairPolicy;
-use Workflow\V2\TaskWatchdog;
 
 class SystemController
 {
     public function __construct(
         private readonly TaskQueueBuildIdRolloutSnapshot $buildIdRollouts,
+        private readonly MatchingRole $matchingRole,
     ) {}
 
     public function repairPass(Request $request): JsonResponse
@@ -32,22 +33,34 @@ public function repairPass(Request $request): JsonResponse
             'run_ids' => ['nullable', 'array', 'max:100'],
             'run_ids.*' => ['string', 'min:1', 'max:128'],
             'instance_id' => ['nullable', 'string', 'min:1', 'max:128'],
+            'connection' => ['nullable', 'string', 'max:128'],
+            'queue' => ['nullable', 'string', 'max:128'],
+            'respect_throttle' => ['nullable', 'boolean'],
         ]);
 
         $runIds = array_values(array_map(
             static fn (string $v): string => trim($v),
             $validated['run_ids'] ?? [],
         ));
 
-        $instanceId = isset($validated['instance_id']) && is_string($validated['instance_id'])
-            ? trim($validated['instance_id'])
-            : null;
+        $instanceId = $this->trimmedString($validated['instance_id'] ?? null);
+        $connection = $this->trimmedString($validated['connection'] ?? null);
+        $queue = $this->trimmedString($validated['queue'] ?? null);
+        $respectThrottle = (bool) ($validated['respect_throttle'] ?? false);
 
-        $report = TaskWatchdog::runPass(
+        $report = $this->matchingRole->runPass(
+            connection: $connection,
+            queue: $queue,
+            respectThrottle: $respectThrottle,
             runIds: $runIds,
             instanceId: $instanceId,
         );
         $report = array_replace([
+            'connection' => $connection,
+            'queue' => $queue,
+            'run_ids' => $runIds,
+            'instance_id' => $instanceId,
+            'respect_throttle' => $respectThrottle,
             'selected_command_contract_candidates' => 0,
             'backfilled_command_contracts' => 0,
             'command_contract_backfill_unavailable' => 0,
@@ -281,4 +294,15 @@ public function retentionEnforcePass(Request $request): JsonResponse
 
         return ControlPlaneProtocol::json($report, $hasFailures ? 207 : 200);
     }
+
+    private function trimmedString(mixed $value): ?string
+    {
+        if (! is_string($value)) {
+            return null;
+        }
+
+        $value = trim($value);
+
+        return $value === '' ? null : $value;
+    }
 }
diff --git a/app/Support/WorkflowPackageApiFloor.php b/app/Support/WorkflowPackageApiFloor.php
@@ -7,8 +7,10 @@
 use ReflectionMethod;
 use RuntimeException;
 use Workflow\Serializers\CodecRegistry;
+use Workflow\V2\Contracts\MatchingRole;
 use Workflow\V2\Support\BackendCapabilities;
 use Workflow\V2\Support\ChildWorkflowNamespaceProjection;
+use Workflow\V2\Support\DefaultMatchingRole;
 use Workflow\V2\Support\MatchingRoleSnapshot;
 
 /**
@@ -55,6 +57,19 @@ final class WorkflowPackageApiFloor
         // local WorkflowLink / WorkflowRunLineageEntry observer glue.
         [ChildWorkflowNamespaceProjection::class, 'projectLink'],
         [ChildWorkflowNamespaceProjection::class, 'projectLineageEntry'],
+        // Server control-plane repair passes resolve through the package's
+        // dedicated matching-role implementation instead of hard-coding the
+        // in-process watchdog.
+        [DefaultMatchingRole::class, 'wake'],
+        [DefaultMatchingRole::class, 'runPass'],
+    ];
+
+    /**
+     * Interface methods the server type-hints directly.
+     */
+    private const REQUIRED_INTERFACE_APIS = [
+        [MatchingRole::class, 'wake'],
+        [MatchingRole::class, 'runPass'],
     ];
 
     /**
@@ -96,6 +111,12 @@ public static function assert(): void
             }
         }
 
+        foreach (self::REQUIRED_INTERFACE_APIS as [$class, $method]) {
+            if (! self::hasInterfaceMethod($class, $method)) {
+                $missing[] = sprintf('%s::%s()', $class, $method);
+            }
+        }
+
         if (! class_exists(self::POLL_MODE_DEMOTION_CLASS)) {
             $missing[] = self::POLL_MODE_DEMOTION_CLASS;
         } elseif (! self::confirmsPollModeDemotion(self::POLL_MODE_DEMOTION_CLASS, self::POLL_MODE_DEMOTION_METHOD)) {
@@ -114,8 +135,9 @@ public static function assert(): void
             "Installed durable-workflow/workflow package is older than the server's API floor. "
             .'Missing: %s. Re-run `composer update durable-workflow/workflow` against a v2 snapshot that '
             .'includes CodecRegistry::universal(), CodecRegistry::engineSpecific(), MatchingRoleSnapshot::current(), '
-            .'and the poll-mode queue capability demotion, plus ChildWorkflowNamespaceProjection for package-owned '
-            .'child namespace propagation (see repos/workflow commits 8e132d0, cfd8e95, and f666b25, or newer).',
+            .'the poll-mode queue capability demotion, and the matching-role repair-pass contract, plus '
+            .'ChildWorkflowNamespaceProjection for package-owned child namespace propagation '
+            .'(see repos/workflow commits 8e132d0, cfd8e95, and f666b25, or newer).',
             implode(', ', $missing),
         ));
     }
@@ -152,6 +174,22 @@ private static function hasInstanceMethod(string $class, string $method): bool
         return $methodReflection->isPublic() && ! $methodReflection->isStatic();
     }
 
+    private static function hasInterfaceMethod(string $class, string $method): bool
+    {
+        if (! interface_exists($class)) {
+            return false;
+        }
+
+        try {
+            $reflection = new ReflectionClass($class);
+            $methodReflection = $reflection->getMethod($method);
+        } catch (ReflectionException) {
+            return false;
+        }
+
+        return $methodReflection->isPublic() && ! $methodReflection->isStatic();
+    }
+
     /**
      * Prove the installed BackendCapabilities::queue() contains the
      * poll-mode demotion logic from workflow@f666b25.
diff --git a/tests/Feature/ControlPlaneOperationalSuccessContractTest.php b/tests/Feature/ControlPlaneOperationalSuccessContractTest.php
@@ -237,6 +237,11 @@ public static function operationalSuccessProvider(): array
                 'path' => '/api/system/repair/pass',
                 'body' => [],
                 'structure' => [
+                    'connection',
+                    'queue',
+                    'run_ids',
+                    'instance_id',
+                    'respect_throttle',
                     'throttled',
                     'selected_existing_task_candidates',
                     'selected_missing_task_candidates',
diff --git a/tests/Feature/ControlPlaneValidationContractTest.php b/tests/Feature/ControlPlaneValidationContractTest.php
@@ -84,6 +84,13 @@ public static function validationEndpointProvider(): array
                 'errorField' => 'run_ids',
                 'controlOperation' => null,
             ],
+            'system.repair_pass_respect_throttle' => [
+                'method' => 'post',
+                'path' => '/api/system/repair/pass',
+                'body' => ['respect_throttle' => 'not-a-bool'],
+                'errorField' => 'respect_throttle',
+                'controlOperation' => null,
+            ],
             'system.activity_timeout_status' => [
                 'method' => 'get',
                 'path' => '/api/system/activity-timeouts?limit=0',
diff --git a/tests/Feature/TransportRepairTest.php b/tests/Feature/TransportRepairTest.php
@@ -8,6 +8,7 @@
 use Tests\Feature\Concerns\ServerTestHelpers;
 use Tests\Fixtures\ExternalGreetingWorkflow;
 use Tests\TestCase;
+use Workflow\V2\Contracts\MatchingRole;
 use Workflow\V2\Enums\TaskStatus;
 use Workflow\V2\Models\WorkflowTask;
 
@@ -113,6 +114,11 @@ public function test_system_repair_pass_returns_report(): void
             ->postJson('/api/system/repair/pass')
             ->assertOk()
             ->assertJsonStructure([
+                'connection',
+                'queue',
+                'run_ids',
+                'instance_id',
+                'respect_throttle',
                 'throttled',
                 'selected_existing_task_candidates',
                 'selected_missing_task_candidates',
@@ -135,6 +141,10 @@ public function test_system_repair_pass_with_empty_system_returns_zero_repairs()
             ->postJson('/api/system/repair/pass');
 
         $response->assertOk()
+            ->assertJsonPath('connection', null)
+            ->assertJsonPath('queue', null)
+            ->assertJsonPath('instance_id', null)
+            ->assertJsonPath('respect_throttle', false)
             ->assertJsonPath('throttled', false)
             ->assertJsonPath('selected_total_candidates', 0)
             ->assertJsonPath('repaired_existing_tasks', 0)
@@ -144,6 +154,7 @@ public function test_system_repair_pass_with_empty_system_returns_zero_repairs()
             ->assertJsonPath('backfilled_command_contracts', 0)
             ->assertJsonPath('command_contract_backfill_unavailable', 0);
 
+        $this->assertSame([], $response->json('run_ids'));
         $this->assertSame([], $response->json('existing_task_failures'));
         $this->assertSame([], $response->json('missing_run_failures'));
         $this->assertSame([], $response->json('command_contract_failures'));
@@ -156,6 +167,7 @@ public function test_system_repair_pass_accepts_run_id_filter(): void
                 'run_ids' => ['non-existent-run-id'],
             ])
             ->assertOk()
+            ->assertJsonPath('run_ids.0', 'non-existent-run-id')
             ->assertJsonPath('selected_total_candidates', 0);
     }
 
@@ -166,9 +178,93 @@ public function test_system_repair_pass_accepts_instance_id_filter(): void
                 'instance_id' => 'non-existent-instance',
             ])
             ->assertOk()
+            ->assertJsonPath('instance_id', 'non-existent-instance')
             ->assertJsonPath('selected_total_candidates', 0);
     }
 
+    public function test_system_repair_pass_uses_matching_role_binding_and_forwards_scope_options(): void
+    {
+        $fake = new class implements MatchingRole
+        {
+            /**
+             * @var array{connection: string|null, queue: string|null, respectThrottle: bool, runIds: list<string>, instanceId: string|null}|null
+             */
+            public ?array $lastRunPassArguments = null;
+
+            public function wake(?string $connection = null, ?string $queue = null): void {}
+
+            public function runPass(
+                ?string $connection = null,
+                ?string $queue = null,
+                bool $respectThrottle = false,
+                array $runIds = [],
+                ?string $instanceId = null,
+            ): array {
+                $this->lastRunPassArguments = [
+                    'connection' => $connection,
+                    'queue' => $queue,
+                    'respectThrottle' => $respectThrottle,
+                    'runIds' => $runIds,
+                    'instanceId' => $instanceId,
+                ];
+
+                return [
+                    'connection' => $connection,
+                    'queue' => $queue,
+                    'run_ids' => $runIds,
+                    'instance_id' => $instanceId,
+                    'respect_throttle' => $respectThrottle,
+                    'throttled' => false,
+                    'selected_existing_task_candidates' => 2,
+                    'selected_missing_task_candidates' => 1,
+                    'selected_total_candidates' => 3,
+                    'repaired_existing_tasks' => 2,
+                    'repaired_missing_tasks' => 1,
+                    'dispatched_tasks' => 3,
+                    'existing_task_failures' => [],
+                    'missing_run_failures' => [],
+                    'deadline_expired_candidates' => 0,
+                    'deadline_expired_tasks_created' => 0,
+                    'deadline_expired_failures' => [],
+                    'activity_timeout_candidates' => 0,
+                    'activity_timeouts_enforced' => 0,
+                    'activity_timeout_failures' => [],
+                ];
+            }
+        };
+
+        $this->app->instance(MatchingRole::class, $fake);
+
+        $response = $this->withHeaders($this->apiHeaders())
+            ->postJson('/api/system/repair/pass', [
+                'connection' => ' redis ',
+                'queue' => ' critical ',
+                'run_ids' => [' run-a ', 'run-b'],
+                'instance_id' => ' instance-42 ',
+                'respect_throttle' => true,
+            ]);
+
+        $response->assertOk()
+            ->assertJsonPath('connection', 'redis')
+            ->assertJsonPath('queue', 'critical')
+            ->assertJsonPath('run_ids.0', 'run-a')
+            ->assertJsonPath('run_ids.1', 'run-b')
+            ->assertJsonPath('instance_id', 'instance-42')
+            ->assertJsonPath('respect_throttle', true)
+            ->assertJsonPath('selected_total_candidates', 3)
+            ->assertJsonPath('repaired_existing_tasks', 2)
+            ->assertJsonPath('repaired_missing_tasks', 1)
+            ->assertJsonPath('dispatched_tasks', 3);
+
+        $this->assertSame([
+            'connection' => 'redis',
+            'queue' => 'critical',
+            'respectThrottle' => true,
+            'runIds' => ['run-a', 'run-b'],
+            'instanceId' => 'instance-42',
+        ], $fake->lastRunPassArguments);
+    }
+
     public function test_system_repair_pass_recovers_expired_poll_mode_workflow_task_leases(): void
     {
         config(['server.polling.timeout' => 0]);
diff --git a/tests/Unit/Support/WorkflowPackageApiFloorTest.php b/tests/Unit/Support/WorkflowPackageApiFloorTest.php
@@ -8,8 +8,10 @@
 use ReflectionMethod;
 use Tests\Fixtures\StaleBackendCapabilities;
 use Workflow\Serializers\CodecRegistry;
+use Workflow\V2\Contracts\MatchingRole;
 use Workflow\V2\Support\BackendCapabilities;
 use Workflow\V2\Support\ChildWorkflowNamespaceProjection;
+use Workflow\V2\Support\DefaultMatchingRole;
 use Workflow\V2\Support\MatchingRoleSnapshot;
 
 /**
@@ -71,6 +73,30 @@ public function test_child_workflow_namespace_projection_is_public_instance_api(
         }
     }
 
+    public function test_matching_role_contract_is_public_instance_api(): void
+    {
+        $reflection = new ReflectionClass(MatchingRole::class);
+
+        foreach (['wake', 'runPass'] as $methodName) {
+            $method = $reflection->getMethod($methodName);
+
+            $this->assertTrue($method->isPublic());
+            $this->assertFalse($method->isStatic());
+        }
+    }
+
+    public function test_default_matching_role_exposes_public_instance_repair_methods(): void
+    {
+        $reflection = new ReflectionClass(DefaultMatchingRole::class);
+
+        foreach (['wake', 'runPass'] as $methodName) {
+            $method = $reflection->getMethod($methodName);
+
+            $this->assertTrue($method->isPublic());
+            $this->assertFalse($method->isStatic());
+        }
+    }
+
     public function test_poll_mode_demotion_check_accepts_current_workflow_package(): void
     {
         $confirms = $this->invokeConfirmsPollModeDemotion(BackendCapabilities::class, 'queue');
