SqlBagOStuff.php

Go to the documentation of this file.

<?php
use Wikimedia\AtEase\AtEase;
use Wikimedia\ObjectFactory;
use Wikimedia\Rdbms\Blob;
use Wikimedia\Rdbms\Database;
use Wikimedia\Rdbms\DBConnectionError;
use Wikimedia\Rdbms\DBError;
use Wikimedia\Rdbms\DBQueryError;
use Wikimedia\Rdbms\IDatabase;
use Wikimedia\Rdbms\ILoadBalancer;
use Wikimedia\Rdbms\IMaintainableDatabase;
use Wikimedia\ScopedCallback;
use Wikimedia\Timestamp\ConvertibleTimestamp;
use Wikimedia\WaitConditionLoop;
 
class SqlBagOStuff extends MediumSpecificBagOStuff {
    protected $localKeyLb;
    protected $globalKeyLb;
 
    protected $serverInfos = [];
    protected $serverTags = [];
    protected $lastGarbageCollect = 0;
    protected $purgePeriod = 10;
    protected $purgeLimit = 100;
    protected $numTableShards = 1;
    protected $writeBatchSize = 100;
    protected $tableName = 'objectcache';
    protected $replicaOnly;
    protected $multiPrimaryModeType;
 
    protected $conns;
    protected $connFailureTimes = [];
    protected $connFailureErrors = [];
 
    private const SHARD_LOCAL = 'local';
    private const SHARD_GLOBAL = 'global';
 
    private const SAFE_CLOCK_BOUND_SEC = 15;
    private const SAFE_PURGE_DELAY_SEC = 3600;
    private const TOMB_SERIAL = '';
    private const TOMB_EXPTIME = -self::SAFE_CLOCK_BOUND_SEC;
    private const GC_DELAY_SEC = 1;
 
    private const BLOB_VALUE = 0;
    private const BLOB_EXPIRY = 1;
    private const BLOB_CASTOKEN = 2;
 
    private const INF_TIMESTAMP_PLACEHOLDER = '99991231235959';
 
    public function __construct( $params ) {
        parent::__construct( $params );
 
        $dbType = null;
        if ( isset( $params['servers'] ) || isset( $params['server'] ) ) {
            // Configuration uses a direct list of servers.
            // Object data is horizontally partitioned via key hash.
            $index = 0;
            foreach ( ( $params['servers'] ?? [ $params['server'] ] ) as $tag => $info ) {
                $this->serverInfos[$index] = $info;
                // Allow integer-indexes arrays for b/c
                $this->serverTags[$index] = is_string( $tag ) ? $tag : "#$index";
                $dbType = $info['type'];
                ++$index;
            }
        } else {
            // Configuration uses the servers defined in LoadBalancer instances.
            // Object data is vertically partitioned via global vs local keys.
            if ( isset( $params['globalKeyLB'] ) ) {
                $this->globalKeyLb = ( $params['globalKeyLB'] instanceof ILoadBalancer )
                    ? $params['globalKeyLB']
                    : ObjectFactory::getObjectFromSpec( $params['globalKeyLB'] );
            }
            if ( isset( $params['localKeyLB'] ) ) {
                $this->localKeyLb = ( $params['localKeyLB'] instanceof ILoadBalancer )
                    ? $params['localKeyLB']
                    : ObjectFactory::getObjectFromSpec( $params['localKeyLB'] );
            } else {
                $this->localKeyLb = $this->globalKeyLb;
            }
            // When using LoadBalancer instances, one *must* be defined for local keys
            if ( !$this->localKeyLb ) {
                throw new InvalidArgumentException(
                    "Config requires 'server', 'servers', or 'localKeyLB'/'globalKeyLB'"
                );
            }
        }
 
        $this->purgePeriod = intval( $params['purgePeriod'] ?? $this->purgePeriod );
        $this->purgeLimit = intval( $params['purgeLimit'] ?? $this->purgeLimit );
        $this->tableName = $params['tableName'] ?? $this->tableName;
        $this->numTableShards = intval( $params['shards'] ?? $this->numTableShards );
        $this->writeBatchSize = intval( $params['writeBatchSize'] ?? $this->writeBatchSize );
        $this->replicaOnly = $params['replicaOnly'] ?? false;
 
        if ( $params['multiPrimaryMode'] ?? false ) {
            if ( $dbType !== 'mysql' ) {
                throw new InvalidArgumentException( "Multi-primary mode only supports MySQL" );
            }
 
            $this->multiPrimaryModeType = $dbType;
        }
 
        $this->attrMap[self::ATTR_DURABILITY] = self::QOS_DURABILITY_RDBMS;
        $this->attrMap[self::ATTR_EMULATION] = self::QOS_EMULATION_SQL;
    }
 
    protected function doGet( $key, $flags = 0, &$casToken = null ) {
        $getToken = ( $casToken === self::PASS_BY_REF );
        $casToken = null;
 
        $data = $this->fetchBlobs( [ $key ], $getToken )[$key];
        if ( $data ) {
            $result = $this->unserialize( $data[self::BLOB_VALUE] );
            if ( $getToken && $result !== false ) {
                $casToken = $data[self::BLOB_CASTOKEN];
            }
            $valueSize = strlen( $data[self::BLOB_VALUE] );
        } else {
            $result = false;
            $valueSize = false;
        }
 
        $this->updateOpStats( self::METRIC_OP_GET, [ $key => [ null, $valueSize ] ] );
 
        return $result;
    }
 
    protected function doSet( $key, $value, $exptime = 0, $flags = 0 ) {
        $mtime = $this->getCurrentTime();
 
        return $this->modifyBlobs(
            [ $this, 'modifyTableSpecificBlobsForSet' ],
            $mtime,
            [ $key => [ $value, $exptime ] ],
            $flags
        );
    }
 
    protected function doDelete( $key, $flags = 0 ) {
        $mtime = $this->getCurrentTime();
 
        return $this->modifyBlobs(
            [ $this, 'modifyTableSpecificBlobsForDelete' ],
            $mtime,
            [ $key => [] ],
            $flags
        );
    }
 
    protected function doAdd( $key, $value, $exptime = 0, $flags = 0 ) {
        $mtime = $this->newLockingWriteSectionModificationTimestamp( $key, $scope );
        if ( $mtime === null ) {
            // Timeout or I/O error during lock acquisition
            return false;
        }
 
        return $this->modifyBlobs(
            [ $this, 'modifyTableSpecificBlobsForAdd' ],
            $mtime,
            [ $key => [ $value, $exptime ] ],
            $flags
        );
    }
 
