ChronologyProtector.php

Go to the documentation of this file.

<?php
namespace Wikimedia\Rdbms;
 
use LogicException;
use Psr\Log\LoggerInterface;
use Psr\Log\NullLogger;
use Wikimedia\ObjectCache\BagOStuff;
use Wikimedia\ObjectCache\EmptyBagOStuff;
 
class ChronologyProtector {
    private $requestInfo;
    private string $secret;
    private bool $cliMode;
    private $store;
    protected $logger;
 
    protected $key;
    protected $clientId;
    protected $clientLogInfo;
    protected $waitForPosIndex;
 
    protected $enabled = true;
 
    protected $startupPositionsByPrimary = null;
    protected $shutdownPositionsByPrimary = [];
 
    private $hasNewClientId = false;
 
    public const POSITION_COOKIE_TTL = 10;
    private const POSITION_STORE_TTL = 60;
 
    private const LOCK_TIMEOUT = 3;
    private const LOCK_TTL = 6;
 
    private const FLD_POSITIONS = 'positions';
    private const FLD_WRITE_INDEX = 'writeIndex';
 
    public function __construct( $cpStash = null, $secret = null, $cliMode = null, $logger = null ) {
        $this->requestInfo = [
            'IPAddress' => $_SERVER['REMOTE_ADDR'] ?? '',
            'UserAgent' => $_SERVER['HTTP_USER_AGENT'] ?? '',
            // Headers application can inject via LBFactory::setRequestInfo()
            'ChronologyClientId' => null, // prior $cpClientId value from LBFactory::shutdown()
            'ChronologyPositionIndex' => null // prior $cpIndex value from LBFactory::shutdown()
        ];
        $this->store = $cpStash ?? new EmptyBagOStuff();
        $this->secret = $secret ?? '';
        $this->logger = $logger ?? new NullLogger();
        $this->cliMode = $cliMode ?? ( PHP_SAPI === 'cli' || PHP_SAPI === 'phpdbg' );
    }
 
    private function load() {
        // Not enabled or already loaded, short-circuit.
        if ( !$this->enabled || $this->clientId ) {
            return;
        }
        $client = [
            'ip' => $this->requestInfo['IPAddress'],
            'agent' => $this->requestInfo['UserAgent'],
            'clientId' => $this->requestInfo['ChronologyClientId'] ?: null
        ];
        if ( $this->cliMode ) {
            $this->setEnabled( false );
        } elseif ( $this->store instanceof EmptyBagOStuff ) {
            // No where to store any DB positions and wait for them to appear
            $this->setEnabled( false );
            $this->logger->debug( 'Cannot use ChronologyProtector with EmptyBagOStuff' );
        }
 
        if ( isset( $client['clientId'] ) ) {
            $this->clientId = $client['clientId'];
        } else {
            $this->hasNewClientId = true;
            $this->clientId = ( $this->secret != '' )
                ? hash_hmac( 'md5', $client['ip'] . "\n" . $client['agent'], $this->secret )
                : md5( $client['ip'] . "\n" . $client['agent'] );
        }
        $this->key = $this->store->makeGlobalKey( __CLASS__, $this->clientId, 'v4' );
        $this->waitForPosIndex = $this->requestInfo['ChronologyPositionIndex'];
 
        $this->clientLogInfo = [
            'clientIP' => $client['ip'],
            'clientAgent' => $client['agent'],
            'clientId' => $client['clientId'] ?? null
        ];
    }
 
    public function setRequestInfo( array $info ) {
        if ( $this->clientId ) {
            throw new LogicException( 'ChronologyProtector already initialized' );
        }
 
        $this->requestInfo = $info + $this->requestInfo;
    }
 
    public function getClientId() {
        $this->load();
        return $this->clientId;
    }
 
    public function setEnabled( $enabled ) {
        $this->enabled = $enabled;
    }
 
    public function getSessionPrimaryPos( ILoadBalancer $lb ) {
        $this->load();
        if ( !$this->enabled ) {
            return null;
        }
 
        $cluster = $lb->getClusterName();
        $primaryName = $lb->getServerName( ServerInfo::WRITER_INDEX );
 
        $pos = $this->getStartupSessionPositions()[$primaryName] ?? null;
        if ( $pos instanceof DBPrimaryPos ) {
            $this->logger->debug( "ChronologyProtector will wait for '$pos' on $cluster ($primaryName)'" );
        } else {
            $this->logger->debug( "ChronologyProtector skips wait on $cluster ($primaryName)" );
        }
 
        return $pos;
    }
 
