tests/phpunit/includes/jobqueue/JobQueueTest.php

   1 <?php
   2
   3 use MediaWiki\MediaWikiServices;
   4 use MediaWiki\WikiMap\WikiMap;
   5 use Wikimedia\ObjectCache\HashBagOStuff;
   6 use Wikimedia\ObjectCache\WANObjectCache;
   7
   8 /**
   9  * @group JobQueue
  10  * @group medium
  11  * @group Database
  12  * @covers \JobQueue
  13  */
  14 class JobQueueTest extends MediaWikiIntegrationTestCase {
  15         protected ?JobQueue $queueRand;
  16         protected ?JobQueue $queueRandTTL;
  17         protected ?JobQueue $queueTimestamp;
  18         protected ?JobQueue $queueTimestampTTL;
  19         protected ?JobQueue $queueFifo;
  20         protected ?JobQueue $queueFifoTTL;
  21
  22         protected function setUp(): void {
  23                 global $wgJobTypeConf;
  24                 parent::setUp();
  25
  26                 $services = $this->getServiceContainer();
  27                 if ( $this->getCliArg( 'use-jobqueue' ) ) {
  28                         $name = $this->getCliArg( 'use-jobqueue' );
  29                         if ( !isset( $wgJobTypeConf[$name] ) ) {
  30                                 throw new RuntimeException( "No \$wgJobTypeConf entry for '$name'." );
  31                         }
  32                         $baseConfig = $wgJobTypeConf[$name];
  33                 } else {
  34                         $baseConfig = [ 'class' => JobQueueDBSingle::class ];
  35                 }
  36                 $baseConfig['type'] = 'null';
  37                 $baseConfig['domain'] = WikiMap::getCurrentWikiDbDomain()->getId();
  38                 $baseConfig['stash'] = new HashBagOStuff();
  39                 $baseConfig['wanCache'] = new WANObjectCache( [ 'cache' => new HashBagOStuff() ] );
  40                 $baseConfig['idGenerator'] = $services->getGlobalIdGenerator();
  41                 $variants = [
  42                         'queueRand' => [ 'order' => 'random', 'claimTTL' => 0 ],
  43                         'queueRandTTL' => [ 'order' => 'random', 'claimTTL' => 10 ],
  44                         'queueTimestamp' => [ 'order' => 'timestamp', 'claimTTL' => 0 ],
  45                         'queueTimestampTTL' => [ 'order' => 'timestamp', 'claimTTL' => 10 ],
  46                         'queueFifo' => [ 'order' => 'fifo', 'claimTTL' => 0 ],
  47                         'queueFifoTTL' => [ 'order' => 'fifo', 'claimTTL' => 10 ],
  48                 ];
  49                 foreach ( $variants as $q => $settings ) {
  50                         $this->$q = JobQueue::factory( $settings + $baseConfig );
  51                 }
  52         }
  53
  54         protected function tearDown(): void {
  55                 foreach (
  56                         [
  57                                 'queueRand', 'queueRandTTL', 'queueTimestamp', 'queueTimestampTTL',
  58                                 'queueFifo', 'queueFifoTTL'
  59                         ] as $q
  60                 ) {
  61                         if ( $this->$q ) {
  62                                 $this->$q->delete();
  63                         }
  64                         $this->$q = null;
  65                 }
  66                 parent::tearDown();
  67         }
  68
  69         /**
  70          * @dataProvider provider_queueLists
  71          */
  72         public function testGetType( $queue, $recycles, $desc ) {
  73                 $queue = $this->$queue;
  74                 if ( !$queue ) {
  75                         $this->markTestSkipped( $desc );
  76                 }
  77                 $this->assertEquals( 'null', $queue->getType(), "Proper job type ($desc)" );
  78         }
  79
  80         /**
  81          * @dataProvider provider_queueLists
  82          */
  83         public function testBasicOperations( $queue, $recycles, $desc ) {
  84                 $queue = $this->$queue;
  85                 if ( !$queue ) {
  86                         $this->markTestSkipped( $desc );
  87                 }
  88
  89                 $this->assertTrue( $queue->isEmpty(), "Queue is empty ($desc)" );
  90
  91                 $queue->flushCaches();
  92                 $this->assertSame( 0, $queue->getSize(), "Queue is empty ($desc)" );
  93                 $this->assertSame( 0, $queue->getAcquiredCount(), "Queue is empty ($desc)" );
  94
  95                 $this->assertNull( $queue->push( $this->newJob() ), "Push worked ($desc)" );
  96                 $this->assertNull( $queue->batchPush( [ $this->newJob() ] ), "Push worked ($desc)" );
  97
  98                 $this->assertFalse( $queue->isEmpty(), "Queue is not empty ($desc)" );
  99
 100                 $queue->flushCaches();
 101                 $this->assertEquals( 2, $queue->getSize(), "Queue size is correct ($desc)" );
 102                 $this->assertSame( 0, $queue->getAcquiredCount(), "No jobs active ($desc)" );
 103                 $jobs = iterator_to_array( $queue->getAllQueuedJobs() );
 104                 $this->assertCount( 2, $jobs, "Queue iterator size is correct ($desc)" );
 105
 106                 $job1 = $queue->pop();
 107                 $this->assertFalse( $queue->isEmpty(), "Queue is not empty ($desc)" );
 108
 109                 $queue->flushCaches();
 110                 $this->assertSame( 1, $queue->getSize(), "Queue size is correct ($desc)" );
 111
 112                 $queue->flushCaches();
 113                 if ( $recycles ) {
 114                         $this->assertSame( 1, $queue->getAcquiredCount(), "Active job count ($desc)" );
 115                 }
 116
 117                 $job2 = $queue->pop();
 118                 $this->assertTrue( $queue->isEmpty(), "Queue is empty ($desc)" );
 119                 $this->assertSame( 0, $queue->getSize(), "Queue is empty ($desc)" );
 120
 121                 $queue->flushCaches();
 122                 if ( $recycles ) {
 123                         $this->assertEquals( 2, $queue->getAcquiredCount(), "Active job count ($desc)" );
 124                 }
 125
 126                 $queue->ack( $job1 );
 127
 128                 $queue->flushCaches();
 129                 if ( $recycles ) {
 130                         $this->assertSame( 1, $queue->getAcquiredCount(), "Active job count ($desc)" );
 131                 }
 132
 133                 $queue->ack( $job2 );
 134
 135                 $queue->flushCaches();
 136                 $this->assertSame( 0, $queue->getAcquiredCount(), "Active job count ($desc)" );
 137
 138                 $this->assertNull( $queue->batchPush( [ $this->newJob(), $this->newJob() ] ),
 139                         "Push worked ($desc)" );
 140                 $this->assertFalse( $queue->isEmpty(), "Queue is not empty ($desc)" );
 141
 142                 $queue->delete();
 143                 $queue->flushCaches();
 144                 $this->assertTrue( $queue->isEmpty(), "Queue is empty ($desc)" );
 145                 $this->assertSame( 0, $queue->getSize(), "Queue is empty ($desc)" );
 146         }
 147
 148         /**
 149          * @dataProvider provider_queueLists
 150          */
 151         public function testBasicDeduplication( $queue, $recycles, $desc ) {
 152                 $queue = $this->$queue;
 153                 if ( !$queue ) {
 154                         $this->markTestSkipped( $desc );
 155                 }
 156
 157                 $this->assertTrue( $queue->isEmpty(), "Queue is empty ($desc)" );
 158
 159                 $queue->flushCaches();
 160                 $this->assertSame( 0, $queue->getSize(), "Queue is empty ($desc)" );
 161                 $this->assertSame( 0, $queue->getAcquiredCount(), "Queue is empty ($desc)" );
 162
 163                 $this->assertNull(
 164                         $queue->batchPush(
 165                                 [ $this->newDedupedJob(), $this->newDedupedJob(), $this->newDedupedJob() ]
 166                         ),
 167                         "Push worked ($desc)" );
 168
 169                 $this->assertFalse( $queue->isEmpty(), "Queue is not empty ($desc)" );
 170
 171                 $queue->flushCaches();
 172                 $this->assertSame( 1, $queue->getSize(), "Queue size is correct ($desc)" );
 173                 $this->assertSame( 0, $queue->getAcquiredCount(), "No jobs active ($desc)" );
 174
 175                 $this->assertNull(
 176                         $queue->batchPush(
 177                                 [ $this->newDedupedJob(), $this->newDedupedJob(), $this->newDedupedJob() ]
 178                         ),
 179                         "Push worked ($desc)"
 180                 );
 181
 182                 $this->assertFalse( $queue->isEmpty(), "Queue is not empty ($desc)" );
 183
 184                 $queue->flushCaches();
 185                 $this->assertSame( 1, $queue->getSize(), "Queue size is correct ($desc)" );
 186                 $this->assertSame( 0, $queue->getAcquiredCount(), "No jobs active ($desc)" );
 187
 188                 $job1 = $queue->pop();
 189                 $this->assertTrue( $queue->isEmpty(), "Queue is empty ($desc)" );
 190
 191                 $queue->flushCaches();
 192                 $this->assertSame( 0, $queue->getSize(), "Queue is empty ($desc)" );
 193                 if ( $recycles ) {
 194                         $this->assertSame( 1, $queue->getAcquiredCount(), "Active job count ($desc)" );
 195                 }
 196
 197                 $queue->ack( $job1 );
 198
 199                 $queue->flushCaches();
 200                 $this->assertSame( 0, $queue->getAcquiredCount(), "Active job count ($desc)" );
 201         }
 202
 203         /**
 204          * @dataProvider provider_queueLists
 205          */
 206         public function testDeduplicationWhileClaimed( $queue, $recycles, $desc ) {
 207                 $queue = $this->$queue;
 208                 if ( !$queue ) {
 209                         $this->markTestSkipped( $desc );
 210                 }
 211
 212                 $job = $this->newDedupedJob();
 213                 $queue->push( $job );
 214
 215                 // De-duplication does not apply to already-claimed jobs
 216                 $j = $queue->pop();
 217                 $queue->push( $job );
 218                 $queue->ack( $j );
 219
 220                 $j = $queue->pop();
 221                 // Make sure ack() of the twin did not delete the sibling data
 222                 $this->assertInstanceOf( NullJob::class, $j );
 223         }
 224
 225         /**
 226          * @dataProvider provider_queueLists
 227          */
 228         public function testRootDeduplication( $queue, $recycles, $desc ) {
 229                 $queue = $this->$queue;
 230                 if ( !$queue ) {
 231                         $this->markTestSkipped( $desc );
 232                 }
 233
 234                 $this->assertTrue( $queue->isEmpty(), "Queue is empty ($desc)" );
 235
 236                 $queue->flushCaches();
 237                 $this->assertSame( 0, $queue->getSize(), "Queue is empty ($desc)" );
 238                 $this->assertSame( 0, $queue->getAcquiredCount(), "Queue is empty ($desc)" );
 239
 240                 $root1 = Job::newRootJobParams( "nulljobspam:testId" ); // task ID/timestamp
 241                 for ( $i = 0; $i < 5; ++$i ) {
 242                         $this->assertNull( $queue->push( $this->newJob( 0, $root1 ) ), "Push worked ($desc)" );
 243                 }
 244                 $queue->deduplicateRootJob( $this->newJob( 0, $root1 ) );
 245
 246                 $root2 = $root1;
 247                 # Add a second to UNIX epoch and format back to TS_MW
 248                 $root2_ts = strtotime( $root2['rootJobTimestamp'] );
 249                 $root2_ts++;
 250                 $root2['rootJobTimestamp'] = wfTimestamp( TS_MW, $root2_ts );
 251
 252                 $this->assertNotEquals( $root1['rootJobTimestamp'], $root2['rootJobTimestamp'],
 253                         "Root job signatures have different timestamps." );
 254                 for ( $i = 0; $i < 5; ++$i ) {
 255                         $this->assertNull( $queue->push( $this->newJob( 0, $root2 ) ), "Push worked ($desc)" );
 256                 }
 257                 $queue->deduplicateRootJob( $this->newJob( 0, $root2 ) );
 258
 259                 $this->assertFalse( $queue->isEmpty(), "Queue is not empty ($desc)" );
 260
 261                 $queue->flushCaches();
 262                 $this->assertEquals( 10, $queue->getSize(), "Queue size is correct ($desc)" );
 263                 $this->assertSame( 0, $queue->getAcquiredCount(), "No jobs active ($desc)" );
 264
 265                 $dupcount = 0;
 266                 $jobs = [];
 267                 do {
 268                         $job = $queue->pop();
 269                         if ( $job ) {
 270                                 $jobs[] = $job;
 271                                 $queue->ack( $job );
 272                         }
 273                         if ( $job instanceof DuplicateJob ) {
 274                                 ++$dupcount;
 275                         }
 276                 } while ( $job );
 277
 278                 $this->assertCount( 10, $jobs, "Correct number of jobs popped ($desc)" );
 279                 $this->assertEquals( 5, $dupcount, "Correct number of duplicate jobs popped ($desc)" );
 280         }
 281
 282         /**
 283          * @dataProvider provider_fifoQueueLists
 284          */
 285         public function testJobOrder( $queue, $recycles, $desc ) {
 286                 $queue = $this->$queue;
 287                 if ( !$queue ) {
 288                         $this->markTestSkipped( $desc );
 289                 }
 290
 291                 $this->assertTrue( $queue->isEmpty(), "Queue is empty ($desc)" );
 292
 293                 $queue->flushCaches();
 294                 $this->assertSame( 0, $queue->getSize(), "Queue is empty ($desc)" );
 295                 $this->assertSame( 0, $queue->getAcquiredCount(), "Queue is empty ($desc)" );
 296
 297                 for ( $i = 0; $i < 10; ++$i ) {
 298                         $this->assertNull( $queue->push( $this->newJob( $i ) ), "Push worked ($desc)" );
 299                 }
 300
 301                 for ( $i = 0; $i < 10; ++$i ) {
 302                         $job = $queue->pop();
 303                         $this->assertTrue( $job instanceof Job, "Jobs popped from queue ($desc)" );
 304                         $params = $job->getParams();
 305                         $this->assertEquals( $i, $params['i'], "Job popped from queue is FIFO ($desc)" );
 306                         $queue->ack( $job );
 307                 }
 308
 309                 $this->assertFalse( $queue->pop(), "Queue is not empty ($desc)" );
 310
 311                 $queue->flushCaches();
 312                 $this->assertSame( 0, $queue->getSize(), "Queue is empty ($desc)" );
 313                 $this->assertSame( 0, $queue->getAcquiredCount(), "No jobs active ($desc)" );
 314         }
 315
 316         public function testQueueAggregateTable() {
 317                 $this->hideDeprecated( 'JobQueue::getWiki' );
 318
 319                 $queue = $this->queueFifo;
 320                 if ( !$queue || !method_exists( $queue, 'getServerQueuesWithJobs' ) ) {
 321                         $this->markTestSkipped();
 322                 }
 323
 324                 $this->assertNotContains(
 325                         [ $queue->getType(), $queue->getWiki() ],
 326                         $queue->getServerQueuesWithJobs(),
 327                         "Null queue not in listing"
 328                 );
 329
 330                 $queue->push( $this->newJob( 0 ) );
 331
 332                 $this->assertContains(
 333                         [ $queue->getType(), $queue->getWiki() ],
 334                         $queue->getServerQueuesWithJobs(),
 335                         "Null queue in listing"
 336                 );
 337         }
 338
 339         public static function provider_queueLists() {
 340                 return [
 341                         [ 'queueRand', false, 'Random queue without ack()' ],
 342                         [ 'queueRandTTL', true, 'Random queue with ack()' ],
 343                         [ 'queueTimestamp', false, 'Time ordered queue without ack()' ],
 344                         [ 'queueTimestampTTL', true, 'Time ordered queue with ack()' ],
 345                         [ 'queueFifo', false, 'FIFO ordered queue without ack()' ],
 346                         [ 'queueFifoTTL', true, 'FIFO ordered queue with ack()' ]
 347                 ];
 348         }
 349
 350         public static function provider_fifoQueueLists() {
 351                 return [
 352                         [ 'queueFifo', false, 'Ordered queue without ack()' ],
 353                         [ 'queueFifoTTL', true, 'Ordered queue with ack()' ]
 354                 ];
 355         }
 356
 357         protected function newJob( $i = 0, $rootJob = [] ) {
 358                 $params = [
 359                         'namespace' => NS_MAIN,
 360                         'title' => 'Main_Page',
 361                         'lives' => 0,
 362                         'usleep' => 0,
 363                         'removeDuplicates' => 0,
 364                         'i' => $i
 365                 ] + $rootJob;
 366
 367                 return $this->getServiceContainer()->getJobFactory()->newJob( 'null', $params );
 368         }
 369
 370         protected function newDedupedJob( $i = 0, $rootJob = [] ) {
 371                 $params = [
 372                                 'namespace' => NS_MAIN,
 373                                 'title' => 'Main_Page',
 374                                 'lives' => 0,
 375                                 'usleep' => 0,
 376                                 'removeDuplicates' => 1,
 377                                 'i' => $i
 378                         ] + $rootJob;
 379
 380                 return $this->getServiceContainer()->getJobFactory()->newJob( 'null', $params );
 381         }
 382 }
 383
 384 class JobQueueDBSingle extends JobQueueDB {
 385         protected function getDB( $index ) {
 386                 $lb = MediaWikiServices::getInstance()->getDBLoadBalancer();
 387                 // Override to not use CONN_TRX_AUTOCOMMIT so that we see the same temporary `job` table
 388                 return $lb->getConnection( $index, [], $this->domain );
 389         }
 390 }