    protected function doCas( $casToken, $key, $value, $exptime = 0, $flags = 0 ) {
        $mtime = $this->newLockingWriteSectionModificationTimestamp( $key, $scope );
        if ( $mtime === null ) {
            // Timeout or I/O error during lock acquisition
            return false;
        }
 
        return $this->modifyBlobs(
            [ $this, 'modifyTableSpecificBlobsForCas' ],
            $mtime,
            [ $key => [ $value, $exptime, $casToken ] ],
            $flags
        );
    }
 
    protected function doChangeTTL( $key, $exptime, $flags ) {
        $mtime = $this->getCurrentTime();
 
        return $this->modifyBlobs(
            [ $this, 'modifyTableSpecificBlobsForChangeTTL' ],
            $mtime,
            [ $key => [ $exptime ] ],
            $flags
        );
    }
 
    public function incrWithInit( $key, $exptime, $value = 1, $init = null, $flags = 0 ) {
        $value = (int)$value;
        $init = is_int( $init ) ? $init : $value;
 
        $mtime = $this->getCurrentTime();
 
        $result = $this->modifyBlobs(
            [ $this, 'modifyTableSpecificBlobsForIncrInit' ],
            $mtime,
            [ $key => [ $value, $init, $exptime ] ],
            $flags,
            $resByKey
        ) ? $resByKey[$key] : false;
 
        return $result;
    }
 
    public function incr( $key, $value = 1, $flags = 0 ) {
        return $this->doIncr( $key, $value, $flags );
    }
 
    public function decr( $key, $value = 1, $flags = 0 ) {
        return $this->doIncr( $key, -$value, $flags );
    }
 
    private function doIncr( $key, $value = 1, $flags = 0 ) {
        $mtime = $this->newLockingWriteSectionModificationTimestamp( $key, $scope );
        if ( $mtime === null ) {
            // Timeout or I/O error during lock acquisition
            return false;
        }
 
        $data = $this->fetchBlobs( [ $key ] )[$key];
        if ( $data ) {
            $serialValue = $data[self::BLOB_VALUE];
            if ( $this->isInteger( $serialValue ) ) {
                $newValue = max( (int)$serialValue + (int)$value, 0 );
                $result = $this->modifyBlobs(
                    [ $this, 'modifyTableSpecificBlobsForSet' ],
                    $mtime,
                    // Preserve the old expiry timestamp
                    [ $key => [ $newValue, $data[self::BLOB_EXPIRY] ] ],
                    $flags
                ) ? $newValue : false;
            } else {
                $result = false;
                $this->logger->warning( __METHOD__ . ": $key is a non-integer" );
            }
        } else {
            $result = false;
            $this->logger->debug( __METHOD__ . ": $key does not exists" );
        }
 
        return $result;
    }
 
    protected function doGetMulti( array $keys, $flags = 0 ) {
        $result = [];
        $valueSizeByKey = [];
 
        $dataByKey = $this->fetchBlobs( $keys );
        foreach ( $keys as $key ) {
            $data = $dataByKey[$key];
            if ( $data ) {
                $serialValue = $data[self::BLOB_VALUE];
                $value = $this->unserialize( $serialValue );
                if ( $value !== false ) {
                    $result[$key] = $value;
                }
                $valueSize = strlen( $serialValue );
            } else {
                $valueSize = false;
            }
            $valueSizeByKey[$key] = [ null, $valueSize ];
        }
 
        $this->updateOpStats( self::METRIC_OP_GET, $valueSizeByKey );
 
        return $result;
    }
 
    protected function doSetMulti( array $data, $exptime = 0, $flags = 0 ) {
        $mtime = $this->getCurrentTime();
 
        return $this->modifyBlobs(
            [ $this, 'modifyTableSpecificBlobsForSet' ],
            $mtime,
            array_map(
                static function ( $value ) use ( $exptime ) {
                    return [ $value, $exptime ];
                },
                $data
            ),
            $flags
        );
    }
 
    protected function doDeleteMulti( array $keys, $flags = 0 ) {
        $mtime = $this->getCurrentTime();
 
        return $this->modifyBlobs(
            [ $this, 'modifyTableSpecificBlobsForDelete' ],
            $mtime,
            array_fill_keys( $keys, [] ),
            $flags
        );
    }
 
    public function doChangeTTLMulti( array $keys, $exptime, $flags = 0 ) {
        $mtime = $this->getCurrentTime();
 
        return $this->modifyBlobs(
            [ $this, 'modifyTableSpecificBlobsForChangeTTL' ],
            $mtime,
            array_fill_keys( $keys, [ $exptime ] ),
            $flags
        );
    }
 
    private function getConnection( $shardIndex ) {
        // Don't keep timing out trying to connect if the server is down
        if (
            isset( $this->connFailureErrors[$shardIndex] ) &&
            ( $this->getCurrentTime() - $this->connFailureTimes[$shardIndex] ) < 60
        ) {
            throw $this->connFailureErrors[$shardIndex];
        }
 
        if ( $shardIndex === self::SHARD_LOCAL ) {
            $conn = $this->getConnectionViaLoadBalancer( $shardIndex );
        } elseif ( $shardIndex === self::SHARD_GLOBAL ) {
            $conn = $this->getConnectionViaLoadBalancer( $shardIndex );
        } elseif ( is_int( $shardIndex ) ) {
            if ( isset( $this->serverInfos[$shardIndex] ) ) {
                $server = $this->serverInfos[$shardIndex];
                $conn = $this->getConnectionFromServerInfo( $shardIndex, $server );
            } else {
                throw new UnexpectedValueException( "Invalid server index #$shardIndex" );
            }
        } else {
            throw new UnexpectedValueException( "Invalid server index '$shardIndex'" );
        }
 
        return $conn;
    }
 
    private function getKeyLocation( $key ) {
        if ( $this->serverTags ) {
            // Striped array of database servers
            if ( count( $this->serverTags ) == 1 ) {
                $shardIndex = 0; // short-circuit
            } else {
                $sortedServers = $this->serverTags;
                ArrayUtils::consistentHashSort( $sortedServers, $key );
                reset( $sortedServers );
                $shardIndex = key( $sortedServers );
            }
        } else {
            // LoadBalancer based configuration
            $shardIndex = ( strpos( $key, 'global:' ) === 0 && $this->globalKeyLb )
                ? self::SHARD_GLOBAL
                : self::SHARD_LOCAL;
        }
 
        if ( $this->numTableShards > 1 ) {
            $hash = hexdec( substr( md5( $key ), 0, 8 ) ) & 0x7fffffff;
            $tableIndex = $hash % $this->numTableShards;
        } else {
            $tableIndex = null;
        }
 
        return [ $shardIndex, $this->getTableNameByShard( $tableIndex ) ];
    }
 
    private function getTableNameByShard( $index ) {
        if ( $index !== null && $this->numTableShards > 1 ) {
            $decimals = strlen( $this->numTableShards - 1 );
 
            return $this->tableName . sprintf( "%0{$decimals}d", $index );
        }
 
        return $this->tableName;
    }
 