    public function stageSessionPrimaryPos( ILoadBalancer $lb ) {
        $this->load();
        if ( !$this->enabled || !$lb->hasOrMadeRecentPrimaryChanges( INF ) ) {
            return;
        }
 
        $cluster = $lb->getClusterName();
        $masterName = $lb->getServerName( ServerInfo::WRITER_INDEX );
 
        // If LoadBalancer::hasStreamingReplicaServers is false (single DB host),
        // or if the database type has no replication (i.e. SQLite), then we do not need to
        // save any position data, as it would never be loaded or waited for. When we save this
        // data, it is for DB_REPLICA queries in future requests, which load it via
        // ChronologyProtector::getSessionPrimaryPos (from LoadBalancer::getReaderIndex)
        // and wait for that position. In a single-server setup, all queries go the primary DB.
        //
        // In that case we still store a null value, so that ChronologyProtector::getTouched
        // reliably detects recent writes for non-database purposes,
        // such as ParserOutputAccess/PoolWorkArticleViewCurrent. This also makes getTouched()
        // easier to setup and test, as we it work work always once CP is enabled
        // (e.g. wgMicroStashType or wgMainCacheType set to a non-DB cache).
        if ( $lb->hasStreamingReplicaServers() ) {
            $pos = $lb->getPrimaryPos();
            if ( $pos ) {
                $this->logger->debug( __METHOD__ . ": $cluster ($masterName) position now '$pos'" );
                $this->shutdownPositionsByPrimary[$masterName] = $pos;
            } else {
                $this->logger->debug( __METHOD__ . ": $cluster ($masterName) position unknown" );
                $this->shutdownPositionsByPrimary[$masterName] = null;
            }
        } else {
            $this->logger->debug( __METHOD__ . ": $cluster ($masterName) has no replication" );
            $this->shutdownPositionsByPrimary[$masterName] = null;
        }
    }
 
    public function persistSessionReplicationPositions( &$clientPosIndex = null ) {
        $this->load();
        if ( !$this->enabled ) {
            return [];
        }
 
        if ( !$this->shutdownPositionsByPrimary ) {
            $this->logger->debug( __METHOD__ . ": no primary positions data to save" );
 
            return [];
        }
 
        $scopeLock = $this->store->getScopedLock( $this->key, self::LOCK_TIMEOUT, self::LOCK_TTL );
        if ( $scopeLock ) {
            $positions = $this->mergePositions(
                $this->unmarshalPositions( $this->store->get( $this->key ) ),
                $this->shutdownPositionsByPrimary,
                $clientPosIndex
            );
 
            $ok = $this->store->set(
                $this->key,
                $this->marshalPositions( $positions ),
                self::POSITION_STORE_TTL
            );
            unset( $scopeLock );
        } else {
            $ok = false;
        }
 
        $primaryList = implode( ', ', array_keys( $this->shutdownPositionsByPrimary ) );
 
        if ( $ok ) {
            $this->logger->debug( "ChronologyProtector saved position data for $primaryList" );
            $bouncedPositions = [];
        } else {
            // Maybe position store is down
            $this->logger->warning( "ChronologyProtector failed to save position data for $primaryList" );
            $clientPosIndex = null;
            $bouncedPositions = $this->shutdownPositionsByPrimary;
        }
 
        return $bouncedPositions;
    }
 
    public function getTouched() {
        $this->load();
 
        if ( !$this->enabled ) {
            return false;
        }
 
        if ( $this->getStartupSessionPositions() ) {
            $this->logger->debug( __METHOD__ . ": found recent writes" );
            return true;
        }
 
        $this->logger->debug( __METHOD__ . ": found no recent writes" );
        return false;
    }
 
    protected function getStartupSessionPositions() {
        $this->lazyStartup();
 
        return $this->startupPositionsByPrimary;
    }
 
