ChronologyProtector.php

Go to the documentation of this file.

<?php
namespace Wikimedia\Rdbms;
 
use BagOStuff;
use Psr\Log\LoggerAwareInterface;
use Psr\Log\LoggerInterface;
use Psr\Log\NullLogger;
 
class ChronologyProtector implements LoggerAwareInterface {
    protected $store;
    protected $logger;
 
    protected $key;
    protected $clientId;
    protected $clientLogInfo;
    protected $waitForPosIndex;
 
    protected $enabled = true;
    protected $positionWaitsEnabled = true;
    protected $startupTimestamp;
 
    protected $startupPositionsByPrimary = [];
    protected $shutdownPositionsByPrimary = [];
    protected $startupTimestampsByCluster = [];
    protected $shutdownTimestampsByCluster = [];
 
    private $wallClockOverride;
 
    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(
        BagOStuff $store,
        array $client,
        ?int $clientPosIndex,
        string $secret = ''
    ) {
        $this->store = $store;
 
        if ( isset( $client['clientId'] ) ) {
            $this->clientId = $client['clientId'];
        } else {
            $this->clientId = ( $secret != '' )
                ? hash_hmac( 'md5', $client['ip'] . "\n" . $client['agent'], $secret )
                : md5( $client['ip'] . "\n" . $client['agent'] );
        }
        $this->key = $store->makeGlobalKey( __CLASS__, $this->clientId, 'v3' );
        $this->waitForPosIndex = $clientPosIndex;
 
        $this->clientLogInfo = [
            'clientIP' => $client['ip'],
            'clientAgent' => $client['agent'],
            'clientId' => $client['clientId'] ?? null
        ];
 
        $this->logger = new NullLogger();
    }
 
    public function setLogger( LoggerInterface $logger ) {
        $this->logger = $logger;
    }
 
    public function getClientId() {
        return $this->clientId;
    }
 
    public function setEnabled( $enabled ) {
        $this->enabled = $enabled;
    }
 
    public function setWaitEnabled( $enabled ) {
        $this->positionWaitsEnabled = $enabled;
    }
 
    public function applySessionReplicationPosition( ILoadBalancer $lb ) {
        if ( !$this->enabled || !$this->positionWaitsEnabled ) {
            return;
        }
 
        $cluster = $lb->getClusterName();
        $primaryName = $lb->getServerName( $lb->getWriterIndex() );
 
        $pos = $this->getStartupSessionPositions()[$primaryName] ?? null;
        if ( $pos instanceof DBPrimaryPos ) {
            $this->logger->debug( __METHOD__ . ": $cluster ($primaryName) position is '$pos'" );
            $lb->waitFor( $pos );
        } else {
            $this->logger->debug( __METHOD__ . ": $cluster ($primaryName) has no position" );
        }
    }
 
    public function stageSessionReplicationPosition( ILoadBalancer $lb ) {
        if ( !$this->enabled || !$lb->hasOrMadeRecentPrimaryChanges( INF ) ) {
            return;
        }
 
        $cluster = $lb->getClusterName();
        $masterName = $lb->getServerName( $lb->getWriterIndex() );
 
        if ( $lb->hasStreamingReplicaServers() ) {
            $pos = $lb->getReplicaResumePos();
            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 ) {
        if ( !$this->enabled ) {
            return [];
        }
 
        if ( !$this->shutdownTimestampsByCluster ) {
            $this->logger->debug( __METHOD__ . ": no primary positions/timestamps 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 ) {
            $bouncedPositions = [];
            $this->logger->debug(
                __METHOD__ . ": saved primary positions/timestamp for DB cluster(s) $clusterList"
            );
 
        } else {
            $clientPosIndex = null; // nothing saved
            $bouncedPositions = $this->shutdownPositionsByPrimary;
            // Raced out too many times or stash is down
            $this->logger->warning(
                __METHOD__ . ": failed to save primary positions for DB cluster(s) $clusterList"
            );
        }
 
        return $bouncedPositions;
    }
 
    public function getTouched( ILoadBalancer $lb ) {
        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();
        $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->positionWaitsEnabled && $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->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;
    }
}