ChronologyProtector.php

Go to the documentation of this file.

<?php
namespace Wikimedia\Rdbms;
 
use LogicException;
use Psr\Log\LoggerAwareInterface;
use Psr\Log\LoggerInterface;
use Psr\Log\NullLogger;
use Wikimedia\ObjectCache\BagOStuff;
use Wikimedia\ObjectCache\EmptyBagOStuff;
 
class ChronologyProtector implements LoggerAwareInterface {
    private $requestInfo;
    private string $secret;
    private bool $cliMode;
    private $store;
    protected $logger;
 
    protected $key;
    protected $clientId;
    protected $clientLogInfo;
    protected $waitForPosIndex;
 
    protected $enabled = true;
    protected $startupTimestamp;
 
    protected $startupPositionsByPrimary = [];
    protected $shutdownPositionsByPrimary = [];
    protected $startupTimestampsByCluster = [];
    protected $shutdownTimestampsByCluster = [];
 
    private $wallClockOverride;
 
    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_TIMESTAMPS = 'timestamps';
    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 setLogger( LoggerInterface $logger ) {
        $this->load();
        $this->logger = $logger;
    }
 
    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 ( $lb->hasStreamingReplicaServers() ) {
            $pos = $lb->getPrimaryPos();
            if ( $pos ) {
                $this->logger->debug( __METHOD__ . ": $cluster ($masterName) position now '$pos'" );
                $this->shutdownPositionsByPrimary[$masterName] = $pos;
                $this->shutdownTimestampsByCluster[$cluster] = $pos->asOfTime();
            } else {
                $this->logger->debug( __METHOD__ . ": $cluster ($masterName) position unknown" );
                $this->shutdownTimestampsByCluster[$cluster] = $this->getCurrentTime();
            }
        } else {
            $this->logger->debug( __METHOD__ . ": $cluster ($masterName) has no replication" );
            $this->shutdownTimestampsByCluster[$cluster] = $this->getCurrentTime();
        }
    }
 
    public function persistSessionReplicationPositions( &$clientPosIndex = null ) {
        $this->load();
        if ( !$this->enabled ) {
            return [];
        }
 
        if ( !$this->shutdownTimestampsByCluster ) {
            $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,
                $this->shutdownTimestampsByCluster,
                $clientPosIndex
            );
 
            $ok = $this->store->set(
                $this->key,
                $this->marshalPositions( $positions ),
                self::POSITION_STORE_TTL
            );
            unset( $scopeLock );
        } else {
            $ok = false;
        }
 
        $clusterList = implode( ', ', array_keys( $this->shutdownTimestampsByCluster ) );
 
        if ( $ok ) {
            $this->logger->debug( "ChronologyProtector saved position data for $clusterList" );
            $bouncedPositions = [];
        } else {
            // Maybe position store is down
            $this->logger->warning( "ChronologyProtector failed to save position data for $clusterList" );
            $clientPosIndex = null;
            $bouncedPositions = $this->shutdownPositionsByPrimary;
        }
 
        return $bouncedPositions;
    }
 
    public function getTouched( ILoadBalancer $lb ) {
        $this->load();
        if ( !$this->enabled ) {
            return false;
        }
 
        $cluster = $lb->getClusterName();
 
        $timestampsByCluster = $this->getStartupSessionTimestamps();
        $timestamp = $timestampsByCluster[$cluster] ?? null;
        if ( $timestamp === null ) {
            $recentTouchTimestamp = false;
        } elseif ( ( $this->startupTimestamp - $timestamp ) > self::POSITION_COOKIE_TTL ) {
            // If the position store is not replicated among datacenters and the cookie that
            // sticks the client to the primary datacenter expires, then the touch timestamp
            // will be found for requests in one datacenter but not others. For consistency,
            // return false once the user is no longer routed to the primary datacenter.
            $recentTouchTimestamp = false;
            $this->logger->debug( __METHOD__ . ": old timestamp ($timestamp) for $cluster" );
        } else {
            $recentTouchTimestamp = $timestamp;
            $this->logger->debug( __METHOD__ . ": recent timestamp ($timestamp) for $cluster" );
        }
 
        return $recentTouchTimestamp;
    }
 
    protected function getStartupSessionPositions() {
        $this->lazyStartup();
 
        return $this->startupPositionsByPrimary;
    }
 
    protected function getStartupSessionTimestamps() {
        $this->lazyStartup();
 
        return $this->startupTimestampsByCluster;
    }
 
    protected function lazyStartup() {
        if ( $this->startupTimestamp !== null ) {
            return;
        }
 
        $this->startupTimestamp = $this->getCurrentTime();
 
        // 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 = [];
            $this->startupTimestampsByCluster = [];
            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] : [];
        $this->startupTimestampsByCluster = $data[self::FLD_TIMESTAMPS] ?? [];
 
        // 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,
        array $shutdownTimestamps,
        ?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;
            }
        }
 
        $mergedTimestamps = $storedValue[self::FLD_TIMESTAMPS] ?? [];
        // Use the newest touch timestamp for each DB primary
        foreach ( $shutdownTimestamps as $cluster => $timestamp ) {
            if (
                !isset( $mergedTimestamps[$cluster] ) ||
                $timestamp > $mergedTimestamps[$cluster]
            ) {
                $mergedTimestamps[$cluster] = $timestamp;
            }
        }
 
        $clientPosIndex = ( $storedValue[self::FLD_WRITE_INDEX] ?? 0 ) + 1;
 
        return [
            self::FLD_POSITIONS => $mergedPositions,
            self::FLD_TIMESTAMPS => $mergedTimestamps,
            self::FLD_WRITE_INDEX => $clientPosIndex
        ];
    }
 
    protected function getCurrentTime() {
        if ( $this->wallClockOverride ) {
            return $this->wallClockOverride;
        }
 
        $clockTime = (float)time(); // call this first
        // microtime() can severely drift from time() and the microtime() value of other threads.
        // Instead of seeing the current time as being in the past, use the value of time().
        return max( microtime( true ), $clockTime );
    }
 
    public function setMockTime( &$time ) {
        $this->load();
        $this->wallClockOverride =& $time;
    }
 
    private function marshalPositions( array $positions ) {
        foreach ( $positions[ self::FLD_POSITIONS ] as $key => $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 ) {
            $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 ];
    }
}