    private function fetchBlobs( array $keys, bool $getCasToken = false ) {
        $silenceScope = $this->silenceTransactionProfiler();
 
        // Initialize order-preserved per-key results; set values for live keys below
        $dataByKey = array_fill_keys( $keys, null );
 
        $readTime = (int)$this->getCurrentTime();
        $keysByTableByShard = [];
        foreach ( $keys as $key ) {
            list( $shardIndex, $partitionTable ) = $this->getKeyLocation( $key );
            $keysByTableByShard[$shardIndex][$partitionTable][] = $key;
        }
 
        foreach ( $keysByTableByShard as $shardIndex => $serverKeys ) {
            try {
                $db = $this->getConnection( $shardIndex );
                foreach ( $serverKeys as $partitionTable => $tableKeys ) {
                    $res = $db->select(
                        $partitionTable,
                        $getCasToken
                            ? $this->addCasTokenFields( $db, [ 'keyname', 'value', 'exptime' ] )
                            : [ 'keyname', 'value', 'exptime' ],
                        $this->buildExistenceConditions( $db, $tableKeys, $readTime ),
                        __METHOD__
                    );
                    foreach ( $res as $row ) {
                        $row->shardIndex = $shardIndex;
                        $row->tableName = $partitionTable;
                        $dataByKey[$row->keyname] = $row;
                    }
                }
            } catch ( DBError $e ) {
                $this->handleDBError( $e, $shardIndex );
            }
        }
 
        foreach ( $keys as $key ) {
            $row = $dataByKey[$key] ?? null;
            if ( !$row ) {
                continue;
            }
 
            $this->debug( __METHOD__ . ": retrieved $key; expiry time is {$row->exptime}" );
            try {
                $db = $this->getConnection( $row->shardIndex );
                $dataByKey[$key] = [
                    self::BLOB_VALUE => $this->dbDecodeSerialValue( $db, $row->value ),
                    self::BLOB_EXPIRY => $this->decodeDbExpiry( $db, $row->exptime ),
                    self::BLOB_CASTOKEN => $getCasToken
                        ? $this->getCasTokenFromRow( $db, $row )
                        : null
                ];
            } catch ( DBQueryError $e ) {
                $this->handleDBError( $e, $row->shardIndex );
            }
        }
 
        return $dataByKey;
    }
 
    private function modifyBlobs(
        callable $tableWriteCallback,
        float $mtime,
        array $argsByKey,
        int $flags,
        &$resByKey = []
    ) {
        // Initialize order-preserved per-key results; callbacks mark successful results
        $resByKey = array_fill_keys( array_keys( $argsByKey ), false );
 
        $silenceScope = $this->silenceTransactionProfiler();
 
        $argsByKeyByTableByShard = [];
        foreach ( $argsByKey as $key => $args ) {
            list( $shardIndex, $partitionTable ) = $this->getKeyLocation( $key );
            $argsByKeyByTableByShard[$shardIndex][$partitionTable][$key] = $args;
        }
 
        $shardIndexesAffected = [];
        foreach ( $argsByKeyByTableByShard as $shardIndex => $argsByKeyByTables ) {
            foreach ( $argsByKeyByTables as $table => $ptKeyArgs ) {
                try {
                    $db = $this->getConnection( $shardIndex );
                    $shardIndexesAffected[] = $shardIndex;
                    $tableWriteCallback( $db, $table, $mtime, $ptKeyArgs, $resByKey );
                } catch ( DBError $e ) {
                    $this->handleDBError( $e, $shardIndex );
                    continue;
                }
            }
        }
 
        $success = !in_array( false, $resByKey, true );
 
        if ( $this->fieldHasFlags( $flags, self::WRITE_SYNC ) ) {
            foreach ( $shardIndexesAffected as $shardIndex ) {
                if ( !$this->waitForReplication( $shardIndex ) ) {
                    $success = false;
                }
            }
        }
 
        foreach ( $shardIndexesAffected as $shardIndex ) {
            try {
                $db = $this->getConnection( $shardIndex );
                $this->occasionallyGarbageCollect( $db );
            } catch ( DBError $e ) {
                $this->handleDBError( $e, $shardIndex );
            }
        }
 
        return $success;
    }
 
    private function modifyTableSpecificBlobsForSet(
        IDatabase $db,
        string $ptable,
        float $mtime,
        array $argsByKey,
        array &$resByKey
    ) {
        $valueSizesByKey = [];
 
        $mt = $this->makeTimestampedModificationToken( $mtime, $db );
 
        if ( $this->multiPrimaryModeType !== null ) {
            // @TODO: use multi-row upsert() with VALUES() once supported in Database
            foreach ( $argsByKey as $key => list( $value, $exptime ) ) {
                $serialValue = $this->getSerialized( $value, $key );
                $expiry = $this->makeNewKeyExpiry( $exptime, (int)$mtime );
                $db->upsert(
                    $ptable,
                    $this->buildUpsertRow( $db, $key, $serialValue, $expiry, $mt ),
                    [ [ 'keyname' ] ],
                    $this->buildUpsertSetForOverwrite( $db, $serialValue, $expiry, $mt ),
                    __METHOD__
                );
                $resByKey[$key] = true;
 
                $valueSizesByKey[$key] = [ strlen( $serialValue ), null ];
            }
        } else {
            // T288998: use REPLACE, if possible, to avoid cluttering the binlogs
            $rows = [];
            foreach ( $argsByKey as $key => list( $value, $exptime ) ) {
                $expiry = $this->makeNewKeyExpiry( $exptime, (int)$mtime );
                $serialValue = $this->getSerialized( $value, $key );
                $rows[] = $this->buildUpsertRow( $db, $key, $serialValue, $expiry, $mt );
 
                $valueSizesByKey[$key] = [ strlen( $serialValue ), null ];
            }
            $db->replace( $ptable, 'keyname', $rows, __METHOD__ );
            foreach ( $argsByKey as $key => $unused ) {
                $resByKey[$key] = true;
            }
        }
 
        $this->updateOpStats( self::METRIC_OP_SET, $valueSizesByKey );
    }
 
    private function modifyTableSpecificBlobsForDelete(
        IDatabase $db,
        string $ptable,
        float $mtime,
        array $argsByKey,
        array &$resByKey
    ) {
        if ( $this->isMultiPrimaryModeEnabled() ) {
            // Tombstone keys in order to respect eventual consistency
            $mt = $this->makeTimestampedModificationToken( $mtime, $db );
            $expiry = $this->makeNewKeyExpiry( self::TOMB_EXPTIME, (int)$mtime );
            $rows = [];
            foreach ( $argsByKey as $key => $arg ) {
                $rows[] = $this->buildUpsertRow( $db, $key, self::TOMB_SERIAL, $expiry, $mt );
            }
            $db->upsert(
                $ptable,
                $rows,
                [ [ 'keyname' ] ],
                $this->buildUpsertSetForOverwrite( $db, self::TOMB_SERIAL, $expiry, $mt ),
                __METHOD__
            );
        } else {
            // Just purge the keys since there is only one primary (e.g. "source of truth")
            $db->delete( $ptable, [ 'keyname' => array_keys( $argsByKey ) ], __METHOD__ );
        }
 
        foreach ( $argsByKey as $key => $arg ) {
            $resByKey[$key] = true;
        }
 
        $this->updateOpStats( self::METRIC_OP_DELETE, array_keys( $argsByKey ) );
    }
 
