SqlBlobStore.php

Go to the documentation of this file.

<?php
namespace MediaWiki\Storage;
 
use AppendIterator;
use DBAccessObjectUtils;
use ExternalStoreAccess;
use ExternalStoreException;
use HistoryBlobUtils;
use IDBAccessObject;
use InvalidArgumentException;
use StatusValue;
use WANObjectCache;
use Wikimedia\Assert\Assert;
use Wikimedia\AtEase\AtEase;
use Wikimedia\Rdbms\IDatabase;
use Wikimedia\Rdbms\ILoadBalancer;
 
class SqlBlobStore implements BlobStore {
 
    // Note: the name has been taken unchanged from the old Revision class.
    public const TEXT_CACHE_GROUP = 'revisiontext:10';
 
    public const DEFAULT_TTL = 7 * 24 * 3600; // 7 days
 
    private $dbLoadBalancer;
 
    private $extStoreAccess;
 
    private $cache;
 
    private $dbDomain;
 
    private $cacheExpiry = self::DEFAULT_TTL;
 
    private $compressBlobs = false;
 
    private $legacyEncoding = false;
 
    private $useExternalStore = false;
 
    public function __construct(
        ILoadBalancer $dbLoadBalancer,
        ExternalStoreAccess $extStoreAccess,
        WANObjectCache $cache,
        $dbDomain = false
    ) {
        $this->dbLoadBalancer = $dbLoadBalancer;
        $this->extStoreAccess = $extStoreAccess;
        $this->cache = $cache;
        $this->dbDomain = $dbDomain;
    }
 
    public function getCacheExpiry() {
        return $this->cacheExpiry;
    }
 
    public function setCacheExpiry( int $cacheExpiry ) {
        $this->cacheExpiry = $cacheExpiry;
    }
 
    public function getCompressBlobs() {
        return $this->compressBlobs;
    }
 
    public function setCompressBlobs( $compressBlobs ) {
        $this->compressBlobs = $compressBlobs;
    }
 
    public function getLegacyEncoding() {
        return $this->legacyEncoding;
    }
 
    public function setLegacyEncoding( string $legacyEncoding ) {
        $this->legacyEncoding = $legacyEncoding;
    }
 
    public function getUseExternalStore() {
        return $this->useExternalStore;
    }
 
    public function setUseExternalStore( bool $useExternalStore ) {
        $this->useExternalStore = $useExternalStore;
    }
 
    private function getDBLoadBalancer() {
        return $this->dbLoadBalancer;
    }
 
    private function getDBConnection( $index ) {
        $lb = $this->getDBLoadBalancer();
        return $lb->getConnection( $index, [], $this->dbDomain );
    }
 
    public function storeBlob( $data, $hints = [] ) {
        $flags = $this->compressData( $data );
 
        # Write to external storage if required
        if ( $this->useExternalStore ) {
            // Store and get the URL
            try {
                $data = $this->extStoreAccess->insert( $data, [ 'domain' => $this->dbDomain ] );
            } catch ( ExternalStoreException $e ) {
                throw new BlobAccessException( $e->getMessage(), 0, $e );
            }
            if ( !$data ) {
                throw new BlobAccessException( "Failed to store text to external storage" );
            }
            if ( $flags ) {
                return 'es:' . $data . '?flags=' . $flags;
            } else {
                return 'es:' . $data;
            }
        } else {
            $dbw = $this->getDBConnection( DB_PRIMARY );
 
            $dbw->newInsertQueryBuilder()
                ->insertInto( 'text' )
                ->row( [ 'old_text' => $data, 'old_flags' => $flags ] )
                ->caller( __METHOD__ )->execute();
 
            $textId = $dbw->insertId();
 
            return self::makeAddressFromTextId( $textId );
        }
    }
 
    public function getBlob( $blobAddress, $queryFlags = 0 ) {
        Assert::parameterType( 'string', $blobAddress, '$blobAddress' );
 
        $error = null;
        $blob = $this->cache->getWithSetCallback(
            $this->getCacheKey( $blobAddress ),
            $this->getCacheTTL(),
            function ( $unused, &$ttl, &$setOpts ) use ( $blobAddress, $queryFlags, &$error ) {
                // Ignore $setOpts; blobs are immutable and negatives are not cached
                [ $result, $errors ] = $this->fetchBlobs( [ $blobAddress ], $queryFlags );
                // No negative caching; negative hits on text rows may be due to corrupted replica DBs
                $error = $errors[$blobAddress] ?? null;
                if ( $error ) {
                    $ttl = WANObjectCache::TTL_UNCACHEABLE;
                }
                return $result[$blobAddress];
            },
            $this->getCacheOptions()
        );
 
        if ( $error ) {
            if ( $error[0] === 'badrevision' ) {
                throw new BadBlobException( $error[1] );
            } else {
                throw new BlobAccessException( $error[1] );
            }
        }
 
        Assert::postcondition( is_string( $blob ), 'Blob must not be null' );
        return $blob;
    }
 
