Skip to content

Commit 58105ab

Browse files
michalsnmichalsn
authored
fix: test isolation (#91)
* fix: test isolation for queue backends * fix: clear RabbitMQ queues across handler instances * test: prevent timing race in RabbitMQ delay test
1 parent 343b284 commit 58105ab

6 files changed

Lines changed: 271 additions & 167 deletions

File tree

‎src/Handlers/RabbitMQHandler.php‎

Lines changed: 38 additions & 8 deletions
Original file line numberDiff line numberDiff line change
@@ -27,6 +27,7 @@
2727
use PhpAmqpLib\Connection\AbstractConnection;
2828
use PhpAmqpLib\Connection\AMQPConnectionConfig;
2929
use PhpAmqpLib\Connection\AMQPConnectionFactory;
30+
use PhpAmqpLib\Exception\AMQPProtocolChannelException;
3031
use PhpAmqpLib\Message\AMQPMessage;
3132
use PhpAmqpLib\Wire\AMQPTable;
3233
use Throwable;
@@ -35,8 +36,9 @@ class RabbitMQHandler extends BaseHandler
3536
{
3637
private readonly AbstractConnection $connection;
3738
private readonly AMQPChannel $channel;
38-
private array $declaredQueues = [];
39-
private array $declaredExchanges = [];
39+
private array $declaredLogicalQueues = [];
40+
private array $declaredQueues = [];
41+
private array $declaredExchanges = [];
4042

4143
public function __construct(protected QueueConfig $config)
4244
{
@@ -273,8 +275,12 @@ public function clear(?string $queue = null): bool
273275
{
274276
try {
275277
if ($queue === null) {
276-
// Clear all configured queues
277-
foreach (array_keys($this->config->queuePriorities) as $queueName) {
278+
$queueNames = array_unique([
279+
...array_keys($this->config->queuePriorities),
280+
...array_keys($this->declaredLogicalQueues),
281+
]);
282+
283+
foreach ($queueNames as $queueName) {
278284
$this->clearQueue($queueName);
279285
}
280286
} else {
@@ -300,7 +306,8 @@ public function clear(?string $queue = null): bool
300306
*/
301307
private function declareQueue(string $queue): void
302308
{
303-
$priorities = $this->config->queuePriorities[$queue] ?? ['default'];
309+
$this->declaredLogicalQueues[$queue] = true;
310+
$priorities = $this->config->queuePriorities[$queue] ?? ['default'];
304311

305312
foreach ($priorities as $priority) {
306313
$queueName = $this->getQueueName($queue, $priority);
@@ -459,15 +466,38 @@ private function publishWithOptionalConfirm(AMQPMessage $message, string $exchan
459466
*/
460467
private function clearQueue(string $queue): void
461468
{
469+
// Purge delayed jobs first so they cannot expire into a queue
470+
// that has already been purged.
471+
$this->purgeQueueIfExists($this->getDelayQueueName($queue));
472+
462473
$priorities = $this->config->queuePriorities[$queue] ?? ['default'];
463474

464475
foreach ($priorities as $priority) {
465-
$queueName = $this->getQueueName($queue, $priority);
476+
$this->purgeQueueIfExists($this->getQueueName($queue, $priority));
477+
}
478+
}
479+
480+
/**
481+
* Purge a queue without risking the handler's main channel.
482+
*/
483+
private function purgeQueueIfExists(string $queue): void
484+
{
485+
$channel = $this->connection->channel();
466486

487+
try {
488+
$channel->queue_purge($queue);
489+
} catch (Throwable $e) {
490+
// RabbitMQ closes the channel with 404 when the queue is missing.
491+
if (! $e instanceof AMQPProtocolChannelException || $e->getCode() !== 404) {
492+
throw $e;
493+
}
494+
} finally {
467495
try {
468-
$this->channel->queue_purge($queueName);
496+
if ($channel->is_open()) {
497+
$channel->close();
498+
}
469499
} catch (Throwable) {
470-
// Queue might not exist, ignore
500+
// The broker may already have closed this disposable channel.
471501
}
472502
}
473503
}

‎tests/PushAndPopWithDelayTest.php‎

Lines changed: 40 additions & 36 deletions
Original file line numberDiff line numberDiff line change
@@ -44,43 +44,47 @@ public function testPushAndPopWithDelay(string $name, string $class): void
4444
Time::setTestNow('2023-12-29 14:15:16');
4545

4646
$handler = new $class($this->config);
47-
$result = $handler->setDelay(MINUTE)->push('queue-delay', 'success', ['key1' => 'value1']);
48-
49-
$this->assertNotNull($result);
50-
51-
$result = $handler->push('queue-delay', 'success', ['key2' => 'value2']);
52-
53-
$this->assertNotNull($result);
54-
55-
if ($name === 'database') {
56-
$this->seeInDatabase('queue_jobs', [
57-
'queue' => 'queue-delay',
58-
'payload' => json_encode(['job' => 'success', 'data' => ['key1' => 'value1'], 'metadata' => []]),
59-
'available_at' => 1703859376,
60-
]);
61-
62-
$this->seeInDatabase('queue_jobs', [
63-
'queue' => 'queue-delay',
64-
'payload' => json_encode(['job' => 'success', 'data' => ['key2' => 'value2'], 'metadata' => []]),
65-
'available_at' => 1703859316,
66-
]);
47+
$handler->clear('queue-delay');
48+
49+
try {
50+
$result = $handler->setDelay(MINUTE)->push('queue-delay', 'success', ['key1' => 'value1']);
51+
$this->assertTrue($result->getStatus());
52+
53+
$result = $handler->push('queue-delay', 'success', ['key2' => 'value2']);
54+
$this->assertTrue($result->getStatus());
55+
56+
if ($name === 'database') {
57+
$this->seeInDatabase('queue_jobs', [
58+
'queue' => 'queue-delay',
59+
'payload' => json_encode(['job' => 'success', 'data' => ['key1' => 'value1'], 'metadata' => []]),
60+
'available_at' => 1703859376,
61+
]);
62+
63+
$this->seeInDatabase('queue_jobs', [
64+
'queue' => 'queue-delay',
65+
'payload' => json_encode(['job' => 'success', 'data' => ['key2' => 'value2'], 'metadata' => []]),
66+
'available_at' => 1703859316,
67+
]);
68+
}
69+
70+
$result = $handler->pop('queue-delay', ['default']);
71+
$this->assertInstanceOf(QueueJob::class, $result);
72+
$payload = ['job' => 'success', 'data' => ['key2' => 'value2'], 'metadata' => []];
73+
$this->assertSame($payload, $result->payload);
74+
75+
$result = $handler->pop('queue-delay', ['default']);
76+
$this->assertNull($result);
77+
78+
// add 1 minute
79+
Time::setTestNow('2023-12-29 14:16:16');
80+
81+
$result = $handler->pop('queue-delay', ['default']);
82+
$this->assertInstanceOf(QueueJob::class, $result);
83+
$payload = ['job' => 'success', 'data' => ['key1' => 'value1'], 'metadata' => []];
84+
$this->assertSame($payload, $result->payload);
85+
} finally {
86+
$handler->clear('queue-delay');
6787
}
68-
69-
$result = $handler->pop('queue-delay', ['default']);
70-
$this->assertInstanceOf(QueueJob::class, $result);
71-
$payload = ['job' => 'success', 'data' => ['key2' => 'value2'], 'metadata' => []];
72-
$this->assertSame($payload, $result->payload);
73-
74-
$result = $handler->pop('queue-delay', ['default']);
75-
$this->assertNull($result);
76-
77-
// add 1 minute
78-
Time::setTestNow('2023-12-29 14:16:16');
79-
80-
$result = $handler->pop('queue-delay', ['default']);
81-
$this->assertInstanceOf(QueueJob::class, $result);
82-
$payload = ['job' => 'success', 'data' => ['key1' => 'value1'], 'metadata' => []];
83-
$this->assertSame($payload, $result->payload);
8488
}
8589

8690
public static function providePushAndPopWithDelay(): iterable

‎tests/RabbitMQDelayTest.php‎

Lines changed: 51 additions & 21 deletions
Original file line numberDiff line numberDiff line change
@@ -17,6 +17,7 @@
1717
use CodeIgniter\Queue\Entities\QueueJob;
1818
use CodeIgniter\Queue\Handlers\RabbitMQHandler;
1919
use CodeIgniter\Queue\QueuePushResult;
20+
use CodeIgniter\Test\ReflectionHelper;
2021
use PhpAmqpLib\Connection\AMQPConnectionFactory;
2122
use Tests\Support\Config\Queue as QueueConfig;
2223
use Tests\Support\TestCase;
@@ -29,13 +30,17 @@
2930
*/
3031
final class RabbitMQDelayTest extends TestCase
3132
{
33+
use ReflectionHelper;
34+
3235
private ?RabbitMQHandler $handler = null;
36+
private string $queue;
3337

3438
protected function setUp(): void
3539
{
3640
parent::setUp();
3741

38-
$config = config(QueueConfig::class);
42+
$config = config(QueueConfig::class);
43+
$this->queue = 'delay-test-queue-' . bin2hex(random_bytes(6));
3944

4045
// Skip tests if RabbitMQ is not available
4146
if (! $this->isRabbitMQAvailable()) {
@@ -52,9 +57,8 @@ protected function setUp(): void
5257
protected function tearDown(): void
5358
{
5459
if ($this->handler !== null) {
55-
// Clear test queues
5660
try {
57-
$this->handler->clear('delay-test-queue');
61+
$this->deleteDeclaredResources();
5862
} catch (Throwable) {
5963
// Ignore cleanup errors
6064
}
@@ -70,31 +74,31 @@ public function testDelayedMessageWithRealTiming(): void
7074
$startTime = time();
7175

7276
// Push a delayed job
73-
$result = $this->handler->setDelay($delaySeconds)->push('delay-test-queue', 'success', ['type' => 'delayed']);
77+
$result = $this->handler->setDelay($delaySeconds)->push($this->queue, 'success', ['type' => 'delayed']);
7478
$this->assertInstanceOf(QueuePushResult::class, $result);
7579
$this->assertTrue($result->getStatus());
7680

7781
// Push an immediate job
78-
$result = $this->handler->push('delay-test-queue', 'success', ['type' => 'immediate']);
82+
$result = $this->handler->push($this->queue, 'success', ['type' => 'immediate']);
7983
$this->assertInstanceOf(QueuePushResult::class, $result);
8084
$this->assertTrue($result->getStatus());
8185

8286
// Should get immediate job first
83-
$job = $this->handler->pop('delay-test-queue', ['default']);
87+
$job = $this->handler->pop($this->queue, ['default']);
8488
$this->assertInstanceOf(QueueJob::class, $job);
8589
$this->assertSame('immediate', $job->payload['data']['type']);
8690
$this->handler->done($job);
8791

8892
// Should not get delayed job yet (within first second)
89-
$job = $this->handler->pop('delay-test-queue', ['default']);
93+
$job = $this->handler->pop($this->queue, ['default']);
9094
$this->assertNull($job);
9195

9296
// Wait for delay to expire (with a small buffer)
9397
$waitTime = $delaySeconds + 1;
9498
sleep($waitTime);
9599

96100
// Should now get the delayed job
97-
$job = $this->handler->pop('delay-test-queue', ['default']);
101+
$job = $this->handler->pop($this->queue, ['default']);
98102
$this->assertInstanceOf(QueueJob::class, $job);
99103
$this->assertSame('delayed', $job->payload['data']['type']);
100104

@@ -108,35 +112,32 @@ public function testDelayedMessageWithRealTiming(): void
108112

109113
public function testMultipleDelayedJobsWithDifferentDelays(): void
110114
{
111-
// Push jobs with different delays
112-
$result1 = $this->handler->setDelay(1)->push('delay-test-queue', 'success', ['order' => 'first', 'delay' => 1]);
113-
$result2 = $this->handler->setDelay(3)->push('delay-test-queue', 'success', ['order' => 'second', 'delay' => 3]);
114-
$result3 = $this->handler->push('delay-test-queue', 'success', ['order' => 'immediate', 'delay' => 0]);
115+
// Push the immediate job first so slow test setup cannot allow a
116+
// delayed job to reach the main queue ahead of it.
117+
$result1 = $this->handler->push($this->queue, 'success', ['order' => 'immediate', 'delay' => 0]);
118+
$result2 = $this->handler->setDelay(1)->push($this->queue, 'success', ['order' => 'first', 'delay' => 1]);
119+
$result3 = $this->handler->setDelay(3)->push($this->queue, 'success', ['order' => 'second', 'delay' => 3]);
115120

116121
$this->assertTrue($result1->getStatus());
117122
$this->assertTrue($result2->getStatus());
118123
$this->assertTrue($result3->getStatus());
119124

120125
// Should get immediate job first
121-
$job = $this->handler->pop('delay-test-queue', ['default']);
126+
$job = $this->handler->pop($this->queue, ['default']);
122127
$this->assertInstanceOf(QueueJob::class, $job);
123128
$this->assertSame('immediate', $job->payload['data']['order']);
124129
$this->handler->done($job);
125130

126131
// Wait 2 seconds - should get first delayed job
127132
sleep(2);
128-
$job = $this->handler->pop('delay-test-queue', ['default']);
133+
$job = $this->handler->pop($this->queue, ['default']);
129134
$this->assertInstanceOf(QueueJob::class, $job);
130135
$this->assertSame('first', $job->payload['data']['order']);
131136
$this->handler->done($job);
132137

133-
// Should not get second job yet
134-
$job = $this->handler->pop('delay-test-queue', ['default']);
135-
$this->assertNull($job);
136-
137138
// Wait another 2 seconds - should get second delayed job
138139
sleep(2);
139-
$job = $this->handler->pop('delay-test-queue', ['default']);
140+
$job = $this->handler->pop($this->queue, ['default']);
140141
$this->assertInstanceOf(QueueJob::class, $job);
141142
$this->assertSame('second', $job->payload['data']['order']);
142143
$this->handler->done($job);
@@ -145,22 +146,51 @@ public function testMultipleDelayedJobsWithDifferentDelays(): void
145146
public function testZeroDelayWorksImmediately(): void
146147
{
147148
// Jobs with 0 delay should work immediately
148-
$result = $this->handler->setDelay(0)->push('delay-test-queue', 'success', ['type' => 'zero-delay']);
149+
$result = $this->handler->setDelay(0)->push($this->queue, 'success', ['type' => 'zero-delay']);
149150
$this->assertTrue($result->getStatus());
150151

151152
// Should be able to pop immediately
152-
$job = $this->handler->pop('delay-test-queue', ['default']);
153+
$job = $this->handler->pop($this->queue, ['default']);
153154
$this->assertInstanceOf(QueueJob::class, $job);
154155
$this->assertSame('zero-delay', $job->payload['data']['type']);
155156

156157
$this->handler->done($job);
157158
}
158159

160+
public function testFreshHandlerClearsImmediateAndDelayedJobs(): void
161+
{
162+
$delayedResult = $this->handler->setDelay(1)->push($this->queue, 'success', ['type' => 'delayed']);
163+
$this->assertTrue($delayedResult->getStatus());
164+
165+
$immediateResult = $this->handler->push($this->queue, 'success', ['type' => 'immediate']);
166+
$this->assertTrue($immediateResult->getStatus());
167+
168+
$clearer = new RabbitMQHandler(config(QueueConfig::class));
169+
$this->assertTrue($clearer->clear($this->queue));
170+
171+
sleep(2);
172+
173+
$this->assertNull($clearer->pop($this->queue, ['default']));
174+
}
175+
159176
/**
160177
* Check if RabbitMQ is available for testing.
161178
*/
162179
private function isRabbitMQAvailable(): bool
163180
{
164181
return class_exists(AMQPConnectionFactory::class);
165182
}
183+
184+
private function deleteDeclaredResources(): void
185+
{
186+
$channel = self::getPrivateProperty($this->handler, 'channel');
187+
188+
foreach (array_keys(self::getPrivateProperty($this->handler, 'declaredQueues')) as $queue) {
189+
$channel->queue_delete($queue);
190+
}
191+
192+
foreach (array_keys(self::getPrivateProperty($this->handler, 'declaredExchanges')) as $exchange) {
193+
$channel->exchange_delete($exchange);
194+
}
195+
}
166196
}

0 commit comments

Comments
 (0)