    private function modifyTableSpecificBlobsForAdd(
        IDatabase $db,
        string $ptable,
        float $mtime,
        array $argsByKey,
        array &$resByKey
    ) {
        $valueSizesByKey = [];
 
        $mt = $this->makeTimestampedModificationToken( $mtime, $db );
 
        // This check must happen outside the write query to respect eventual consistency
        $existingKeys = $db->selectFieldValues(
            $ptable,
            'keyname',
            $this->buildExistenceConditions( $db, array_keys( $argsByKey ), (int)$mtime ),
            __METHOD__
        );
        $existingByKey = array_fill_keys( $existingKeys, true );
 
        // @TODO: use multi-row upsert() with VALUES() once supported in Database
        foreach ( $argsByKey as $key => list( $value, $exptime ) ) {
            if ( isset( $existingByKey[$key] ) ) {
                $this->logger->debug( __METHOD__ . ": $key already exists" );
                continue;
            }
 
            $serialValue = $this->getSerialized( $value, $key );
            $expiry = $this->makeNewKeyExpiry( $exptime, (int)$mtime );
            $db->upsert(
                $ptable,
                $this->buildUpsertRow( $db, $key, $serialValue, $expiry, $mt ),
                [ [ 'keyname' ] ],
                $this->buildUpsertSetForOverwrite( $db, $serialValue, $expiry, $mt ),
                __METHOD__
            );
            $resByKey[$key] = true;
 
            $valueSizesByKey[$key] = [ strlen( $serialValue ), null ];
        }
 
        $this->updateOpStats( self::METRIC_OP_ADD, $valueSizesByKey );
    }
 
    private function modifyTableSpecificBlobsForCas(
        IDatabase $db,
        string $ptable,
        float $mtime,
        array $argsByKey,
        array &$resByKey
    ) {
        $valueSizesByKey = [];
 
        $mt = $this->makeTimestampedModificationToken( $mtime, $db );
 
        // This check must happen outside the write query to respect eventual consistency
        $res = $db->select(
            $ptable,
            $this->addCasTokenFields( $db, [ 'keyname' ] ),
            $this->buildExistenceConditions( $db, array_keys( $argsByKey ), (int)$mtime ),
            __METHOD__
        );
 
        $curTokensByKey = [];
        foreach ( $res as $row ) {
            $curTokensByKey[$row->keyname] = $this->getCasTokenFromRow( $db, $row );
        }
 
        // @TODO: use multi-row upsert() with VALUES() once supported in Database
        foreach ( $argsByKey as $key => list( $value, $exptime, $casToken ) ) {
            $curToken = $curTokensByKey[$key] ?? null;
            if ( $curToken === null ) {
                $this->logger->debug( __METHOD__ . ": $key does not exists" );
                continue;
            }
 
            if ( $curToken !== $casToken ) {
                $this->logger->debug( __METHOD__ . ": $key does not have a matching token" );
                continue;
            }
 
            $serialValue = $this->getSerialized( $value, $key );
            $expiry = $this->makeNewKeyExpiry( $exptime, (int)$mtime );
            $db->upsert(
                $ptable,
                $this->buildUpsertRow( $db, $key, $serialValue, $expiry, $mt ),
                [ [ 'keyname' ] ],
                $this->buildUpsertSetForOverwrite( $db, $serialValue, $expiry, $mt ),
                __METHOD__
            );
            $resByKey[$key] = true;
 
            $valueSizesByKey[$key] = [ strlen( $serialValue ), null ];
        }
 
        $this->updateOpStats( self::METRIC_OP_CAS, $valueSizesByKey );
    }
 
    private function modifyTableSpecificBlobsForChangeTTL(
        IDatabase $db,
        string $ptable,
        float $mtime,
        array $argsByKey,
        array &$resByKey
    ) {
        if ( $this->isMultiPrimaryModeEnabled() ) {
            $mt = $this->makeTimestampedModificationToken( $mtime, $db );
 
            $res = $db->select(
                $ptable,
                [ 'keyname', 'value' ],
                $this->buildExistenceConditions( $db, array_keys( $argsByKey ), (int)$mtime ),
                __METHOD__
            );
            // @TODO: use multi-row upsert() with VALUES() once supported in Database
            foreach ( $res as $curRow ) {
                $key = $curRow->keyname;
                $serialValue = $this->dbDecodeSerialValue( $db, $curRow->value );
                list( $exptime ) = $argsByKey[$key];
                $expiry = $this->makeNewKeyExpiry( $exptime, (int)$mtime );
 
                $db->upsert(
                    $ptable,
                    $this->buildUpsertRow( $db, $key, $serialValue, $expiry, $mt ),
                    [ [ 'keyname' ] ],
                    $this->buildUpsertSetForOverwrite( $db, $serialValue, $expiry, $mt ),
                    __METHOD__
                );
                $resByKey[$key] = true;
            }
        } else {
            $keysBatchesByExpiry = [];
            foreach ( $argsByKey as $key => list( $exptime ) ) {
                $expiry = $this->makeNewKeyExpiry( $exptime, (int)$mtime );
                $keysBatchesByExpiry[$expiry][] = $key;
            }
 
            $existingCount = 0;
            foreach ( $keysBatchesByExpiry as $expiry => $keyBatch ) {
                $db->update(
                    $ptable,
                    [ 'exptime' => $this->encodeDbExpiry( $db, $expiry ) ],
                    $this->buildExistenceConditions( $db, $keyBatch, (int)$mtime ),
                    __METHOD__
                );
                $existingCount += $db->affectedRows();
            }
            if ( $existingCount === count( $argsByKey ) ) {
                foreach ( $argsByKey as $key => $args ) {
                    $resByKey[$key] = true;
                }
            }
        }
 
        $this->updateOpStats( self::METRIC_OP_CHANGE_TTL, array_keys( $argsByKey ) );
    }
 