    public function getBlobBatch( $blobAddresses, $queryFlags = 0 ) {
        // FIXME: All caching has temporarily been removed in I94c6f9ba7b9caeeb due to T235188.
        //        Caching behavior should be restored by reverting I94c6f9ba7b9caeeb as soon as
        //        the root cause of T235188 has been resolved.
 
        [ $blobsByAddress, $errors ] = $this->fetchBlobs( $blobAddresses, $queryFlags );
 
        $blobsByAddress = array_map( static function ( $blob ) {
            return $blob === false ? null : $blob;
        }, $blobsByAddress );
 
        $result = StatusValue::newGood( $blobsByAddress );
        foreach ( $errors as $error ) {
            // @phan-suppress-next-line PhanParamTooFewUnpack
            $result->warning( ...$error );
        }
        return $result;
    }
 
    private function fetchBlobs( $blobAddresses, $queryFlags ) {
        $textIdToBlobAddress = [];
        $result = [];
        $errors = [];
        foreach ( $blobAddresses as $blobAddress ) {
            try {
                [ $schema, $id, $params ] = self::splitBlobAddress( $blobAddress );
            } catch ( InvalidArgumentException $ex ) {
                throw new BlobAccessException(
                    $ex->getMessage() . '. Use findBadBlobs.php to remedy.',
                    0,
                    $ex
                );
            }
 
            if ( $schema === 'es' ) {
                if ( $params && isset( $params['flags'] ) ) {
                    $blob = $this->expandBlob( $id, $params['flags'] . ',external', $blobAddress );
                } else {
                    $blob = $this->expandBlob( $id, 'external', $blobAddress );
                }
 
                if ( $blob === false ) {
                    $errors[$blobAddress] = [
                        'internalerror',
                        "Bad data in external store address $id. Use findBadBlobs.php to remedy."
                    ];
                }
                $result[$blobAddress] = $blob;
            } elseif ( $schema === 'bad' ) {
                // Database row was marked as "known bad"
                wfDebug(
                    __METHOD__
                    . ": loading known-bad content ($blobAddress), returning empty string"
                );
                $result[$blobAddress] = '';
                $errors[$blobAddress] = [
                    'badrevision',
                    'The content of this revision is missing or corrupted (bad schema)'
                ];
            } elseif ( $schema === 'tt' ) {
                $textId = intval( $id );
 
                if ( $textId < 1 || $id !== (string)$textId ) {
                    $errors[$blobAddress] = [
                        'internalerror',
                        "Bad blob address: $blobAddress. Use findBadBlobs.php to remedy."
                    ];
                    $result[$blobAddress] = false;
                }
 
                $textIdToBlobAddress[$textId] = $blobAddress;
            } else {
                $errors[$blobAddress] = [
                    'internalerror',
                    "Unknown blob address schema: $schema. Use findBadBlobs.php to remedy."
                ];
                $result[$blobAddress] = false;
            }
        }
 
        $textIds = array_keys( $textIdToBlobAddress );
        if ( !$textIds ) {
            return [ $result, $errors ];
        }
        // Callers doing updates will pass in READ_LATEST as usual. Since the text/blob tables
        // do not normally get rows changed around, set READ_LATEST_IMMUTABLE in those cases.
        $queryFlags |= DBAccessObjectUtils::hasFlags( $queryFlags, IDBAccessObject::READ_LATEST )
            ? IDBAccessObject::READ_LATEST_IMMUTABLE
            : 0;
        [ $index, $options, $fallbackIndex, $fallbackOptions ] =
            self::getDBOptions( $queryFlags );
        // Text data is immutable; check replica DBs first.
        $dbConnection = $this->getDBConnection( $index );
        $rows = $dbConnection->newSelectQueryBuilder()
            ->select( [ 'old_id', 'old_text', 'old_flags' ] )
            ->from( 'text' )
            ->where( [ 'old_id' => $textIds ] )
            ->options( $options )
            ->caller( __METHOD__ )->fetchResultSet();
        $numRows = $rows->numRows();
 
        // Fallback to DB_PRIMARY in some cases if not all the rows were found, using the appropriate
        // options, such as FOR UPDATE to avoid missing rows due to REPEATABLE-READ.
        if ( $numRows !== count( $textIds ) && $fallbackIndex !== null ) {
            $fetchedTextIds = [];
            foreach ( $rows as $row ) {
                $fetchedTextIds[] = $row->old_id;
            }
            $missingTextIds = array_diff( $textIds, $fetchedTextIds );
            $dbConnection = $this->getDBConnection( $fallbackIndex );
            $rowsFromFallback = $dbConnection->newSelectQueryBuilder()
                ->select( [ 'old_id', 'old_text', 'old_flags' ] )
                ->from( 'text' )
                ->where( [ 'old_id' => $missingTextIds ] )
                ->options( $fallbackOptions )
                ->caller( __METHOD__ )->fetchResultSet();
            $appendIterator = new AppendIterator();
            $appendIterator->append( $rows );
            $appendIterator->append( $rowsFromFallback );
            $rows = $appendIterator;
        }
 
        foreach ( $rows as $row ) {
            $blobAddress = $textIdToBlobAddress[$row->old_id];
            $blob = false;
            if ( $row->old_text !== null ) {
                $blob = $this->expandBlob( $row->old_text, $row->old_flags, $blobAddress );
            }
            if ( $blob === false ) {
                $errors[$blobAddress] = [
                    'internalerror',
                    "Bad data in text row {$row->old_id}. Use findBadBlobs.php to remedy."
                ];
            }
            $result[$blobAddress] = $blob;
        }
 
        // If we're still missing some of the rows, set errors for missing blobs.
        if ( count( $result ) !== count( $blobAddresses ) ) {
            foreach ( $blobAddresses as $blobAddress ) {
                if ( !isset( $result[$blobAddress ] ) ) {
                    $errors[$blobAddress] = [
                        'internalerror',
                        "Unable to fetch blob at $blobAddress. Use findBadBlobs.php to remedy."
                    ];
                    $result[$blobAddress] = false;
                }
            }
        }
        return [ $result, $errors ];
    }
 
