DeferredUpdates.php

Go to the documentation of this file.

<?php
use MediaWiki\Logger\LoggerFactory;
use Wikimedia\Rdbms\IDatabase;
use Wikimedia\ScopedCallback;
 
class DeferredUpdates {
    public const ALL = 0;
    public const PRESEND = 1;
    public const POSTSEND = 2;
 
    public const STAGES = [ self::PRESEND, self::POSTSEND ];
 
    private const BIG_QUEUE_SIZE = 100;
 
    private static $scopeStack;
 
    private static $preventOpportunisticUpdates = 0;
 
    private static function getScopeStack(): DeferredUpdatesScopeStack {
        self::$scopeStack ??= new DeferredUpdatesScopeMediaWikiStack();
        return self::$scopeStack;
    }
 
    public static function setScopeStack( DeferredUpdatesScopeStack $scopeStack ): void {
        if ( !defined( 'MW_PHPUNIT_TEST' ) ) {
            throw new LogicException( 'Cannot reconfigure DeferredUpdates outside tests' );
        }
        self::$scopeStack = $scopeStack;
    }
 
    public static function addUpdate( DeferrableUpdate $update, $stage = self::POSTSEND ) {
        self::getScopeStack()->current()->addUpdate( $update, $stage );
        self::tryOpportunisticExecute();
    }
 
    public static function addCallableUpdate( $callable, $stage = self::POSTSEND, $dbw = null ) {
        self::addUpdate( new MWCallableUpdate( $callable, wfGetCaller(), $dbw ), $stage );
    }
 
    private static function run( DeferrableUpdate $update ): ?Throwable {
        $logger = LoggerFactory::getInstance( 'DeferredUpdates' );
 
        $type = get_class( $update )
            . ( $update instanceof DeferrableCallback ? '_' . $update->getOrigin() : '' );
        $updateId = spl_object_id( $update );
        $logger->debug( __METHOD__ . ": started $type #$updateId" );
 
        $updateException = null;
 
        $startTime = microtime( true );
        try {
            self::attemptUpdate( $update );
        } catch ( Throwable $updateException ) {
            MWExceptionHandler::logException( $updateException );
            $logger->error(
                "Deferred update '{deferred_type}' failed to run.",
                [
                    'deferred_type' => $type,
                    'exception' => $updateException,
                ]
            );
            self::getScopeStack()->onRunUpdateFailed( $update );
        } finally {
            $walltime = microtime( true ) - $startTime;
            $logger->debug( __METHOD__ . ": ended $type #$updateId, processing time: $walltime" );
        }
 
        // Try to push the update as a job so it can run later if possible
        if ( $updateException && $update instanceof EnqueueableDataUpdate ) {
            try {
                self::getScopeStack()->queueDataUpdate( $update );
            } catch ( Throwable $jobException ) {
                MWExceptionHandler::logException( $jobException );
                $logger->error(
                    "Deferred update '{deferred_type}' failed to enqueue as a job.",
                    [
                        'deferred_type' => $type,
                        'exception' => $jobException,
                    ]
                );
                self::getScopeStack()->onRunUpdateFailed( $update );
            }
        }
 
        return $updateException;
    }
 