    private function modifyTableSpecificBlobsForIncrInit(
        IDatabase $db,
        string $ptable,
        float $mtime,
        array $argsByKey,
        array &$resByKey
    ) {
        foreach ( $argsByKey as $key => list( $step, $init, $exptime ) ) {
            $mt = $this->makeTimestampedModificationToken( $mtime, $db );
            $expiry = $this->makeNewKeyExpiry( $exptime, (int)$mtime );
 
            // Use a transaction so that changes from other threads are not visible due to
            // "consistent reads". This way, the exact post-increment value can be returned.
            // The "live key exists" check can go inside the write query and remain safe for
            // replication since the TTL for such keys is either indefinite or very short.
            $db->startAtomic( __METHOD__ );
            $db->upsert(
                $ptable,
                $this->buildUpsertRow( $db, $key, $init, $expiry, $mt ),
                [ [ 'keyname' ] ],
                $this->buildIncrUpsertSet( $db, $step, $init, $expiry, $mt, (int)$mtime ),
                __METHOD__
            );
            $affectedCount = $db->affectedRows();
            $row = $db->selectRow( $ptable, 'value', [ 'keyname' => $key ], __METHOD__ );
            $db->endAtomic( __METHOD__ );
 
            if ( !$affectedCount || $row === false ) {
                $this->logger->warning( __METHOD__ . ": failed to set new $key value" );
                continue;
            }
 
            $serialValue = $this->dbDecodeSerialValue( $db, $row->value );
            if ( !$this->isInteger( $serialValue ) ) {
                $this->logger->warning( __METHOD__ . ": got non-integer $key value" );
                continue;
            }
 
            $resByKey[$key] = (int)$serialValue;
        }
 
        $this->updateOpStats( self::METRIC_OP_INCR, array_keys( $argsByKey ) );
    }
 
    private function makeNewKeyExpiry( $exptime, int $nowTsUnix ) {
        $expiry = $this->getExpirationAsTimestamp( $exptime );
        // Eventual consistency requires the preservation of recently modified keys.
        // Do not create rows with `exptime` fields so low that they might get garbage
        // collected before being replicated.
        if ( $expiry !== self::TTL_INDEFINITE ) {
            $expiry = max( $expiry, $nowTsUnix - self::SAFE_CLOCK_BOUND_SEC );
        }
 
        return $expiry;
    }
 
    private function newLockingWriteSectionModificationTimestamp( $key, &$scope ) {
        if ( !$this->lock( $key, 0 ) ) {
            return null;
        }
 
        $scope = new ScopedCallback( function () use ( $key ) {
            $this->unlock( $key );
        } );
 
        return sprintf( '%.6f', $this->locks[$key][self::LOCK_TIME] );
    }
 
    private function makeTimestampedModificationToken( $mtime, IDatabase $db ) {
        // We have reserved space for upto 6 digits in the microsecond portion of the token.
        // This is for future use only (maybe CAS tokens) and not currently used.
        // It is currently populated by the microsecond portion returned by microtime,
        // which generally has fewer than 6 digits of meaningful precision but can still be useful
        // in debugging (to see the token continuously change even during rapid testing).
        $seconds = (int)$mtime;
        list( , $microseconds ) = explode( '.', sprintf( '%.6f', $mtime ) );
 
        $id = $db->getTopologyBasedServerId() ?? sprintf( '%u', crc32( $db->getServerName() ) );
 
        $token = implode( '', [
            // 67 bit integral portion of UNIX timestamp, qualified
            \Wikimedia\base_convert(
                // 35 bit integral seconds portion of UNIX timestamp
                str_pad( base_convert( $seconds, 10, 2 ), 35, '0', STR_PAD_LEFT ) .
                // 32 bit ID of the primary database server handling the write
                str_pad( base_convert( $id, 10, 2 ), 32, '0', STR_PAD_LEFT ),
                2,
                36,
                13
            ),
            // 20 bit fractional portion of UNIX timestamp, as integral microseconds
            str_pad( base_convert( $microseconds, 10, 36 ), 4, '0', STR_PAD_LEFT )
        ] );
 
        if ( strlen( $token ) !== 17 ) {
            throw new RuntimeException( "Modification timestamp overflow detected" );
        }
 
        return $token;
    }
 
    private function buildExistenceConditions( IDatabase $db, $keys, string $time ) {
        // Note that tombstones always have past expiration dates
        return [
            'keyname' => $keys,
            'exptime >= ' . $db->addQuotes( $db->timestamp( (int)$time ) )
        ];
    }
 
    private function buildUpsertRow(
        IDatabase $db,
        $key,
        $serialValue,
        int $expiry,
        string $mt
    ) {
        $row = [
            'keyname' => $key,
            'value'   => $this->dbEncodeSerialValue( $db, $serialValue ),
            'exptime' => $this->encodeDbExpiry( $db, $expiry )
        ];
        if ( $this->isMultiPrimaryModeEnabled() ) {
            $row['modtoken'] = $mt;
        }
 
        return $row;
    }
 
    private function buildUpsertSetForOverwrite(
        IDatabase $db,
        $serialValue,
        int $expiry,
        string $mt
    ) {
        $expressionsByColumn = [
            'value'   => $db->addQuotes( $this->dbEncodeSerialValue( $db, $serialValue ) ),
            'exptime' => $db->addQuotes( $this->encodeDbExpiry( $db, $expiry ) )
        ];
 
        $set = [];
        if ( $this->isMultiPrimaryModeEnabled() ) {
            // The query might take a while to replicate, during which newer values might get
            // written. Qualify the query so that it does not override such values. Note that
            // duplicate tokens generated around the same time for a key should still result
            // in convergence given the use of server_id in modtoken (providing a total order
            // among primary DB servers) and MySQL binlog ordering (providing a total order
            // for writes replicating from a given primary DB server).
            $expressionsByColumn['modtoken'] = $db->addQuotes( $mt );
            foreach ( $expressionsByColumn as $column => $updateExpression ) {
                $rhs = $db->conditional(
                    $db->addQuotes( substr( $mt, 0, 13 ) ) . ' >= SUBSTR(modtoken,0,13)',
                    $updateExpression,
                    $column
                );
                $set[] = "{$column}=" . trim( $rhs );
            }
        } else {
            foreach ( $expressionsByColumn as $column => $updateExpression ) {
                $set[] = "{$column}={$updateExpression}";
            }
        }
 
        return $set;
    }
 
    private function buildIncrUpsertSet(
        IDatabase $db,
        int $step,
        int $init,
        int $expiry,
        string $mt,
        int $mtUnixTs
    ) {
        // Map of (column => (SQL for non-expired key rows, SQL for expired key rows))
        $expressionsByColumn = [
            'value'   => [
                $db->buildIntegerCast( 'value' ) . " + {$db->addQuotes( $step )}",
                $db->addQuotes( $this->dbEncodeSerialValue( $db, $init ) )
            ],
            'exptime' => [
                'exptime',
                $db->addQuotes( $this->encodeDbExpiry( $db, $expiry ) )
            ]
        ];
        if ( $this->isMultiPrimaryModeEnabled() ) {
            $expressionsByColumn['modtoken'] = [ 'modtoken', $db->addQuotes( $mt ) ];
        }
 
        $set = [];
        foreach ( $expressionsByColumn as $column => list( $updateExpression, $initExpression ) ) {
            $rhs = $db->conditional(
                'exptime >= ' . $db->addQuotes( $db->timestamp( $mtUnixTs ) ),
                $updateExpression,
                $initExpression
            );
            $set[] = "{$column}=" . trim( $rhs );
        }
 
        return $set;
    }
 