    private static function getDBOptions( $bitfield ) {
        if ( DBAccessObjectUtils::hasFlags( $bitfield, IDBAccessObject::READ_LATEST_IMMUTABLE ) ) {
            $index = DB_REPLICA; // override READ_LATEST if set
            $fallbackIndex = DB_PRIMARY;
        } elseif ( DBAccessObjectUtils::hasFlags( $bitfield, IDBAccessObject::READ_LATEST ) ) {
            $index = DB_PRIMARY;
            $fallbackIndex = null;
        } else {
            $index = DB_REPLICA;
            $fallbackIndex = null;
        }
 
        $lockingOptions = [];
        if ( DBAccessObjectUtils::hasFlags( $bitfield, IDBAccessObject::READ_EXCLUSIVE ) ) {
            $lockingOptions[] = 'FOR UPDATE';
        } elseif ( DBAccessObjectUtils::hasFlags( $bitfield, IDBAccessObject::READ_LOCKING ) ) {
            $lockingOptions[] = 'LOCK IN SHARE MODE';
        }
 
        if ( $fallbackIndex !== null ) {
            $options = []; // locks on DB_REPLICA make no sense
            $fallbackOptions = $lockingOptions;
        } else {
            $options = $lockingOptions;
            $fallbackOptions = []; // no fallback
        }
 
        return [ $index, $options, $fallbackIndex, $fallbackOptions ];
    }
 
    private function getCacheKey( $blobAddress ) {
        return $this->cache->makeGlobalKey(
            'SqlBlobStore-blob',
            $this->dbLoadBalancer->resolveDomainID( $this->dbDomain ),
            $blobAddress
        );
    }
 
    private function getCacheOptions() {
        return [
            'pcGroup' => self::TEXT_CACHE_GROUP,
            'pcTTL' => WANObjectCache::TTL_PROC_LONG,
            'segmentable' => true
        ];
    }
 
    public function expandBlob( $raw, $flags, $blobAddress = null ) {
        if ( is_string( $flags ) ) {
            $flags = self::explodeFlags( $flags );
        }
        if ( in_array( 'error', $flags ) ) {
            throw new BadBlobException(
                "The content of this revision is missing or corrupted (error flag)"
            );
        }
 
        // Use external methods for external objects, text in table is URL-only then
        if ( in_array( 'external', $flags ) ) {
            $url = $raw;
            $parts = explode( '://', $url, 2 );
            if ( count( $parts ) == 1 || $parts[1] == '' ) {
                return false;
            }
 
            if ( $blobAddress ) {
                // The cached value should be decompressed, so handle that and return here.
                return $this->cache->getWithSetCallback(
                    $this->getCacheKey( $blobAddress ),
                    $this->getCacheTTL(),
                    function () use ( $url, $flags ) {
                        // Ignore $setOpts; blobs are immutable and negatives are not cached
                        $blob = $this->extStoreAccess
                            ->fetchFromURL( $url, [ 'domain' => $this->dbDomain ] );
 
                        return $blob === false ? false : $this->decompressData( $blob, $flags );
                    },
                    $this->getCacheOptions()
                );
            } else {
                $blob = $this->extStoreAccess->fetchFromURL( $url, [ 'domain' => $this->dbDomain ] );
                return $blob === false ? false : $this->decompressData( $blob, $flags );
            }
        } else {
            return $this->decompressData( $raw, $flags );
        }
    }
 
