ChronologyProtector.php

Go to the documentation of this file.

<?php
namespace Wikimedia\Rdbms;

use Psr\Log\LoggerAwareInterface;
use Psr\Log\LoggerInterface;
use Psr\Log\NullLogger;
use Wikimedia\WaitConditionLoop;
use BagOStuff;

class ChronologyProtector implements LoggerAwareInterface {
    protected $store;
    protected $logger;

    protected $key;
    protected $clientId;
    protected $clientLogInfo;
    protected $waitForPosIndex;
    protected $waitForPosStoreTimeout = self::POS_STORE_WAIT_TIMEOUT;
    protected $enabled = true;
    protected $wait = true;

    protected $initialized = false;
    protected $startupPositions = [];
    protected $shutdownPositions = [];
    protected $shutdownTouchDBs = [];

    const POSITION_TTL = 60;
    const POSITION_COOKIE_TTL = 10;
    const POS_STORE_WAIT_TIMEOUT = 5;

    public function __construct( BagOStuff $store, array $client, $posIndex, $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 = $posIndex;

        $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->wait = $enabled;
    }

    public function applySessionReplicationPosition( ILoadBalancer $lb ) {
        if ( !$this->enabled ) {
            return; // disabled
        }

        $masterName = $lb->getServerName( $lb->getWriterIndex() );
        $startupPositions = $this->getStartupMasterPositions();

        $pos = $startupPositions[$masterName] ?? null;
        if ( $pos instanceof DBMasterPos ) {
            $this->logger->debug( __METHOD__ . ": pos for DB '$masterName' set to '$pos'\n" );
            $lb->waitFor( $pos );
        }
    }

    public function storeSessionReplicationPosition( ILoadBalancer $lb ) {
        if ( !$this->enabled ) {
            return; // disabled
        } elseif ( !$lb->hasOrMadeRecentMasterChanges( INF ) ) {
            // Only save the position if writes have been done on the connection
            return;
        }

        $masterName = $lb->getServerName( $lb->getWriterIndex() );
        if ( $lb->hasStreamingReplicaServers() ) {
            $pos = $lb->getReplicaResumePos();
            if ( $pos ) {
                $this->logger->debug( __METHOD__ . ": LB for '$masterName' has pos $pos\n" );
                $this->shutdownPositions[$masterName] = $pos;
            }
        } else {
            $this->logger->debug( __METHOD__ . ": DB '$masterName' touched\n" );
        }
        $this->shutdownTouchDBs[$masterName] = 1;
    }

    public function shutdown( callable $workCallback = null, $mode = 'sync', &$cpIndex = null ) {
        if ( !$this->enabled ) {
            return [];
        }

        $store = $this->store;
        // Some callers might want to know if a user recently touched a DB.
        // These writes do not need to block on all datacenters receiving them.
        foreach ( $this->shutdownTouchDBs as $dbName => $unused ) {
            $store->set(
                $this->getTouchedKey( $this->store, $dbName ),
                microtime( true ),
                $store::TTL_DAY
            );
        }

        if ( $this->shutdownPositions === [] ) {
            $this->logger->debug( __METHOD__ . ": no master positions to save\n" );

            return []; // nothing to save
        }

        $this->logger->debug(
            __METHOD__ . ": saving master pos for " .
            implode( ', ', array_keys( $this->shutdownPositions ) ) . "\n"
        );

        // CP-protected writes should overwhelmingly go to the master datacenter, so use a
        // DC-local lock to merge the values. Use a DC-local get() and a synchronous all-DC
        // set(). This makes it possible for the BagOStuff class to write in parallel to all
        // DCs with one RTT. The use of WRITE_SYNC avoids needing READ_LATEST for the get().
        if ( $store->lock( $this->key, 3 ) ) {
            if ( $workCallback ) {
                // Let the store run the work before blocking on a replication sync barrier.
                // If replication caught up while the work finished, the barrier will be fast.
                $store->addBusyCallback( $workCallback );
            }
            $ok = $store->set(
                $this->key,
                $this->mergePositions(
                    $store->get( $this->key ),
                    $this->shutdownPositions,
                    $cpIndex
                ),
                self::POSITION_TTL,
                ( $mode === 'sync' ) ? $store::WRITE_SYNC : 0
            );
            $store->unlock( $this->key );
        } else {
            $ok = false;
        }

        if ( !$ok ) {
            $cpIndex = null; // nothing saved
            $bouncedPositions = $this->shutdownPositions;
            // Raced out too many times or stash is down
            $this->logger->warning( __METHOD__ . ": failed to save master pos for " .
                implode( ', ', array_keys( $this->shutdownPositions ) ) . "\n"
            );
        } elseif ( $mode === 'sync' &&
            $store->getQoS( $store::ATTR_SYNCWRITES ) < $store::QOS_SYNCWRITES_BE
        ) {
            // Positions may not be in all datacenters, force LBFactory to play it safe
            $this->logger->info( __METHOD__ . ": store may not support synchronous writes." );
            $bouncedPositions = $this->shutdownPositions;
        } else {
            $bouncedPositions = [];
        }

        return $bouncedPositions;
    }

