Skip to content

Commit 82a244c

Browse files
rmcdanielrmcdaniel
authored
Repair inherited default queue connections (#752)
* Include inherited connections in default worker repair scopes * Make repair fixture overdue in availability and dispatch * Format overdue timestamps with the repository style * Exercise real default worker recovery and prepare Workflow 2.5.9 * Refresh persisted state while observing default worker recovery * Assert the typed activity completion status --------- Co-authored-by: Durable Workflow <support@durable-workflow.com>
1 parent 58f387d commit 82a244c

6 files changed

Lines changed: 216 additions & 1 deletion

File tree

‎composer.json‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -79,7 +79,7 @@
7979
"dev-main": "2.0.x-dev"
8080
},
8181
"durable-workflow": {
82-
"product-train": "2.5.8",
82+
"product-train": "2.5.9",
8383
"laravel-embedded-upgrade-contract": "resources/laravel-embedded-upgrade-contract.json",
8484
"laravel-dependency-security-policy": "resources/laravel-dependency-security-policy.json"
8585
},

‎src/V2/Support/RequestedTaskScope.php‎

Lines changed: 6 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -20,6 +20,12 @@ private static function applyValue($query, string $column, ?string $value): void
2020
return;
2121
}
2222

23+
// Tasks that inherit the default connection store null instead of the
24+
// resolved driver name supplied by a queue worker's Looping event.
25+
if ($column === 'connection' && in_array(config('queue.default'), $values, true)) {
26+
$values[] = 'default';
27+
}
28+
2329
$query->where(static function ($requestedScope) use ($column, $values): void {
2430
$nonDefault = array_values(array_filter(
2531
$values,

‎tests/Feature/V2/V2DefaultConnectionRecoveryTest.php‎

Lines changed: 64 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,64 @@
1+
<?php
2+
3+
declare(strict_types=1);
4+
5+
namespace Tests\Feature\V2;
6+
7+
use Illuminate\Support\Facades\Queue;
8+
use Tests\Fixtures\V2\TestGreetingWorkflow;
9+
use Tests\TestCase;
10+
use Workflow\V2\Enums\ActivityStatus;
11+
use Workflow\V2\Enums\HistoryEventType;
12+
use Workflow\V2\Enums\TaskStatus;
13+
use Workflow\V2\Models\ActivityExecution;
14+
use Workflow\V2\Models\WorkflowHistoryEvent;
15+
use Workflow\V2\Models\WorkflowTask;
16+
use Workflow\V2\WorkflowStub;
17+
18+
final class V2DefaultConnectionRecoveryTest extends TestCase
19+
{
20+
public function testRealDefaultWorkersRecoverALostEnvelopeWithoutRepeatingActivity(): void
21+
{
22+
$this->assertSame('redis', config('queue.default'));
23+
$queue = $this->app->make('queue');
24+
// Drop only the initial queue envelope. Separate workers continue to
25+
// use Redis, and the durable task remains committed in the database.
26+
Queue::fake();
27+
$workflow = WorkflowStub::make(TestGreetingWorkflow::class, 'default-connection-recovery');
28+
$workflow->start('recovery');
29+
$runId = $workflow->runId();
30+
$task = WorkflowTask::query()->where('workflow_run_id', $runId)->firstOrFail();
31+
$this->assertNull($task->connection);
32+
$task->forceFill([
33+
'available_at' => now()
34+
->subMinute(),
35+
'last_dispatched_at' => now()
36+
->subMinute(),
37+
])->save();
38+
Queue::swap($queue);
39+
$this->enableAutomaticTaskRepair();
40+
41+
$this->waitForWorkflow($workflow, static fn (WorkflowStub $run): bool => ! $run->refresh()->running());
42+
43+
$this->assertTrue($workflow->refresh()->completed());
44+
$this->assertSame($runId, $workflow->runId());
45+
$this->assertSame([
46+
'greeting' => 'Hello, recovery!',
47+
'workflow_id' => 'default-connection-recovery',
48+
'run_id' => $runId,
49+
], $workflow->output());
50+
$task->refresh();
51+
$this->assertSame(TaskStatus::Completed, $task->status);
52+
$this->assertSame(1, $task->attempt_count);
53+
$this->assertGreaterThanOrEqual(1, $task->repair_count);
54+
$activities = ActivityExecution::query()->where('workflow_run_id', $runId)->get();
55+
$this->assertCount(1, $activities);
56+
$this->assertSame(ActivityStatus::Completed, $activities->first()->status);
57+
$this->assertSame(1, $activities->first()->attempt_count);
58+
foreach ([HistoryEventType::ActivityCompleted, HistoryEventType::WorkflowCompleted] as $eventType) {
59+
$this->assertSame(1, WorkflowHistoryEvent::query()->where('workflow_run_id', $runId)
60+
->where('event_type', $eventType->value)
61+
->count());
62+
}
63+
}
64+
}

‎tests/Feature/V2/V2EmbeddedParallelChildConcurrencyTest.php‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -29,6 +29,7 @@ final class V2EmbeddedParallelChildConcurrencyTest extends TestCase
2929
{
3030
public function testThreeFourChildParentsCompleteWithConcurrentQueueWorkers(): void
3131
{
32+
$this->enableAutomaticTaskRepair();
3233
$workflows = [];
3334
for ($index = 0; $index < 3; $index++) {
3435
$workflow = WorkflowStub::make(TestParallelGreetingChildrenWorkflow::class, 'parallel-parent-' . $index);

‎tests/TestCase.php‎

Lines changed: 22 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -326,6 +326,28 @@ protected static function restartQueueWorkers(): void
326326
self::startWorkers();
327327
}
328328

329+
protected function enableAutomaticTaskRepair(): void
330+
{
331+
Cache::forget(TaskWatchdog::LOOP_THROTTLE_KEY);
332+
$redis = self::redisConnection();
333+
334+
if (! $redis instanceof \Redis) {
335+
throw new RuntimeException('Automatic task repair tests require a Redis connection.');
336+
}
337+
338+
try {
339+
$redis->select(self::redisCacheDatabase());
340+
341+
// Remove every cache prefix primed for the separate test workers.
342+
// Their subsequent Looping events use the normal repair throttle.
343+
foreach (self::cacheKeyCandidates(TaskWatchdog::LOOP_THROTTLE_KEY) as $key) {
344+
$redis->del($key);
345+
}
346+
} finally {
347+
$redis->close();
348+
}
349+
}
350+
329351
/**
330352
* @return array<int, array{
331353
* worker: int,

‎tests/Unit/V2/RequestedTaskScopeTest.php‎

Lines changed: 122 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,122 @@
1+
<?php
2+
3+
declare(strict_types=1);
4+
5+
namespace Tests\Unit\V2;
6+
7+
use Illuminate\Support\Facades\Queue;
8+
use PHPUnit\Framework\Attributes\DataProvider;
9+
use Tests\Fixtures\V2\TestGreetingWorkflow;
10+
use Tests\TestCase;
11+
use Workflow\V2\Jobs\RunWorkflowTask;
12+
use Workflow\V2\Models\WorkflowTask;
13+
use Workflow\V2\Support\RequestedTaskScope;
14+
use Workflow\V2\TaskWatchdog;
15+
use Workflow\V2\WorkflowStub;
16+
use Workflow\WorkflowOptions;
17+
18+
final class RequestedTaskScopeTest extends TestCase
19+
{
20+
/**
21+
* @param list<string> $expected
22+
*/
23+
#[DataProvider('connectionScopes')]
24+
public function testEffectiveDefaultConnectionIncludesInheritedTasksWithoutCrossingOtherScopes(
25+
string $defaultConnection,
26+
?string $requestedConnection,
27+
array $expected,
28+
): void {
29+
Queue::fake();
30+
config([
31+
'queue.default' => $defaultConnection,
32+
]);
33+
$rows = [
34+
'implicit' => [null, 'default'],
35+
'empty' => ['', 'default'],
36+
'symbolic' => ['default', 'default'],
37+
'redis' => ['redis', 'default'],
38+
'database' => ['database', 'default'],
39+
'sqs' => ['sqs', 'default'],
40+
'other-queue' => [null, 'other-queue'],
41+
];
42+
$labels = [];
43+
foreach ($rows as $label => [$connection, $queue]) {
44+
$task = $this->createTask($label, $connection, $queue);
45+
$labels[$task->id] = $label;
46+
}
47+
48+
$query = WorkflowTask::query();
49+
RequestedTaskScope::apply($query, $requestedConnection, 'default');
50+
$actual = $query->pluck('id')
51+
->map(static fn (string $id): string => $labels[$id])->all();
52+
53+
$this->assertEqualsCanonicalizing($expected, $actual);
54+
}
55+
56+
public static function connectionScopes(): array
57+
{
58+
return [
59+
'resolved default Redis' => ['redis', 'redis', ['implicit', 'empty', 'symbolic', 'redis']],
60+
'resolved default database' => ['database', 'database', ['implicit', 'empty', 'symbolic', 'database']],
61+
'explicit nondefault database' => ['redis', 'database', ['database']],
62+
'explicit nondefault Redis' => ['database', 'redis', ['redis']],
63+
'comma separated includes resolved default' => [
64+
'redis',
65+
'redis,sqs',
66+
['implicit', 'empty', 'symbolic', 'redis', 'sqs'],
67+
],
68+
'comma separated excludes resolved default' => ['database', 'redis,sqs', ['redis', 'sqs']],
69+
'existing symbolic default' => ['redis', 'default', ['implicit', 'empty', 'symbolic']],
70+
'unscoped connection' => ['redis', null, ['implicit', 'empty', 'symbolic', 'redis', 'database', 'sqs']],
71+
];
72+
}
73+
74+
public function testDefaultWorkerRepairRedispatchesInheritedTaskAndPreservesOtherConnection(): void
75+
{
76+
Queue::fake();
77+
config([
78+
'queue.default' => 'redis',
79+
]);
80+
$implicit = $this->createTask('implicit-repair', null);
81+
$other = $this->createTask('other-repair', 'database');
82+
foreach ([$implicit, $other] as $task) {
83+
$task->forceFill([
84+
'available_at' => now()
85+
->subMinute(),
86+
'created_at' => now()
87+
->subMinute(),
88+
'last_dispatched_at' => now()
89+
->subMinute(),
90+
])->save();
91+
}
92+
Queue::fake();
93+
94+
$report = TaskWatchdog::runPass(connection: 'redis', queue: 'default');
95+
96+
$this->assertSame(1, $report['selected_existing_task_candidates']);
97+
$this->assertSame(1, $report['dispatched_tasks']);
98+
$this->assertSame([], $report['existing_task_failures']);
99+
$this->assertSame(1, $implicit->fresh()->repair_count);
100+
$this->assertSame(0, $other->fresh()->repair_count);
101+
Queue::assertPushed(
102+
RunWorkflowTask::class,
103+
static fn (RunWorkflowTask $job): bool => $job->taskId === $implicit->id
104+
);
105+
Queue::assertPushed(RunWorkflowTask::class, 1);
106+
}
107+
108+
private function createTask(string $label, ?string $connection, string $queue = 'default'): WorkflowTask
109+
{
110+
$workflow = WorkflowStub::make(TestGreetingWorkflow::class, 'scope-' . $label);
111+
$workflow->start('scope', new WorkflowOptions(queue: $queue));
112+
113+
$task = WorkflowTask::query()->where('workflow_run_id', $workflow->runId())->firstOrFail();
114+
// Include legacy stored representations without dispatching an
115+
// envelope through an unconfigured connection during fixture setup.
116+
$task->forceFill([
117+
'connection' => $connection,
118+
])->save();
119+
120+
return $task;
121+
}
122+
}

0 commit comments

Comments
 (0)