    private function encodeDbExpiry( IDatabase $db, int $expiry ) {
        return ( $expiry === self::TTL_INDEFINITE )
            // Use the maximum timestamp that the column can store
            ? $db->timestamp( self::INF_TIMESTAMP_PLACEHOLDER )
            // Convert the absolute timestamp into the DB timestamp format
            : $db->timestamp( $expiry );
    }
 
    private function decodeDbExpiry( IDatabase $db, string $dbExpiry ) {
        return ( $dbExpiry === $db->timestamp( self::INF_TIMESTAMP_PLACEHOLDER ) )
            ? self::TTL_INDEFINITE
            : (int)ConvertibleTimestamp::convert( TS_UNIX, $dbExpiry );
    }
 
    private function dbEncodeSerialValue( IDatabase $db, $serialValue ) {
        return is_int( $serialValue ) ? $serialValue : $db->encodeBlob( $serialValue );
    }
 
    private function dbDecodeSerialValue( IDatabase $db, $blob ) {
        return $this->isInteger( $blob ) ? (int)$blob : $db->decodeBlob( $blob );
    }
 
    private function addCasTokenFields( IDatabase $db, array $fields ) {
        $type = $db->getType();
 
        if ( $type === 'mysql' ) {
            $fields['castoken'] = $db->buildConcat( [
                'SHA1(value)',
                $db->addQuotes( '@' ),
                'exptime'
            ] );
        } elseif ( $type === 'postgres' ) {
            $fields['castoken'] = $db->buildConcat( [
                'md5(value)',
                $db->addQuotes( '@' ),
                'exptime'
            ] );
        } else {
            if ( !in_array( 'value', $fields, true ) ) {
                $fields[] = 'value';
            }
            if ( !in_array( 'exptime', $fields, true ) ) {
                $fields[] = 'exptime';
            }
        }
 
        return $fields;
    }
 
    private function getCasTokenFromRow( IDatabase $db, stdClass $row ) {
        if ( isset( $row->castoken ) ) {
            $token = $row->castoken;
        } else {
            $token = sha1( $this->dbDecodeSerialValue( $db, $row->value ) ) . '@' . $row->exptime;
            $this->logger->debug( __METHOD__ . ": application computed hash for CAS token" );
        }
 
        return $token;
    }
 
    private function occasionallyGarbageCollect( IDatabase $db ) {
        if (
            // Random purging is enabled
            $this->purgePeriod &&
            // Only purge on one in every $this->purgePeriod writes
            mt_rand( 0, $this->purgePeriod - 1 ) == 0 &&
            // Avoid repeating the delete within a few seconds
            ( $this->getCurrentTime() - $this->lastGarbageCollect ) > self::GC_DELAY_SEC
        ) {
            $garbageCollector = function () use ( $db ) {
                $this->deleteServerObjectsExpiringBefore(
                    $db,
                    (int)$this->getCurrentTime(),
                    $this->purgeLimit
                );
                $this->lastGarbageCollect = time();
            };
            if ( $this->asyncHandler ) {
                $this->lastGarbageCollect = $this->getCurrentTime(); // avoid duplicate enqueues
                ( $this->asyncHandler )( $garbageCollector );
            } else {
                $garbageCollector();
            }
        }
    }
 
    public function expireAll() {
        $this->deleteObjectsExpiringBefore( (int)$this->getCurrentTime() );
    }
 
    public function deleteObjectsExpiringBefore(
        $timestamp,
        callable $progress = null,
        $limit = INF,
        string $tag = null
    ) {
        $silenceScope = $this->silenceTransactionProfiler();
 
        if ( $tag !== null ) {
            // Purge one server only, to support concurrent purging in large wiki farms (T282761).
            $shardIndexes = [ $this->getShardServerIndexForTag( $tag ) ];
        } else {
            $shardIndexes = $this->getShardServerIndexes();
            shuffle( $shardIndexes );
        }
 
        $ok = true;
        $numServers = count( $shardIndexes );
 
        $keysDeletedCount = 0;
        foreach ( $shardIndexes as $numServersDone => $shardIndex ) {
            try {
                $db = $this->getConnection( $shardIndex );
                $this->deleteServerObjectsExpiringBefore(
                    $db,
                    $timestamp,
                    $limit,
                    $keysDeletedCount,
                    [ 'fn' => $progress, 'serversDone' => $numServersDone, 'serversTotal' => $numServers ]
                );
            } catch ( DBError $e ) {
                $this->handleDBError( $e, $shardIndex );
                $ok = false;
            }
        }
 
        return $ok;
    }
 