    public function getTouched( $dbName ) {
        return $this->store->get( $this->getTouchedKey( $this->store, $dbName ) );
    }

    private function getTouchedKey( BagOStuff $store, $dbName ) {
        return $store->makeGlobalKey( __CLASS__, 'mtime', $this->clientId, $dbName );
    }

    protected function getStartupMasterPositions() {
        if ( $this->initialized ) {
            return $this->startupPositions;
        }

        $this->initialized = true;
        $this->logger->debug( __METHOD__ . ": client ID is {$this->clientId} (read)\n" );

        if ( $this->wait ) {
            // If there is an expectation to see master 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->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['writeIndex'] ) ) {
                            return WaitConditionLoop::CONDITION_REACHED; // b/c
                        }
                        $indexReached = max( $data['writeIndex'], $indexReached );

                        return ( $data['writeIndex'] >= $this->waitForPosIndex )
                            ? WaitConditionLoop::CONDITION_REACHED
                            : WaitConditionLoop::CONDITION_CONTINUE;
                    },
                    $this->waitForPosStoreTimeout
                );
                $result = $loop->invoke();
                $waitedMs = $loop->getLastWaitTime() * 1e3;

                if ( $result == $loop::CONDITION_REACHED ) {
                    $this->logger->debug(
                        __METHOD__ . ": expected and found position index.",
                        [
                            'cpPosIndex' => $this->waitForPosIndex,
                            'waitTimeMs' => $waitedMs
                        ] + $this->clientLogInfo
                    );
                } else {
                    $this->logger->warning(
                        __METHOD__ . ": expected but failed to find position index.",
                        [
                            'cpPosIndex' => $this->waitForPosIndex,
                            'indexReached' => $indexReached,
                            'waitTimeMs' => $waitedMs
                        ] + $this->clientLogInfo
                    );
                }
            } else {
                $data = $this->store->get( $this->key );
            }

            $this->startupPositions = $data ? $data['positions'] : [];
            $this->logger->debug( __METHOD__ . ": key is {$this->key} (read)\n" );
        } else {
            $this->startupPositions = [];
            $this->logger->debug( __METHOD__ . ": key is {$this->key} (unread)\n" );
        }

        return $this->startupPositions;
    }

    protected function mergePositions( $curValue, array $shutdownPositions, &$cpIndex = null ) {
        $curPositions = $curValue['positions'] ?? [];
        // Use the newest positions for each DB master
        foreach ( $shutdownPositions as $db => $pos ) {
            if (
                !isset( $curPositions[$db] ) ||
                !( $curPositions[$db] instanceof DBMasterPos ) ||
                $pos->asOfTime() > $curPositions[$db]->asOfTime()
            ) {
                $curPositions[$db] = $pos;
            }
        }

        $cpIndex = $curValue['writeIndex'] ?? 0;

        return [
            'positions' => $curPositions,
            'writeIndex' => ++$cpIndex
        ];
    }
}