SqlBagOStuff.php

Go to the documentation of this file.

<?php
use MediaWiki\MediaWikiServices;
use Wikimedia\ArrayUtils\ArrayUtils;
use Wikimedia\ObjectCache\MediumSpecificBagOStuff;
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\Rdbms\RawSQLValue;
use Wikimedia\Rdbms\SelectQueryBuilder;
use Wikimedia\Rdbms\ServerInfo;
use Wikimedia\ScopedCallback;
use Wikimedia\Timestamp\ConvertibleTimestamp;
use Wikimedia\Timestamp\TimestampFormat as TS;
 
class SqlBagOStuff extends MediumSpecificBagOStuff {
    protected $loadBalancerCallback;
    protected $loadBalancer;
    protected $dbDomain;
    protected $useLB = false;
 
    protected $serverInfos = [];
    protected $serverTags = [];
    protected $lastGarbageCollect = 0;
    protected $purgePeriod = 10;
    protected $purgeLimit = 100;
    protected $numTableShards = 1;
    protected $writeBatchSize = 100;
    protected $tableName = 'objectcache';
 
    protected $conns;
    protected $connFailureTimes = [];
    protected $connFailureErrors = [];
 
    private $hasZlib;
 
    private $dataRedundancy;
 
    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 );
 
        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";
                ++$index;
            }
        } elseif ( isset( $params['loadBalancerCallback'] ) ) {
            $this->loadBalancerCallback = $params['loadBalancerCallback'];
            if ( !isset( $params['dbDomain'] ) ) {
                throw new InvalidArgumentException(
                    __METHOD__ . ": 'dbDomain' is required if 'loadBalancerCallback' is given"
                );
            }
            $this->dbDomain = $params['dbDomain'];
            $this->useLB = true;
        } else {
            throw new InvalidArgumentException(
                __METHOD__ . " requires 'server', 'servers', or 'loadBalancerCallback'"
            );
        }
 
        $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->dataRedundancy = min( intval( $params['dataRedundancy'] ?? 1 ), count( $this->serverTags ) );
 
        $this->attrMap[self::ATTR_DURABILITY] = self::QOS_DURABILITY_RDBMS;
 
        $this->hasZlib = extension_loaded( 'zlib' );
    }
 
    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 => [ 0, $valueSize ] ] );
 
        return $result;
    }
 
    protected function doSet( $key, $value, $exptime = 0, $flags = 0 ) {
        $mtime = $this->getCurrentTime();
 
        return $this->modifyBlobs(
            $this->modifyTableSpecificBlobsForSet( ... ),
            $mtime,
            [ $key => [ $value, $exptime ] ]
        );
    }
 
    protected function doDelete( $key, $flags = 0 ) {
        $mtime = $this->getCurrentTime();
 
        return $this->modifyBlobs(
            $this->modifyTableSpecificBlobsForDelete( ... ),
            $mtime,
            [ $key => [] ]
        );
    }
 
    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 ] ]
        );
    }
 
    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 ] ]
        );
    }
 
    protected function doChangeTTL( $key, $exptime, $flags ) {
        $mtime = $this->getCurrentTime();
 
        return $this->modifyBlobs(
            $this->modifyTableSpecificBlobsForChangeTTL( ... ),
            $mtime,
            [ $key => [ $exptime ] ]
        );
    }
 
    protected function doIncrWithInit( $key, $exptime, $step, $init, $flags ) {
        $mtime = $this->getCurrentTime();
 
        if ( $flags & self::WRITE_BACKGROUND ) {
            $callback = $this->modifyTableSpecificBlobsForIncrInitAsync( ... );
        } else {
            $callback = $this->modifyTableSpecificBlobsForIncrInit( ... );
        }
 
        $result = $this->modifyBlobs(
            $callback,
            $mtime,
            [ $key => [ $step, $init, $exptime ] ],
            $resByKey
        ) ? $resByKey[$key] : false;
 
        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] = [ 0, $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
            )
        );
    }
 
    protected function doDeleteMulti( array $keys, $flags = 0 ) {
        $mtime = $this->getCurrentTime();
 
        return $this->modifyBlobs(
            $this->modifyTableSpecificBlobsForDelete( ... ),
            $mtime,
            array_fill_keys( $keys, [] )
        );
    }
 
    public function doChangeTTLMulti( array $keys, $exptime, $flags = 0 ) {
        $mtime = $this->getCurrentTime();
 
        return $this->modifyBlobs(
            $this->modifyTableSpecificBlobsForChangeTTL( ... ),
            $mtime,
            array_fill_keys( $keys, [ $exptime ] )
        );
    }
 
    private function getConnection( $shardIndex ) {
        if ( $this->useLB ) {
            return $this->getConnectionViaLoadBalancer();
        }
 
        // 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 ( isset( $this->serverInfos[$shardIndex] ) ) {
            $server = $this->serverInfos[$shardIndex];
            $conn = $this->getConnectionFromServerInfo( $shardIndex, $server );
        } else {
            throw new UnexpectedValueException( "Invalid server index #$shardIndex" );
        }
 
        return $conn;
    }
 
    private function getShardIndexesForKey( $key, $fallback = false ) {
        // Pick the same shard for sister keys
        // Using the same hash stop as mc-router for consistency
        if ( str_contains( $key, '|#|' ) ) {
            $key = explode( '|#|', $key )[0];
        }
        if ( $this->useLB ) {
            // LoadBalancer based configuration
            return [ 0 ];
        } else {
            // Striped array of database servers
            if ( count( $this->serverTags ) == 1 ) {
                return [ 0 ];
            } else {
                $sortedServers = $this->serverTags;
                // shuffle the servers based on hashing of the keys
                ArrayUtils::consistentHashSort( $sortedServers, $key );
                $shardIndexes = array_keys( $sortedServers );
                if ( $this->dataRedundancy === 1 ) {
                    if ( $fallback ) {
                        return [ $shardIndexes[1] ];
                    } else {
                        return [ $shardIndexes[0] ];
                    }
                } else {
                    return array_slice( $shardIndexes, 0, $this->dataRedundancy );
                }
            }
        }
    }
 
    private function getTableNameForKey( $key ) {
        // Pick the same shard for sister keys
        // Using the same hash stop as mc-router for consistency
        if ( str_contains( $key, '|#|' ) ) {
            $key = explode( '|#|', $key )[0];
        }
 
        if ( $this->numTableShards > 1 ) {
            $hash = hexdec( substr( md5( $key ), 0, 8 ) ) & 0x7fffffff;
            $tableIndex = $hash % $this->numTableShards;
        } else {
            $tableIndex = null;
        }
 
        return $this->getTableNameByShard( $tableIndex );
    }
 
    private function getTableNameByShard( $index ) {
        if ( $index !== null && $this->numTableShards > 1 ) {
            $decimals = strlen( (string)( $this->numTableShards - 1 ) );
 
            return $this->tableName . sprintf( "%0{$decimals}d", $index );
        }
 
        return $this->tableName;
    }
 
    private function fetchBlobs( array $keys, bool $getCasToken = false, $fallback = false ) {
        if ( $fallback && count( $this->serverTags ) > 1 && $this->dataRedundancy > 1 ) {
            // fallback doesn't work with data redundancy
            return [];
        }
 
        $silenceScope = $this->silenceTransactionProfiler();
 
        // Initialize order-preserved per-key results; set values for live keys below
        $dataByKey = array_fill_keys( $keys, null );
        $dataByKeyAndShard = [];
 
        $readTime = (int)$this->getCurrentTime();
        $keysByTableByShard = [];
        foreach ( $keys as $key ) {
            $partitionTable = $this->getTableNameForKey( $key );
            $shardIndexes = $this->getShardIndexesForKey( $key, $fallback );
            foreach ( $shardIndexes as $shardIndex ) {
                $keysByTableByShard[$shardIndex][$partitionTable][] = $key;
            }
        }
        $fallbackResult = [];
 
        foreach ( $keysByTableByShard as $shardIndex => $serverKeys ) {
            try {
                $db = $this->getConnection( $shardIndex );
                foreach ( $serverKeys as $partitionTable => $tableKeys ) {
                    $res = $db->newSelectQueryBuilder()
                        ->select(
                            $getCasToken
                            ? $this->addCasTokenFields( $db, [ 'keyname', 'value', 'exptime' ] )
                            : [ 'keyname', 'value', 'exptime' ] )
                        ->from( $partitionTable )
                        ->where( $this->buildExistenceConditions( $db, $tableKeys, $readTime ) )
                        ->caller( __METHOD__ )
                        ->fetchResultSet();
                    foreach ( $res as $row ) {
                        $row->shardIndex = $shardIndex;
                        $row->tableName = $partitionTable;
                        $dataByKeyAndShard[$row->keyname][$shardIndex] = $row;
                    }
                }
            } catch ( DBError $e ) {
                if ( $fallback ) {
                    $this->handleDBError( $e, $shardIndex );
                } else {
                    $fallbackResult += $this->fetchBlobs(
                        array_merge( ...array_values( $serverKeys ) ),
                        $getCasToken,
                        true
                    );
                }
 
            }
        }
        foreach ( $keys as $key ) {
            $rowsByShard = $dataByKeyAndShard[$key] ?? null;
            if ( !$rowsByShard ) {
                continue;
            }
 
            // One response, no point of consistency checks
            if ( count( $rowsByShard ) == 1 ) {
                $row = array_values( $rowsByShard )[0];
            } else {
                usort( $rowsByShard, static function ( $a, $b ) {
                    if ( $a->exptime == $b->exptime ) {
                        return 0;
                    }
                    return ( $a->exptime < $b->exptime ) ? 1 : -1;
                } );
                $row = array_values( $rowsByShard )[0];
            }
 
            $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 array_merge( $dataByKey, $fallbackResult );
    }
 
    private function modifyBlobs(
        callable $tableWriteCallback,
        float $mtime,
        array $argsByKey,
        &$resByKey = [],
        $fallback = false
    ) {
        if ( $fallback && count( $this->serverTags ) > 1 && $this->dataRedundancy > 1 ) {
            // fallback doesn't work with data redundancy
            return false;
        }
        // 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 ) {
            $partitionTable = $this->getTableNameForKey( $key );
            $shardIndexes = $this->getShardIndexesForKey( $key, $fallback );
            foreach ( $shardIndexes as $shardIndex ) {
                $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 ) {
                    if ( $fallback ) {
                        $this->handleDBError( $e, $shardIndex );
                        continue;
                    } else {
                        $this->modifyBlobs( $tableWriteCallback, $mtime, $ptKeyArgs, $resByKey, true );
                    }
                }
            }
        }
 
        $success = !in_array( false, $resByKey, true );
 
        foreach ( $shardIndexesAffected as $shardIndex ) {
            try {
                if (
                    // Random purging is enabled
                    $this->purgePeriod >= 1 &&
                    // Only purge on one in every $this->purgePeriod writes
                    mt_rand( 1, $this->purgePeriod ) == 1 &&
                    // Avoid repeating the delete within a few seconds
                    ( $this->getCurrentTime() - $this->lastGarbageCollect ) > self::GC_DELAY_SEC
                ) {
                    $this->garbageCollect( $shardIndex );
                }
            } catch ( DBError $e ) {
                $this->handleDBError( $e, $shardIndex );
            }
        }
 
        return $success;
    }
 
    private function modifyTableSpecificBlobsForSet(
        IDatabase $db,
        string $ptable,
        float $mtime,
        array $argsByKey,
        array &$resByKey
    ) {
        $valueSizesByKey = [];
 
        $rows = [];
        foreach ( $argsByKey as $key => [ $value, $exptime ] ) {
            $expiry = $this->makeNewKeyExpiry( $exptime, (int)$mtime );
            $serialValue = $this->getSerialized( $value, $key );
            $rows[] = $this->buildUpsertRow( $db, $key, $serialValue, $expiry );
 
            $valueSizesByKey[$key] = [ strlen( $serialValue ), 0 ];
        }
 
        // T288998: use REPLACE, if possible, to avoid cluttering the binlogs
        $db->newReplaceQueryBuilder()
            ->replaceInto( $ptable )
            ->rows( $rows )
            ->uniqueIndexFields( [ 'keyname' ] )
            ->caller( __METHOD__ )->execute();
 
        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
    ) {
        // Just purge the keys since there is only one primary (e.g. "source of truth")
        $db->newDeleteQueryBuilder()
            ->deleteFrom( $ptable )
            ->where( [ 'keyname' => array_keys( $argsByKey ) ] )
            ->caller( __METHOD__ )->execute();
 
        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 = [];
 
        // This check must happen outside the write query to respect eventual consistency
        $existingKeys = $db->newSelectQueryBuilder()
            ->select( 'keyname' )
            ->from( $ptable )
            ->where( $this->buildExistenceConditions( $db, array_keys( $argsByKey ), (int)$mtime ) )
            ->caller( __METHOD__ )
            ->fetchFieldValues();
        $existingByKey = array_fill_keys( $existingKeys, true );
 
        $rows = [];
        foreach ( $argsByKey as $key => [ $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 );
            $valueSizesByKey[$key] = [ strlen( $serialValue ), 0 ];
            $rows[] = $this->buildUpsertRow( $db, $key, $serialValue, $expiry );
        }
        if ( !$rows ) {
            return;
        }
        $db->newInsertQueryBuilder()
            ->insertInto( $ptable )
            ->rows( $rows )
            ->onDuplicateKeyUpdate()
            ->uniqueIndexFields( [ 'keyname' ] )
            ->set( $this->buildMultiUpsertSetForOverwrite( $db ) )
            ->caller( __METHOD__ )->execute();
 
        foreach ( $argsByKey as $key => $unused ) {
            $resByKey[$key] = !isset( $existingByKey[$key] );
        }
 
        $this->updateOpStats( self::METRIC_OP_ADD, $valueSizesByKey );
    }
 
    private function modifyTableSpecificBlobsForCas(
        IDatabase $db,
        string $ptable,
        float $mtime,
        array $argsByKey,
        array &$resByKey
    ) {
        $valueSizesByKey = [];
 
        // This check must happen outside the write query to respect eventual consistency
        $res = $db->newSelectQueryBuilder()
            ->select( $this->addCasTokenFields( $db, [ 'keyname' ] ) )
            ->from( $ptable )
            ->where( $this->buildExistenceConditions( $db, array_keys( $argsByKey ), (int)$mtime ) )
            ->caller( __METHOD__ )
            ->fetchResultSet();
 
        $curTokensByKey = [];
        foreach ( $res as $row ) {
            $curTokensByKey[$row->keyname] = $this->getCasTokenFromRow( $db, $row );
        }
 
        $nonMatchingByKey = [];
        $rows = [];
        foreach ( $argsByKey as $key => [ $value, $exptime, $casToken ] ) {
            $curToken = $curTokensByKey[$key] ?? null;
            if ( $curToken === null ) {
                $nonMatchingByKey[$key] = true;
                $this->logger->debug( __METHOD__ . ": $key does not exists" );
                continue;
            }
 
            if ( $curToken !== $casToken ) {
                $nonMatchingByKey[$key] = true;
                $this->logger->debug( __METHOD__ . ": $key does not have a matching token" );
                continue;
            }
 
            $serialValue = $this->getSerialized( $value, $key );
            $expiry = $this->makeNewKeyExpiry( $exptime, (int)$mtime );
            $valueSizesByKey[$key] = [ strlen( $serialValue ), 0 ];
 
            $rows[] = $this->buildUpsertRow( $db, $key, $serialValue, $expiry );
        }
        if ( !$rows ) {
            return;
        }
        $db->newInsertQueryBuilder()
            ->insertInto( $ptable )
            ->rows( $rows )
            ->onDuplicateKeyUpdate()
            ->uniqueIndexFields( [ 'keyname' ] )
            ->set( $this->buildMultiUpsertSetForOverwrite( $db ) )
            ->caller( __METHOD__ )->execute();
 
        foreach ( $argsByKey as $key => $unused ) {
            $resByKey[$key] = !isset( $nonMatchingByKey[$key] );
        }
 
        $this->updateOpStats( self::METRIC_OP_CAS, $valueSizesByKey );
    }
 
    private function modifyTableSpecificBlobsForChangeTTL(
        IDatabase $db,
        string $ptable,
        float $mtime,
        array $argsByKey,
        array &$resByKey
    ) {
        $keysBatchesByExpiry = [];
        foreach ( $argsByKey as $key => [ $exptime ] ) {
            $expiry = $this->makeNewKeyExpiry( $exptime, (int)$mtime );
            $keysBatchesByExpiry[$expiry][] = $key;
        }
 
        $existingCount = 0;
        foreach ( $keysBatchesByExpiry as $expiry => $keyBatch ) {
            $db->newUpdateQueryBuilder()
                ->update( $ptable )
                ->set( [ 'exptime' => $this->encodeDbExpiry( $db, $expiry ) ] )
                ->where( $this->buildExistenceConditions( $db, $keyBatch, (int)$mtime ) )
                ->caller( __METHOD__ )->execute();
            $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 => [ $step, $init, $exptime ] ) {
            $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.
            $atomic = $db->startAtomic( __METHOD__, IDatabase::ATOMIC_CANCELABLE );
            try {
                $db->newInsertQueryBuilder()
                    ->insertInto( $ptable )
                    ->rows( $this->buildUpsertRow( $db, $key, $init, $expiry ) )
                    ->onDuplicateKeyUpdate()
                    ->uniqueIndexFields( [ 'keyname' ] )
                    ->set( $this->buildIncrUpsertSet( $db, $step, $init, $expiry, (int)$mtime ) )
                    ->caller( __METHOD__ )->execute();
                $affectedCount = $db->affectedRows();
                $row = $db->newSelectQueryBuilder()
                    ->select( 'value' )
                    ->from( $ptable )
                    ->where( [ 'keyname' => $key ] )
                    ->caller( __METHOD__ )
                    ->fetchRow();
            } catch ( Exception $e ) {
                $db->cancelAtomic( __METHOD__, $atomic );
                throw $e;
            }
            $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 modifyTableSpecificBlobsForIncrInitAsync(
        IDatabase $db,
        string $ptable,
        float $mtime,
        array $argsByKey,
        array &$resByKey
    ) {
        foreach ( $argsByKey as $key => [ $step, $init, $exptime ] ) {
            $expiry = $this->makeNewKeyExpiry( $exptime, (int)$mtime );
            $db->newInsertQueryBuilder()
                ->insertInto( $ptable )
                ->rows( $this->buildUpsertRow( $db, $key, $init, $expiry ) )
                ->onDuplicateKeyUpdate()
                ->uniqueIndexFields( [ 'keyname' ] )
                ->set( $this->buildIncrUpsertSet( $db, $step, $init, $expiry, (int)$mtime ) )
                ->caller( __METHOD__ )->execute();
            if ( !$db->affectedRows() ) {
                $this->logger->warning( __METHOD__ . ": failed to set new $key value" );
            } else {
                $resByKey[$key] = true;
            }
        }
    }
 
    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 );
        } );
 
        // sprintf is used to adjust precision
        return (float)sprintf( '%.6F', $this->locks[$key][self::LOCK_TIME] );
    }
 
    private function buildExistenceConditions( IDatabase $db, $keys, int $time ) {
        // Note that tombstones always have past expiration dates
        return [
            'keyname' => $keys,
            $db->expr( 'exptime', '>=', $db->timestamp( $time ) )
        ];
    }
 
    private function buildUpsertRow(
        IDatabase $db,
        $key,
        $serialValue,
        int $expiry
    ) {
        $row = [
            'keyname' => $key,
            'value'   => $this->dbEncodeSerialValue( $db, $serialValue ),
            'exptime' => $this->encodeDbExpiry( $db, $expiry )
        ];
 
        return $row;
    }
 
    private function buildMultiUpsertSetForOverwrite( IDatabase $db ) {
        $expressionsByColumn = [
            'value'   => $db->buildExcludedValue( 'value' ),
            'exptime' => $db->buildExcludedValue( 'exptime' )
        ];
 
        $set = [];
        foreach ( $expressionsByColumn as $column => $updateExpression ) {
            $set[$column] = new RawSQLValue( $updateExpression );
        }
 
        return $set;
    }
 
    private function buildIncrUpsertSet(
        IDatabase $db,
        int $step,
        int $init,
        int $expiry,
        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 ) )
            ]
        ];
 
        $set = [];
        foreach ( $expressionsByColumn as $column => [ $updateExpression, $initExpression ] ) {
            $rhs = $db->conditional(
                $db->expr( 'exptime', '>=', $db->timestamp( $mtUnixTs ) ),
                $updateExpression,
                $initExpression
            );
            $set[$column] = new RawSQLValue( $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 ) ? (string)$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 garbageCollect( $shardIndex ) {
        // set right away, avoid queuing duplicate async callbacks
        $this->lastGarbageCollect = $this->getCurrentTime();
 
        $garbageCollector = function () use ( $shardIndex ) {
            $db = $this->getConnection( $shardIndex );
            $silenceScope = $this->silenceTransactionProfiler();
            $this->deleteServerObjectsExpiringBefore(
                $db,
                (int)$this->getCurrentTime(),
                $this->purgeLimit
            );
            $this->lastGarbageCollect = $this->getCurrentTime();
        };
 
        if ( $this->asyncHandler ) {
            ( $this->asyncHandler )( $garbageCollector );
        } else {
            $garbageCollector();
        }
    }
 
    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 = [];
            if ( !$this->serverTags ) {
                throw new InvalidArgumentException( "Given a tag but no tags are configured" );
            }
            foreach ( $this->serverTags as $serverShardIndex => $serverTag ) {
                if ( $tag === $serverTag ) {
                    $shardIndexes[] = $serverShardIndex;
                    break;
                }
            }
            if ( !$shardIndexes ) {
                throw new InvalidArgumentException( "Unknown server tag: $tag" );
            }
        } else {
            $shardIndexes = $this->getShardServerIndexes();
            shuffle( $shardIndexes );
        }
 
        $ok = true;
        $numServers = count( $shardIndexes );
 
        $keysDeletedCount = 0;
        foreach ( $shardIndexes as $numServersDone => $shardIndex ) {
            try {
                $db = $this->getConnection( $shardIndex );
 
                // Avoid deadlock (T330377)
                $lockKey = "SqlBagOStuff-purge-shard:$shardIndex";
                if ( !$db->lock( $lockKey, __METHOD__, 0 ) ) {
                    $this->logger->info( "SqlBagOStuff purge for shard $shardIndex already locked, skip" );
                    continue;
                }
 
                $this->deleteServerObjectsExpiringBefore(
                    $db,
                    $timestamp,
                    $limit,
                    $keysDeletedCount,
                    [ 'fn' => $progress, 'serversDone' => $numServersDone, 'serversTotal' => $numServers ]
                );
                $db->unlock( $lockKey, __METHOD__ );
            } catch ( DBError $e ) {
                $this->handleDBError( $e, $shardIndex );
                $ok = false;
            }
        }
 
        return $ok;
    }
 
    private function deleteServerObjectsExpiringBefore(
        IDatabase $db,
        $timestamp,
        $limit,
        &$keysDeletedCount = 0,
        ?array $progress = null
    ) {
        $cutoffUnix = (int)ConvertibleTimestamp::convert( TS::UNIX, $timestamp );
        $tableIndexes = range( 0, $this->numTableShards - 1 );
        shuffle( $tableIndexes );
 
        $batchSize = min( $this->writeBatchSize, $limit );
 
        foreach ( $tableIndexes as $numShardsDone => $tableIndex ) {
            // don't do more than 10% of tables. To avoid overwhelming
            // when there are too many of them. Add one to make sure small number
            // of tables have been taken care of.
            if (
                $numShardsDone > ( ( $this->numTableShards / 10 ) + 1 ) &&
                // running in context of purge maint script. Go through all tables
                $limit !== INF
            ) {
                break;
            }
 
            // 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->newSelectQueryBuilder()
                    ->select( [ 'keyname', 'exptime' ] )
                    ->from( $this->getTableNameByShard( $tableIndex ) )
                    ->where( $db->expr( 'exptime', '<', $db->timestamp( $cutoffUnix ) ) )
                    ->andWhere( $maxExp ? $db->expr( 'exptime', '>=', $maxExp ) : [] )
                    ->orderBy( 'exptime', SelectQueryBuilder::SORT_ASC )
                    ->limit( $batchSize )
                    ->caller( __METHOD__ )
                    ->fetchResultSet();
 
                if ( $res->numRows() ) {
                    $row = $res->current();
                    if ( $minExpUnix === null ) {
                        $minExpUnix = (int)ConvertibleTimestamp::convert( TS::UNIX, $row->exptime );
                        $totalSeconds = max( $cutoffUnix - $minExpUnix, 1 );
                    }
 
                    $keys = [];
                    foreach ( $res as $row ) {
                        $keys[] = $row->keyname;
                        $maxExp = $row->exptime;
                    }
 
                    $db->newDeleteQueryBuilder()
                        ->deleteFrom( $this->getTableNameByShard( $tableIndex ) )
                        ->where( [
                            'keyname' => $keys,
                            $db->expr( 'exptime', '<', $db->timestamp( $cutoffUnix ) ),
                        ] )
                        ->caller( __METHOD__ )->execute();
                    $keysDeletedCount += $db->affectedRows();
                }
 
                if ( $progress && is_callable( $progress['fn'] ) ) {
                    if ( $totalSeconds ) {
                        $maxExpUnix = (int)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'] )( (int)( $overallRatio * 100 ) );
                }
            } while ( $res->numRows() && $keysDeletedCount < $limit );
        }
    }
 
    public function doLock( $key, $timeout = 6, $exptime = 6 ) {
        $silenceScope = $this->silenceTransactionProfiler();
 
        $lockTsUnix = null;
 
        $shardIndexes = $this->getShardIndexesForKey( $key );
        foreach ( $shardIndexes as $shardIndex ) {
            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();
 
        $shardIndexes = $this->getShardIndexesForKey( $key );
        $released = false;
        foreach ( $shardIndexes as $shardIndex ) {
            try {
                $db = $this->getConnection( $shardIndex );
                $released = $db->unlock( $key, __METHOD__ );
            } catch ( DBError $e ) {
                $this->handleDBError( $e, $shardIndex );
                $released = false;
            }
        }
 
        return $released;
    }
 
    protected function makeKeyInternal( $keyspace, $components ) {
        $key = strtr( $keyspace, ' ', '_' );
        foreach ( $components as $component ) {
            $component = strtr( $component ?? '', [
                ' ' => '_', // Avoid unnecessary misses from pre-1.35 code
                ':' => '%3A',
            ] );
            $key .= ':' . $component;
        }
 
        // SQL schema for 'objectcache' specifies keys as varchar(255).
        // * Reserve 45 chars for prefixes used by wrappers like WANObjectCache.
        return $this->makeFallbackKey( $key, 205 );
    }
 
    protected function requireConvertGenericKey(): bool {
        return true;
    }
 
    protected function serialize( $value ) {
        if ( is_int( $value ) ) {
            return $value;
        }
 
        $serial = serialize( $value );
        if ( $this->hasZlib ) {
            // 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 ( $this->hasZlib ) {
            // phpcs:ignore Generic.PHP.NoSilencedErrors.Discouraged
            $decompressed = @gzinflate( $value );
 
            if ( $decompressed !== false ) {
                $value = $decompressed;
            }
        }
 
        return unserialize( $value );
    }
 
    private function getLoadBalancer(): ILoadBalancer {
        if ( !$this->loadBalancer ) {
            $this->loadBalancer = ( $this->loadBalancerCallback )();
        }
        return $this->loadBalancer;
    }
 
    private function getConnectionViaLoadBalancer() {
        $lb = $this->getLoadBalancer();
 
        if ( $lb->getServerAttributes( ServerInfo::WRITER_INDEX )[Database::ATTR_DB_LEVEL_LOCKING] ) {
            // Use the main connection to avoid transaction deadlocks
            $conn = $lb->getMaintenanceConnectionRef( DB_PRIMARY, [], $this->dbDomain );
        } 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(
                DB_PRIMARY,
                [],
                $this->dbDomain,
                $lb::CONN_TRX_AUTOCOMMIT
            );
        }
 
        // Make sure any errors are thrown now while we can more easily handle them
        $conn->ensureConnection();
        return $conn;
    }
 
    private function getConnectionFromServerInfo( $shardIndex, array $server ) {
        if ( !isset( $this->conns[$shardIndex] ) ) {
            $server['logger'] = $this->logger;
            // Always use autocommit mode, even if DBO_TRX is configured
            $server['flags'] ??= 0;
            $server['flags'] &= ~( IDatabase::DBO_TRX | IDatabase::DBO_DEFAULT );
 
            $conn = MediaWikiServices::getInstance()->getDatabaseFactory()
                ->create( $server['type'], $server );
 
            // Automatically create the objectcache table for sqlite as needed
            if ( $conn->getType() === 'sqlite' ) {
                $this->initSqliteDatabase( $conn );
            }
            $this->conns[$shardIndex] = $conn;
        }
 
        // @phan-suppress-next-line PhanTypeMismatchReturnNullable False positive
        return $this->conns[$shardIndex];
    }
 
    private function handleDBError( DBError $exception, $shardIndex ) {
        if ( !$this->useLB && $exception instanceof DBConnectionError ) {
            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;
        }
        $this->logger->error( "DBError: {$exception->getMessage()}", [ 'exception' => $exception ] );
        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 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 IF NOT EXISTS $encShardTable LIKE $encBaseTable", __METHOD__ );
                }
            }
        }
    }
 
    private function getShardServerIndexes() {
        if ( $this->useLB ) {
            // LoadBalancer based configuration
            $shardIndexes = [ 0 ];
        } else {
            // Striped array of database servers
            $shardIndexes = array_keys( $this->serverTags );
        }
 
        return $shardIndexes;
    }
 
    #[\NoDiscard]
    private function silenceTransactionProfiler(): ?ScopedCallback {
        if ( $this->serverInfos ) {
            return null; // no TransactionProfiler injected anyway
        }
        return Profiler::instance()->getTransactionProfiler()->silenceForScope();
    }
}