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;
use Wikimedia\WaitConditionLoop;
 
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 $startupPositionsByMaster = [];
    protected $shutdownPositionsByMaster = [];
    protected $startupTimestampsByCluster = [];
    protected $shutdownTimestampsByCluster = [];
 
    private $wallClockOverride;
 
    public const POSITION_COOKIE_TTL = 10;
    private const POSITION_STORE_TTL = 60;
    private const POSITION_INDEX_WAIT_TIMEOUT = 5;
 
    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, 'v2' );
        $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();
        $masterName = $lb->getServerName( $lb->getWriterIndex() );
 
        $pos = $this->getStartupSessionPositions()[$masterName] ?? null;
        if ( $pos instanceof DBPrimaryPos ) {
            $this->logger->debug( __METHOD__ . ": $cluster ($masterName) position is '$pos'" );
            $lb->waitFor( $pos );
        } else {
            $this->logger->debug( __METHOD__ . ": $cluster ($masterName) 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->shutdownPositionsByMaster[$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 ) {
            $ok = $this->store->set(
                $this->key,
                $this->mergePositions(
                    $this->store->get( $this->key ),
                    $this->shutdownPositionsByMaster,
                    $this->shutdownTimestampsByCluster,
                    $clientPosIndex
                ),
                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->shutdownPositionsByMaster;
            // 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->startupPositionsByMaster;
    }
 
    protected function getStartupSessionTimestamps() {
        $this->lazyStartup();
 
        return $this->startupTimestampsByCluster;
    }
 
    protected function lazyStartup() {
        if ( $this->startupTimestamp !== null ) {
            return;
        }
 
        $this->startupTimestamp = $this->getCurrentTime();
        $this->logger->debug(
            __METHOD__ .
            ": client ID is {$this->clientId}; key is {$this->key}"
        );
 
        // If there is an expectation to see primary positions from a certain write
        // index or higher, then block until it appears, or until a timeout is reached.
        // Since the write index restarts each time the key is created, it is possible that
        // a lagged store has a matching key write index. However, in that case, it should
        // already be expired and thus treated as non-existing, maintaining correctness.
        if ( $this->positionWaitsEnabled && $this->waitForPosIndex > 0 ) {
            $data = null;
            $indexReached = null; // highest index reached in the position store
            $loop = new WaitConditionLoop(
                function () use ( &$data, &$indexReached ) {
                    $data = $this->store->get( $this->key );
                    if ( !is_array( $data ) ) {
                        return WaitConditionLoop::CONDITION_CONTINUE; // not found yet
                    } elseif ( !isset( $data[self::FLD_WRITE_INDEX] ) ) {
                        return WaitConditionLoop::CONDITION_REACHED; // b/c
                    }
                    $indexReached = max( $data[self::FLD_WRITE_INDEX], $indexReached );
 
                    return ( $data[self::FLD_WRITE_INDEX] >= $this->waitForPosIndex )
                        ? WaitConditionLoop::CONDITION_REACHED
                        : WaitConditionLoop::CONDITION_CONTINUE;
                },
                self::POSITION_INDEX_WAIT_TIMEOUT
            );
            $result = $loop->invoke();
            $waitedMs = $loop->getLastWaitTime() * 1e3;
 
            if ( $result == $loop::CONDITION_REACHED ) {
                $this->logger->debug(
                    __METHOD__ . ": expected and found position index {cpPosIndex}.",
                    [
                        'cpPosIndex' => $this->waitForPosIndex,
                        'waitTimeMs' => $waitedMs
                    ] + $this->clientLogInfo
                );
            } else {
                $this->logger->warning(
                    __METHOD__ . ": expected but failed to find position index {cpPosIndex}.",
                    [
                        'cpPosIndex' => $this->waitForPosIndex,
                        'indexReached' => $indexReached,
                        'waitTimeMs' => $waitedMs
                    ] + $this->clientLogInfo
                );
            }
        } else {
            $data = $this->store->get( $this->key );
            $indexReached = $data[self::FLD_WRITE_INDEX] ?? null;
            if ( $indexReached ) {
                $this->logger->debug(
                    __METHOD__ . ": found position/timestamp data with index {indexReached}.",
                    [ 'indexReached' => $indexReached ] + $this->clientLogInfo
                );
            }
        }
 
        $this->startupPositionsByMaster = $data ? $data[self::FLD_POSITIONS] : [];
        $this->startupTimestampsByCluster = $data[self::FLD_TIMESTAMPS] ?? [];
    }
 
    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;
    }
}