Merge "Add $wgMaxJobDBWriteDuration setting for avoiding replication lag"

diff --combined includes/DefaultSettings.php
index 5b7ca3e,5b83347..9436aa6
--- 1/includes/DefaultSettings.php
--- 2/includes/DefaultSettings.php
+++ b/includes/DefaultSettings.php
@@@ -6765,7 -6765,7 +6765,7 @@@ $wgUseRCPatrol = true
  /**
   * Whether to allow users to save their RecentChanges filters
   */
 -$wgStructuredChangeFiltersEnableSaving = false;
 +$wgStructuredChangeFiltersEnableSaving = true;
  
  /**
   * Use new page patrolling to check new pages on Special:Newpages
@@@ -8183,6 -8183,7 +8183,6 @@@ $wgPhpCli = '/usr/bin/php'
   * @note If multiple wikis are being served from the same process (e.g. the
   *  same fastCGI or Apache server), this setting must be the same on all those
   *  wikis.
 - * @see wfInitShellLocale()
   */
  $wgShellLocale = 'C.UTF-8';
  
@@@ -8559,6 -8560,15 +8559,15 @@@ $wgPopularPasswordFile = __DIR__ . '/..
   */
  $wgMaxUserDBWriteDuration = false;
  
+ /*
+  * Max time (in seconds) a job-generated transaction can spend in writes.
+  * If exceeded, the transaction is rolled back with an error instead of being committed.
+  *
+  * @var int|bool Disabled if false
+  * @since 1.30
+  */
+ $wgMaxJobDBWriteDuration = false;
+ 
  /**
   * Mapping of event channels (or channel categories) to EventRelayer configuration.
   *

diff --combined includes/jobqueue/JobRunner.php
index ff90a9e,18cabc5..49b7a45
--- 1/includes/jobqueue/JobRunner.php
--- 2/includes/jobqueue/JobRunner.php
+++ b/includes/jobqueue/JobRunner.php
@@@ -38,6 -38,8 +38,8 @@@ use Wikimedia\Rdbms\DBReplicationWaitEr
   * @since 1.24
   */
  class JobRunner implements LoggerAwareInterface {
+       /** @var Config */
+       protected $config;
        /** @var callable|null Debug output handler */
        protected $debug;
  
@@@ -74,6 -76,7 +76,7 @@@
                        $logger = LoggerFactory::getInstance( 'runJobs' );
                }
                $this->setLogger( $logger );
+               $this->config = MediaWikiServices::getInstance()->getMainConfig();
        }
  
        /**
@@@ -101,7 -104,8 +104,8 @@@
         * @return array Summary response that can easily be JSON serialized
         */
        public function run( array $options ) {
-               global $wgJobClasses, $wgTrxProfilerLimits;
+               $jobClasses = $this->config->get( 'JobClasses' );
+               $profilerLimits = $this->config->get( 'TrxProfilerLimits' );
  
                $response = [ 'jobs' => [], 'reached' => 'none-ready' ];
  
@@@ -111,7 -115,7 +115,7 @@@
                $noThrottle = isset( $options['throttle'] ) && !$options['throttle'];
  
                // Bail if job type is invalid
-               if ( $type !== false && !isset( $wgJobClasses[$type] ) ) {
+               if ( $type !== false && !isset( $jobClasses[$type] ) ) {
                        $response['reached'] = 'none-possible';
                        return $response;
                }
@@@ -136,7 -140,7 +140,7 @@@
                // Catch huge single updates that lead to replica DB lag
                $trxProfiler = Profiler::instance()->getTransactionProfiler();
                $trxProfiler->setLogger( LoggerFactory::getInstance( 'DBPerformance' ) );
-               $trxProfiler->setExpectations( $wgTrxProfilerLimits['JobRunner'], __METHOD__ );
+               $trxProfiler->setExpectations( $profilerLimits['JobRunner'], __METHOD__ );
  
                // Some jobs types should not run until a certain timestamp
                $backoffs = []; // map of (type => UNIX expiry)
@@@ -289,8 -293,6 +293,8 @@@
                        $status = $job->run();
                        $error = $job->getLastError();
                        $this->commitMasterChanges( $lbFactory, $job, $fnameTrxOwner );
 +                      // Important: this must be the last deferred update added (T100085, T154425)
 +                      DeferredUpdates::addCallableUpdate( [ JobQueueGroup::class, 'pushLazyJobs' ] );
                        // Run any deferred update tasks; doUpdates() manages transactions itself
                        DeferredUpdates::doUpdates();
                } catch ( Exception $e ) {
@@@ -362,15 -364,13 +366,13 @@@
         * @see $wgJobBackoffThrottling
         */
        private function getBackoffTimeToWait( Job $job ) {
-               global $wgJobBackoffThrottling;
+               $throttling = $this->config->get( 'JobBackoffThrottling' );
  
-               if ( !isset( $wgJobBackoffThrottling[$job->getType()] ) ||
-                       $job instanceof DuplicateJob // no work was done
-               ) {
+               if ( !isset( $throttling[$job->getType()] ) || $job instanceof DuplicateJob ) {
                        return 0; // not throttled
                }
  
-               $itemsPerSecond = $wgJobBackoffThrottling[$job->getType()];
+               $itemsPerSecond = $throttling[$job->getType()];
                if ( $itemsPerSecond <= 0 ) {
                        return 0; // not throttled
                }
@@@ -518,17 -518,17 +520,17 @@@
         * @throws DBError
         */
        private function commitMasterChanges( LBFactory $lbFactory, Job $job, $fnameTrxOwner ) {
-               global $wgJobSerialCommitThreshold;
+               $syncThreshold = $this->config->get( 'JobSerialCommitThreshold' );
  
                $time = false;
                $lb = $lbFactory->getMainLB( wfWikiID() );
-               if ( $wgJobSerialCommitThreshold !== false && $lb->getServerCount() > 1 ) {
+               if ( $syncThreshold !== false && $lb->getServerCount() > 1 ) {
                        // Generally, there is one master connection to the local DB
                        $dbwSerial = $lb->getAnyOpenConnection( $lb->getWriterIndex() );
                        // We need natively blocking fast locks
                        if ( $dbwSerial && $dbwSerial->namedLocksEnqueue() ) {
                                $time = $dbwSerial->pendingWriteQueryDuration( $dbwSerial::ESTIMATE_DB_APPLY );
-                               if ( $time < $wgJobSerialCommitThreshold ) {
+                               if ( $time < $syncThreshold ) {
                                        $dbwSerial = false;
                                }
                        } else {
@@@ -540,7 -540,12 +542,12 @@@
                }
  
                if ( !$dbwSerial ) {
-                       $lbFactory->commitMasterChanges( $fnameTrxOwner );
+                       $lbFactory->commitMasterChanges(
+                               $fnameTrxOwner,
+                               // Abort if any transaction was too big
+                               [ 'maxWriteDuration' => $this->config->get( 'MaxJobDBWriteDuration' ) ]
+                       );
+ 
                        return;
                }
  
@@@ -565,7 -570,11 +572,11 @@@
                }
  
                // Actually commit the DB master changes
-               $lbFactory->commitMasterChanges( $fnameTrxOwner );
+               $lbFactory->commitMasterChanges(
+                       $fnameTrxOwner,
+                       // Abort if any transaction was too big
+                       [ 'maxWriteDuration' => $this->config->get( 'MaxJobDBWriteDuration' ) ]
+               );
                ScopedCallback::consume( $unlocker );
        }
  }