    private function deleteServerObjectsExpiringBefore(
        IDatabase $db,
        $timestamp,
        $limit,
        &$keysDeletedCount = 0,
        array $progress = null
    ) {
        $cutoffUnix = ConvertibleTimestamp::convert( TS_UNIX, $timestamp );
        if ( $this->isMultiPrimaryModeEnabled() ) {
            // Eventual consistency requires the preservation of any key that was recently
            // modified. The key must exist on this database server long enough for the server
            // to receive, via replication, all writes to the key with lower timestamps. Such
            // writes should be no-ops since the existing key value should "win". If the network
            // partitions between datacenters A and B for 30 minutes, the database servers in
            // each datacenter will see an initial burst of writes with "old" timestamps via
            // replication. This might include writes with lower timestamps that the existing
            // key value. Therefore, clock skew and replication delay are both factors.
            $cutoffUnixSafe = (int)$this->getCurrentTime() - self::SAFE_PURGE_DELAY_SEC;
            $cutoffUnix = min( $cutoffUnix, $cutoffUnixSafe );
        }
        $tableIndexes = range( 0, $this->numTableShards - 1 );
        shuffle( $tableIndexes );
 
        $batchSize = min( $this->writeBatchSize, $limit );
 
        foreach ( $tableIndexes as $numShardsDone => $tableIndex ) {
            // The oldest expiry of a row we have deleted on this shard
            // (the first row that we deleted)
            $minExpUnix = null;
            // The most recent expiry time so far, from a row we have deleted on this shard
            $maxExp = null;
            // Size of the time range we'll delete, in seconds (for progress estimate)
            $totalSeconds = null;
 
            do {
                $res = $db->select(
                    $this->getTableNameByShard( $tableIndex ),
                    [ 'keyname', 'exptime' ],
                    array_merge(
                        [ 'exptime < ' . $db->addQuotes( $db->timestamp( $cutoffUnix ) ) ],
                        $maxExp ? [ 'exptime >= ' . $db->addQuotes( $maxExp ) ] : []
                    ),
                    __METHOD__,
                    [ 'LIMIT' => $batchSize, 'ORDER BY' => 'exptime ASC' ]
                );
 
                if ( $res->numRows() ) {
                    $row = $res->current();
                    if ( $minExpUnix === null ) {
                        $minExpUnix = ConvertibleTimestamp::convert( TS_UNIX, $row->exptime );
                        $totalSeconds = max( $cutoffUnix - $minExpUnix, 1 );
                    }
 
                    $keys = [];
                    foreach ( $res as $row ) {
                        $keys[] = $row->keyname;
                        $maxExp = $row->exptime;
                    }
 
                    $db->delete(
                        $this->getTableNameByShard( $tableIndex ),
                        [
                            'exptime < ' . $db->addQuotes( $db->timestamp( $cutoffUnix ) ),
                            'keyname' => $keys
                        ],
                        __METHOD__
                    );
                    $keysDeletedCount += $db->affectedRows();
                }
 
                if ( $progress && is_callable( $progress['fn'] ) ) {
                    if ( $totalSeconds ) {
                        $maxExpUnix = ConvertibleTimestamp::convert( TS_UNIX, $maxExp );
                        $remainingSeconds = $cutoffUnix - $maxExpUnix;
                        $processedSeconds = max( $totalSeconds - $remainingSeconds, 0 );
                        // For example, if we've done 1.5 table shard, and are thus half-way on the
                        // 2nd of perhaps 5 tables on this server, then this might be:
                        // `( 1 + ( 43200 / 86400 ) ) / 5 = 0.3`, or 30% done, of tables on this server.
                        $tablesDoneRatio =
                            ( $numShardsDone + ( $processedSeconds / $totalSeconds ) ) / $this->numTableShards;
                    } else {
                        $tablesDoneRatio = 1;
                    }
 
                    // For example, if we're 30% done on the last of 10 servers, then this might be:
                    // `( 9 / 10 ) + ( 0.3 / 10 ) = 0.93`, or 93% done, overall.
                    $overallRatio = ( $progress['serversDone'] / $progress['serversTotal'] ) +
                        ( $tablesDoneRatio / $progress['serversTotal'] );
                    ( $progress['fn'] )( $overallRatio * 100 );
                }
            } while ( $res->numRows() && $keysDeletedCount < $limit );
        }
    }
 
    public function deleteAll() {
        $silenceScope = $this->silenceTransactionProfiler();
        foreach ( $this->getShardServerIndexes() as $shardIndex ) {
            $db = null; // in case of connection failure
            try {
                $db = $this->getConnection( $shardIndex );
                for ( $i = 0; $i < $this->numTableShards; $i++ ) {
                    $db->delete( $this->getTableNameByShard( $i ), '*', __METHOD__ );
                }
            } catch ( DBError $e ) {
                $this->handleDBError( $e, $shardIndex );
                return false;
            }
        }
        return true;
    }
 
    public function doLock( $key, $timeout = 6, $exptime = 6 ) {
        $silenceScope = $this->silenceTransactionProfiler();
 
        $lockTsUnix = null;
 
        list( $shardIndex ) = $this->getKeyLocation( $key );
        try {
            $db = $this->getConnection( $shardIndex );
            $lockTsUnix = $db->lock( $key, __METHOD__, $timeout, $db::LOCK_TIMESTAMP );
        } catch ( DBError $e ) {
            $this->handleDBError( $e, $shardIndex );
            $this->logger->warning(
                __METHOD__ . ' failed due to I/O error for {key}.',
                [ 'key' => $key ]
            );
        }
 
        return $lockTsUnix;
    }
 
    public function doUnlock( $key ) {
        $silenceScope = $this->silenceTransactionProfiler();
 
        list( $shardIndex ) = $this->getKeyLocation( $key );
 
        try {
            $db = $this->getConnection( $shardIndex );
            $released = $db->unlock( $key, __METHOD__ );
        } catch ( DBError $e ) {
            $this->handleDBError( $e, $shardIndex );
            $released = false;
        }
 
        return $released;
    }
 
    public function makeKeyInternal( $keyspace, $components ) {
        // SQL schema for 'objectcache' specifies keys as varchar(255). From that,
        // subtract the number of characters we need for the keyspace and for
        // the separator character needed for each argument. To handle some
        // custom prefixes used by thing like WANObjectCache, limit to 205.
        $keyspace = strtr( $keyspace, ' ', '_' );
        $charsLeft = 205 - strlen( $keyspace ) - count( $components );
        foreach ( $components as &$component ) {
            $component = strtr( $component, [
                ' ' => '_', // Avoid unnecessary misses from pre-1.35 code
                ':' => '%3A',
            ] );
 
            // 33 = 32 characters for the MD5 + 1 for the '#' prefix.
            if ( $charsLeft > 33 && strlen( $component ) > $charsLeft ) {
                $component = '#' . md5( $component );
            }
            $charsLeft -= strlen( $component );
        }
 
        if ( $charsLeft < 0 ) {
            return $keyspace . ':BagOStuff-long-key:##' . md5( implode( ':', $components ) );
        }
        return $keyspace . ':' . implode( ':', $components );
    }
 
    protected function serialize( $value ) {
        if ( is_int( $value ) ) {
            return $value;
        }
 
        $serial = serialize( $value );
        if ( function_exists( 'gzdeflate' ) ) {
            // On typical message and page data, this can provide a 3X storage savings
            $serial = gzdeflate( $serial );
        }
 
        return $serial;
    }
 
    protected function unserialize( $value ) {
        if ( $value === self::TOMB_SERIAL ) {
            return false; // tombstone
        }
 
        if ( $this->isInteger( $value ) ) {
            return (int)$value;
        }
 
        if ( function_exists( 'gzinflate' ) ) {
            AtEase::suppressWarnings();
            $decompressed = gzinflate( $value );
            AtEase::restoreWarnings();
 
            if ( $decompressed !== false ) {
                $value = $decompressed;
            }
        }
 
        return unserialize( $value );
    }
 
    private function getConnectionViaLoadBalancer( $shardIndex ) {
        $lb = ( $shardIndex === self::SHARD_LOCAL ) ? $this->localKeyLb : $this->globalKeyLb;
        if ( $lb->getServerAttributes( $lb->getWriterIndex() )[Database::ATTR_DB_LEVEL_LOCKING] ) {
            // Use the main connection to avoid transaction deadlocks
            $conn = $lb->getMaintenanceConnectionRef( DB_PRIMARY );
        } else {
            // If the RDBMs has row/table/page level locking, then use separate auto-commit
            // connection to avoid needless contention and deadlocks.
            $conn = $lb->getMaintenanceConnectionRef(
                $this->replicaOnly ? DB_REPLICA : DB_PRIMARY, [],
                false,
                $lb::CONN_TRX_AUTOCOMMIT
            );
        }
 
        return $conn;
    }
 