    protected function lazyStartup() {
        if ( $this->startupPositionsByPrimary !== null ) {
            return;
        }
 
        // There wasn't a client id in the cookie so we built one
        // There is no point in looking it up.
        if ( $this->hasNewClientId ) {
            $this->startupPositionsByPrimary = [];
            return;
        }
 
        $this->logger->debug( 'ChronologyProtector using store ' . get_class( $this->store ) );
        $this->logger->debug( "ChronologyProtector fetching positions for {$this->clientId}" );
 
        $data = $this->unmarshalPositions( $this->store->get( $this->key ) );
 
        $this->startupPositionsByPrimary = $data ? $data[self::FLD_POSITIONS] : [];
 
        // When a stored array expires and is re-created under the same (deterministic) key,
        // the array value naturally starts again from index zero. As such, it is possible
        // that if certain store writes were lost (e.g. store down), that we unintentionally
        // point to an offset in an older incarnation of the array.
        // We don't try to detect or do something about this because:
        // 1. Waiting for an older offset is harmless and generally no-ops.
        // 2. The older value will have expired by now and thus treated as non-existing,
        //    which means we wouldn't even "see" it here.
        $indexReached = is_array( $data ) ? $data[self::FLD_WRITE_INDEX] : null;
        if ( $this->waitForPosIndex > 0 ) {
            if ( $indexReached >= $this->waitForPosIndex ) {
                $this->logger->debug( 'expected and found position index {cpPosIndex}', [
                    'cpPosIndex' => $this->waitForPosIndex,
                ] + $this->clientLogInfo );
            } else {
                $this->logger->warning( 'expected but failed to find position index {cpPosIndex}', [
                    'cpPosIndex' => $this->waitForPosIndex,
                    'indexReached' => $indexReached,
                    'exception' => new \RuntimeException(),
                ] + $this->clientLogInfo );
            }
        } else {
            if ( $indexReached ) {
                $this->logger->debug( 'found position data with index {indexReached}', [
                    'indexReached' => $indexReached
                ] + $this->clientLogInfo );
            }
        }
    }
 
    protected function mergePositions(
        $storedValue,
        array $shutdownPositions,
        ?int &$clientPosIndex = null
    ) {
        $mergedPositions = $storedValue[self::FLD_POSITIONS] ?? [];
        // Use the newest positions for each DB primary
        foreach ( $shutdownPositions as $masterName => $pos ) {
            if (
                !isset( $mergedPositions[$masterName] ) ||
                !( $mergedPositions[$masterName] instanceof DBPrimaryPos ) ||
                $pos->asOfTime() > $mergedPositions[$masterName]->asOfTime()
            ) {
                $mergedPositions[$masterName] = $pos;
            }
        }
 
        $clientPosIndex = ( $storedValue[self::FLD_WRITE_INDEX] ?? 0 ) + 1;
 
        return [
            self::FLD_POSITIONS => $mergedPositions,
            self::FLD_WRITE_INDEX => $clientPosIndex
        ];
    }
 
    private function marshalPositions( array $positions ): array {
        foreach ( $positions[ self::FLD_POSITIONS ] as $key => $pos ) {
            if ( $pos ) {
                $positions[ self::FLD_POSITIONS ][ $key ] = $pos->toArray();
            }
        }
 
        return $positions;
    }
 
    private function unmarshalPositions( $positions ) {
        if ( !$positions ) {
            return $positions;
        }
 
        foreach ( $positions[ self::FLD_POSITIONS ] as $key => $pos ) {
            if ( $pos ) {
                $class = $pos[ '_type_' ];
                $positions[ self::FLD_POSITIONS ][ $key ] = $class::newFromArray( $pos );
            }
        }
 
        return $positions;
    }
 
    public static function makeCookieValueFromCPIndex(
        int $writeIndex,
        int $time,
        string $clientId
    ) {
        // Format is "<write index>@<write timestamp>#<client ID hash>"
        return "{$writeIndex}@{$time}#{$clientId}";
    }
 
    public static function getCPInfoFromCookieValue( ?string $value, int $minTimestamp ) {
        static $placeholder = [ 'index' => null, 'clientId' => null ];
 
        if ( $value === null ) {
            return $placeholder; // not set
        }
 
        // Format is "<write index>@<write timestamp>#<client ID hash>"
        if ( !preg_match( '/^(\d+)@(\d+)#([0-9a-f]{32})$/', $value, $m ) ) {
            return $placeholder; // invalid
        }
 
        $index = (int)$m[1];
        if ( $index <= 0 ) {
            return $placeholder; // invalid
        } elseif ( isset( $m[2] ) && $m[2] !== '' && (int)$m[2] < $minTimestamp ) {
            return $placeholder; // expired
        }
 
        $clientId = ( isset( $m[3] ) && $m[3] !== '' ) ? $m[3] : null;
 
        return [ 'index' => $index, 'clientId' => $clientId ];
    }
}