    public function compressData( &$blob ) {
        $blobFlags = [];
 
        // Revisions not marked as UTF-8 will have legacy decoding applied by decompressData().
        // XXX: if $this->legacyEncoding is not set, we could skip this. That would however be
        // risky, since $this->legacyEncoding being set in the future would lead to data corruption.
        $blobFlags[] = 'utf-8';
 
        if ( $this->compressBlobs ) {
            if ( function_exists( 'gzdeflate' ) ) {
                $deflated = gzdeflate( $blob );
 
                if ( $deflated === false ) {
                    wfLogWarning( __METHOD__ . ': gzdeflate() failed' );
                } else {
                    $blob = $deflated;
                    $blobFlags[] = 'gzip';
                }
            } else {
                wfDebug( __METHOD__ . " -- no zlib support, not compressing" );
            }
        }
        return implode( ',', $blobFlags );
    }
 
    public function decompressData( string $blob, array $blobFlags ) {
        if ( in_array( 'error', $blobFlags ) ) {
            // Error row, return false
            return false;
        }
 
        if ( in_array( 'gzip', $blobFlags ) ) {
            # Deal with optional compression of archived pages.
            # This can be done periodically via maintenance/compressOld.php, and
            # as pages are saved if $wgCompressRevisions is set.
            $blob = gzinflate( $blob );
 
            if ( $blob === false ) {
                wfWarn( __METHOD__ . ': gzinflate() failed' );
                return false;
            }
        }
 
        if ( in_array( 'object', $blobFlags ) ) {
            # Generic compressed storage
            $obj = HistoryBlobUtils::unserialize( $blob );
            if ( !$obj ) {
                // Invalid object
                return false;
            }
            $blob = $obj->getText();
        }
 
        // Needed to support old revisions from before MW 1.5.
        if ( $blob !== false && $this->legacyEncoding
            && !in_array( 'utf-8', $blobFlags ) && !in_array( 'utf8', $blobFlags )
        ) {
            # Old revisions kept around in a legacy encoding?
            # Upconvert on demand.
            # ("utf8" checked for compatibility with some broken
            #  conversion scripts 2008-12-30)
            # Even with //IGNORE iconv can whine about illegal characters in
            # *input* string. We just ignore those too.
            # REF: https://bugs.php.net/bug.php?id=37166
            # REF: https://phabricator.wikimedia.org/T18885
            AtEase::suppressWarnings();
            $blob = iconv( $this->legacyEncoding, 'UTF-8//IGNORE', $blob );
            AtEase::restoreWarnings();
        }
 
        return $blob;
    }
 
    private function getCacheTTL() {
        $cache = $this->cache;
 
        if ( $cache->getQoS( $cache::ATTR_DURABILITY ) >= $cache::QOS_DURABILITY_RDBMS ) {
            // Do not cache RDBMs blobs in...the RDBMs store
            $ttl = $cache::TTL_UNCACHEABLE;
        } else {
            $ttl = $this->cacheExpiry ?: $cache::TTL_UNCACHEABLE;
        }
 
        return $ttl;
    }
 
    public function getTextIdFromAddress( $address ) {
        [ $schema, $id, ] = self::splitBlobAddress( $address );
 
        if ( $schema !== 'tt' ) {
            return null;
        }
 
        $textId = intval( $id );
 
        if ( !$textId || $id !== (string)$textId ) {
            throw new InvalidArgumentException( "Malformed text_id: $id" );
        }
 
        return $textId;
    }
 
    public static function makeAddressFromTextId( $id ) {
        return 'tt:' . $id;
    }
 
    public static function explodeFlags( string $flagsString ) {
        return $flagsString === '' ? [] : explode( ',', $flagsString );
    }
 
    public static function splitBlobAddress( $address ) {
        if ( !preg_match( '/^([-+.\w]+):([^\s?]+)(\?([^\s]*))?$/', $address, $m ) ) {
            throw new InvalidArgumentException( "Bad blob address: $address" );
        }
 
        $schema = strtolower( $m[1] );
        $id = $m[2];
        $parameters = wfCgiToArray( $m[4] ?? '' );
 
        return [ $schema, $id, $parameters ];
    }
 
    public function isReadOnly() {
        if ( $this->useExternalStore && $this->extStoreAccess->isReadOnly() ) {
            return true;
        }
 
        return ( $this->getDBLoadBalancer()->getReadOnlyReason() !== false );
    }
}