    public static function doUpdates( $stage = self::ALL ) {
        $guiError = null;
        $exception = null;
 
        $scope = self::getScopeStack()->current();
 
        // T249069: recursion is not possible once explicit transaction rounds are involved
        $activeUpdate = $scope->getActiveUpdate();
        if ( $activeUpdate ) {
            $class = get_class( $activeUpdate );
            if ( !( $activeUpdate instanceof TransactionRoundAwareUpdate ) ) {
                throw new LogicException(
                    __METHOD__ . ": reached from $class, which is not TransactionRoundAwareUpdate"
                );
            }
            if ( $activeUpdate->getTransactionRoundRequirement() !== $activeUpdate::TRX_ROUND_ABSENT ) {
                throw new LogicException(
                    __METHOD__ . ": reached from $class, which does not specify TRX_ROUND_ABSENT"
                );
            }
        }
 
        $scope->processUpdates(
            $stage,
            static function ( DeferrableUpdate $update, $activeStage ) use ( &$guiError, &$exception ) {
                $scopeStack = self::getScopeStack();
                $childScope = $scopeStack->descend( $activeStage, $update );
                try {
                    $e = self::run( $update );
                    $guiError = $guiError ?: ( $e instanceof ErrorPageError ? $e : null );
                    $exception = $exception ?: $e;
                    // Any addUpdate() calls between descend() and ascend() used the sub-queue.
                    // In rare cases, DeferrableUpdate::doUpdates() will process them by calling
                    // doUpdates() itself. In any case, process remaining updates in the subqueue.
                    // them, enqueueing them, or transferring them to the parent scope
                    // queues as appropriate...
                    $childScope->processUpdates(
                        $activeStage,
                        static function ( DeferrableUpdate $sub ) use ( &$guiError, &$exception ) {
                            $e = self::run( $sub );
                            $guiError = $guiError ?: ( $e instanceof ErrorPageError ? $e : null );
                            $exception = $exception ?: $e;
                        }
                    );
                } finally {
                    $scopeStack->ascend();
                }
            }
        );
 
        // VW-style hack to work around T190178, so we can make sure
        // PageMetaDataUpdater doesn't throw exceptions.
        if ( $exception && defined( 'MW_PHPUNIT_TEST' ) ) {
            throw $exception;
        }
 
        // Throw the first of any GUI errors as long as the context is HTTP pre-send. However,
        // callers should check permissions *before* enqueueing updates. If the main transaction
        // round actions succeed but some deferred updates fail due to permissions errors then
        // there is a risk that some secondary data was not properly updated.
        if ( $guiError && $stage === self::PRESEND && !headers_sent() ) {
            throw $guiError;
        }
    }
 
    public static function tryOpportunisticExecute(): bool {
        // Leave execution up to the current loop if an update is already in progress
        // or if updates are explicitly disabled
        if ( self::getRecursiveExecutionStackDepth()
            || self::$preventOpportunisticUpdates
        ) {
            return false;
        }
 
        if ( self::getScopeStack()->allowOpportunisticUpdates() ) {
            self::doUpdates( self::ALL );
            return true;
        }
 
        if ( self::pendingUpdatesCount() >= self::BIG_QUEUE_SIZE ) {
            // There are a large number of pending updates and none of them can run yet.
            // The odds of losing updates due to an error increases when executing long queues
            // and when large amounts of time pass while tasks are queued. Mitigate this by
            // trying to eagerly move updates to the JobQueue when possible.
            //
            // TODO: Do we still need this now maintenance scripts automatically call
            // tryOpportunisticExecute from addUpdate, from every commit, and every
            // waitForReplication call?
            self::getScopeStack()->current()->consumeMatchingUpdates(
                self::ALL,
                EnqueueableDataUpdate::class,
                static function ( EnqueueableDataUpdate $update ) {
                    self::getScopeStack()->queueDataUpdate( $update );
                }
            );
        }
 
        return false;
    }
 
    public static function preventOpportunisticUpdates() {
        self::$preventOpportunisticUpdates++;
        return new ScopedCallback( static function () {
            self::$preventOpportunisticUpdates--;
        } );
    }
 
    public static function pendingUpdatesCount() {
        return self::getScopeStack()->current()->pendingUpdatesCount();
    }
 
    public static function getPendingUpdates( $stage = self::ALL ) {
        return self::getScopeStack()->current()->getPendingUpdates( $stage );
    }
 
    public static function clearPendingUpdates() {
        self::getScopeStack()->current()->clearPendingUpdates();
    }
 
    public static function getRecursiveExecutionStackDepth() {
        return self::getScopeStack()->getRecursiveDepth();
    }
 
    public static function attemptUpdate( DeferrableUpdate $update ) {
        self::getScopeStack()->onRunUpdateStart( $update );
 
        $update->doUpdate();
 
        self::getScopeStack()->onRunUpdateEnd( $update );
    }
}