    private function getConnectionFromServerInfo( $shardIndex, array $server ) {
        if ( !isset( $this->conns[$shardIndex] ) ) {
            $conn = Database::factory(
                $server['type'],
                array_merge(
                    $server,
                    [
                        // Make sure the handle uses autocommit mode
                        'flags' => ( $server['flags'] ?? 0 ) & ~IDatabase::DBO_TRX,
                        'connLogger' => $this->logger,
                        'queryLogger' => $this->logger
                    ]
                )
            );
            // Automatically create the objectcache table for sqlite as needed
            if ( $conn->getType() === 'sqlite' ) {
                if ( !$conn->tableExists( 'objectcache', __METHOD__ ) ) {
                    $this->initSqliteDatabase( $conn );
                }
            }
            $this->conns[$shardIndex] = $conn;
        }
 
        return $this->conns[$shardIndex];
    }
 
    private function handleDBError( DBError $exception, $shardIndex ) {
        if ( $exception instanceof DBConnectionError ) {
            $this->markServerDown( $exception, $shardIndex );
        }
 
        $this->setAndLogDBError( $exception );
    }
 
    private function setAndLogDBError( DBError $exception ) {
        $this->logger->error( "DBError: {$exception->getMessage()}" );
        if ( $exception instanceof DBConnectionError ) {
            $this->setLastError( self::ERR_UNREACHABLE );
            $this->logger->warning( __METHOD__ . ": ignoring connection error" );
        } else {
            $this->setLastError( self::ERR_UNEXPECTED );
            $this->logger->warning( __METHOD__ . ": ignoring query error" );
        }
    }
 
    private function markServerDown( DBError $exception, $shardIndex ) {
        unset( $this->conns[$shardIndex] ); // bug T103435
 
        $now = $this->getCurrentTime();
        if ( isset( $this->connFailureTimes[$shardIndex] ) ) {
            if ( $now - $this->connFailureTimes[$shardIndex] >= 60 ) {
                unset( $this->connFailureTimes[$shardIndex] );
                unset( $this->connFailureErrors[$shardIndex] );
            } else {
                $this->logger->debug( __METHOD__ . ": Server #$shardIndex already down" );
                return;
            }
        }
        $this->logger->info( __METHOD__ . ": Server #$shardIndex down until " . ( $now + 60 ) );
        $this->connFailureTimes[$shardIndex] = $now;
        $this->connFailureErrors[$shardIndex] = $exception;
    }
 
    private function initSqliteDatabase( IMaintainableDatabase $db ) {
        if ( $db->tableExists( 'objectcache', __METHOD__ ) ) {
            return;
        }
        // Use one table for SQLite; sharding does not seem to have much benefit
        $db->query( "PRAGMA journal_mode=WAL", __METHOD__ ); // this is permanent
        $db->startAtomic( __METHOD__ ); // atomic DDL
        try {
            $encTable = $db->tableName( 'objectcache' );
            $encExptimeIndex = $db->addIdentifierQuotes( $db->tablePrefix() . 'exptime' );
            $db->query(
                "CREATE TABLE $encTable (\n" .
                "   keyname BLOB NOT NULL default '' PRIMARY KEY,\n" .
                "   value BLOB,\n" .
                "   exptime BLOB NOT NULL\n" .
                ")",
                __METHOD__
            );
            $db->query( "CREATE INDEX $encExptimeIndex ON $encTable (exptime)", __METHOD__ );
            $db->endAtomic( __METHOD__ );
        } catch ( DBError $e ) {
            $db->rollback( __METHOD__ );
            throw $e;
        }
    }
 
    public function createTables() {
        foreach ( $this->getShardServerIndexes() as $shardIndex ) {
            $db = $this->getConnection( $shardIndex );
            if ( in_array( $db->getType(), [ 'mysql', 'postgres' ], true ) ) {
                for ( $i = 0; $i < $this->numTableShards; $i++ ) {
                    $encBaseTable = $db->tableName( 'objectcache' );
                    $encShardTable = $db->tableName( $this->getTableNameByShard( $i ) );
                    $db->query( "CREATE TABLE $encShardTable LIKE $encBaseTable", __METHOD__ );
                }
            }
        }
    }
 
    private function getShardServerIndexes() {
        if ( $this->serverTags ) {
            // Striped array of database servers
            $shardIndexes = array_keys( $this->serverTags );
        } else {
            // LoadBalancer based configuration
            $shardIndexes = [];
            if ( $this->localKeyLb ) {
                $shardIndexes[] = self::SHARD_LOCAL;
            }
            if ( $this->globalKeyLb ) {
                $shardIndexes[] = self::SHARD_GLOBAL;
            }
        }
 
        return $shardIndexes;
    }
 
    private function getShardServerIndexForTag( string $tag ) {
        if ( !$this->serverTags ) {
            throw new InvalidArgumentException( "Given a tag but no tags are configured" );
        }
        foreach ( $this->serverTags as $serverShardIndex => $serverTag ) {
            if ( $tag === $serverTag ) {
                return $serverShardIndex;
            }
        }
        throw new InvalidArgumentException( "Unknown server tag: $tag" );
    }
 
    private function isMultiPrimaryModeEnabled() {
        return ( $this->multiPrimaryModeType !== null );
    }
 
    private function waitForReplication( $shardIndex ) {
        if ( is_int( $shardIndex ) ) {
            return true; // striped only, no LoadBalancer
        }
 
        $lb = ( $shardIndex === self::SHARD_LOCAL ) ? $this->localKeyLb : $this->globalKeyLb;
        if ( !$lb->hasStreamingReplicaServers() ) {
            return true;
        }
 
        try {
            // Wait for any replica DBs to catch up
            $masterPos = $lb->getPrimaryPos();
            if ( !$masterPos ) {
                return true; // not applicable
            }
 
            $loop = new WaitConditionLoop(
                static function () use ( $lb, $masterPos ) {
                    return $lb->waitForAll( $masterPos, 1 );
                },
                $this->syncTimeout,
                $this->busyCallbacks
            );
 
            return ( $loop->invoke() === $loop::CONDITION_REACHED );
        } catch ( DBError $e ) {
            $this->setAndLogDBError( $e );
 
            return false;
        }
    }
 
    private function silenceTransactionProfiler() {
        if ( $this->serverInfos ) {
            return null; // no TransactionProfiler injected anyway
        }
        return Profiler::instance()->getTransactionProfiler()->silenceForScope();
    }
}