SqlBlobStore.php

Go to the documentation of this file.

<?php
namespace MediaWiki\Storage;
 
use DBAccessObjectUtils;
use ExternalStore;
use IDBAccessObject;
use IExpiringStore;
use InvalidArgumentException;
use Language;
use MWException;
use WANObjectCache;
use Wikimedia\Assert\Assert;
use Wikimedia\Rdbms\Database;
use Wikimedia\Rdbms\IDatabase;
use Wikimedia\Rdbms\LoadBalancer;
 
class SqlBlobStore implements IDBAccessObject, BlobStore {
 
    // Note: the name has been taken unchanged from the Revision class.
    const TEXT_CACHE_GROUP = 'revisiontext:10';
 
    private $dbLoadBalancer;
 
    private $cache;
 
    private $wikiId;
 
    private $cacheExpiry = 604800; // 7 days
 
    private $compressBlobs = false;
 
    private $legacyEncoding = false;
 
    private $legacyEncodingConversionLang = null;
 
    private $useExternalStore = false;
 
    public function __construct(
        LoadBalancer $dbLoadBalancer,
        WANObjectCache $cache,
        $wikiId = false
    ) {
        $this->dbLoadBalancer = $dbLoadBalancer;
        $this->cache = $cache;
        $this->wikiId = $wikiId;
    }
 
    public function getCacheExpiry() {
        return $this->cacheExpiry;
    }
 
    public function setCacheExpiry( $cacheExpiry ) {
        Assert::parameterType( 'integer', $cacheExpiry, '$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 getLegacyEncodingConversionLang() {
        return $this->legacyEncodingConversionLang;
    }
 
    public function setLegacyEncoding( $legacyEncoding, Language $language ) {
        Assert::parameterType( 'string', $legacyEncoding, '$legacyEncoding' );
 
        $this->legacyEncoding = $legacyEncoding;
        $this->legacyEncodingConversionLang = $language;
    }
 
    public function getUseExternalStore() {
        return $this->useExternalStore;
    }
 
    public function setUseExternalStore( $useExternalStore ) {
        Assert::parameterType( 'boolean', $useExternalStore, '$useExternalStore' );
 
        $this->useExternalStore = $useExternalStore;
    }
 
    private function getDBLoadBalancer() {
        return $this->dbLoadBalancer;
    }
 
    private function getDBConnection( $index ) {
        $lb = $this->getDBLoadBalancer();
        return $lb->getConnection( $index, [], $this->wikiId );
    }
 
    public function storeBlob( $data, $hints = [] ) {
        try {
            $flags = $this->compressData( $data );
 
            # Write to external storage if required
            if ( $this->useExternalStore ) {
                // Store and get the URL
                $data = ExternalStore::insertToDefault( $data );
                if ( !$data ) {
                    throw new BlobAccessException( "Failed to store text to external storage" );
                }
                if ( $flags ) {
                    $flags .= ',';
                }
                $flags .= 'external';
 
                // TODO: we could also return an address for the external store directly here.
                // That would mean bypassing the text table entirely when the external store is
                // used. We'll need to assess expected fallout before doing that.
            }
 
            $dbw = $this->getDBConnection( DB_MASTER );
 
            $old_id = $dbw->nextSequenceValue( 'text_old_id_seq' );
            $dbw->insert(
                'text',
                [
                    'old_id' => $old_id,
                    'old_text' => $data,
                    'old_flags' => $flags,
                ],
                __METHOD__
            );
 
            $textId = $dbw->insertId();
 
            return 'tt:' . $textId;
        } catch ( MWException $e ) {
            throw new BlobAccessException( $e->getMessage(), 0, $e );
        }
    }
 
    public function getBlob( $blobAddress, $queryFlags = 0 ) {
        Assert::parameterType( 'string', $blobAddress, '$blobAddress' );
 
        // No negative caching; negative hits on text rows may be due to corrupted replica DBs
        $blob = $this->cache->getWithSetCallback(
            // TODO: change key, since this is not necessarily revision text!
            $this->cache->makeKey( 'revisiontext', 'textid', $blobAddress ),
            $this->getCacheTTL(),
            function ( $unused, &$ttl, &$setOpts ) use ( $blobAddress, $queryFlags ) {
                list( $index ) = DBAccessObjectUtils::getDBOptions( $queryFlags );
                $setOpts += Database::getCacheSetOptions( $this->getDBConnection( $index ) );
 
                return $this->fetchBlob( $blobAddress, $queryFlags );
            },
            [ 'pcGroup' => self::TEXT_CACHE_GROUP, 'pcTTL' => IExpiringStore::TTL_PROC_LONG ]
        );
 
        if ( $blob === false ) {
            throw new BlobAccessException( 'Failed to load blob from address ' . $blobAddress );
        }
 
        return $blob;
    }
 
    private function fetchBlob( $blobAddress, $queryFlags ) {
        list( $schema, $id, ) = self::splitBlobAddress( $blobAddress );
 
        //TODO: MCR: also support 'ex' schema with ExternalStore URLs, plus flags encoded in the URL!
        if ( $schema === 'tt' ) {
            $textId = intval( $id );
        } else {
            // XXX: change to better exceptions! That makes migration more difficult, though.
            throw new BlobAccessException( "Unknown blob address schema: $schema" );
        }
 
        if ( !$textId || $id !== (string)$textId ) {
            // XXX: change to better exceptions! That makes migration more difficult, though.
            throw new BlobAccessException( "Bad blob address: $blobAddress" );
        }
 
        // 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, self::READ_LATEST )
            ? self::READ_LATEST_IMMUTABLE
            : 0;
 
        list( $index, $options, $fallbackIndex, $fallbackOptions ) =
            DBAccessObjectUtils::getDBOptions( $queryFlags );
 
        // Text data is immutable; check replica DBs first.
        $row = $this->getDBConnection( $index )->selectRow(
            'text',
            [ 'old_text', 'old_flags' ],
            [ 'old_id' => $textId ],
            __METHOD__,
            $options
        );
 
        // Fallback to DB_MASTER in some cases if the row was not found, using the appropriate
        // options, such as FOR UPDATE to avoid missing rows due to REPEATABLE-READ.
        if ( !$row && $fallbackIndex !== null ) {
            $row = $this->getDBConnection( $fallbackIndex )->selectRow(
                'text',
                [ 'old_text', 'old_flags' ],
                [ 'old_id' => $textId ],
                __METHOD__,
                $fallbackOptions
            );
        }
 
        if ( !$row ) {
            wfWarn( __METHOD__ . ": No text row with ID $textId." );
            return false;
        }
 
        $blob = $this->expandBlob( $row->old_text, $row->old_flags, $blobAddress );
 
        if ( $blob === false ) {
            wfWarn( __METHOD__ . ": Bad data in text row $textId." );
            return false;
        }
 
        return $blob;
    }
 
    public function expandBlob( $raw, $flags, $cacheKey = null ) {
        if ( is_string( $flags ) ) {
            $flags = explode( ',', $flags );
        }
 
        // 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 ( $cacheKey && $this->wikiId === false ) {
                // Make use of the wiki-local revision text cache.
                // The cached value should be decompressed, so handle that and return here.
                // NOTE: we rely on $this->cache being the right cache for $this->wikiId!
                return $this->cache->getWithSetCallback(
                    // TODO: change key, since this is not necessarily revision text!
                    $this->cache->makeKey( 'revisiontext', 'textid', $cacheKey ),
                    $this->getCacheTTL(),
                    function () use ( $url, $flags ) {
                        // No negative caching per BlobStore::getBlob()
                        $blob = ExternalStore::fetchFromURL( $url, [ 'wiki' => $this->wikiId ] );
 
                        return $this->decompressData( $blob, $flags );
                    },
                    [ 'pcGroup' => self::TEXT_CACHE_GROUP, 'pcTTL' => WANObjectCache::TTL_PROC_LONG ]
                );
            } else {
                $blob = ExternalStore::fetchFromURL( $url, [ 'wiki' => $this->wikiId ] );
                return $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\n" );
            }
        }
        return implode( ',', $blobFlags );
    }
 
    public function decompressData( $blob, array $blobFlags ) {
        if ( $blob === false ) {
            // Text failed to be fetched; nothing to do
            return false;
        }
 
        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 ) {
                wfLogWarning( __METHOD__ . ': gzinflate() failed' );
                return false;
            }
        }
 
        if ( in_array( 'object', $blobFlags ) ) {
            # Generic compressed storage
            $obj = unserialize( $blob );
            if ( !is_object( $obj ) ) {
                // Invalid object
                return false;
            }
            $blob = $obj->getText();
        }
 
        // Needed to support old revisions left over from from the 1.4 / 1.5 migration.
        if ( $blob !== false && $this->legacyEncoding && $this->legacyEncodingConversionLang
            && !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)
            $blob = $this->legacyEncodingConversionLang->iconv( $this->legacyEncoding, 'UTF-8', $blob );
        }
 
        return $blob;
    }
 
    private function getCacheTTL() {
        if ( $this->cache->getQoS( WANObjectCache::ATTR_EMULATION )
                <= WANObjectCache::QOS_EMULATION_SQL
        ) {
            // Do not cache RDBMs blobs in...the RDBMs store
            $ttl = WANObjectCache::TTL_UNCACHEABLE;
        } else {
            $ttl = $this->cacheExpiry ?: WANObjectCache::TTL_UNCACHEABLE;
        }
 
        return $ttl;
    }
 
    public function getTextIdFromAddress( $address ) {
        list( $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;
    }
 
    private static function splitBlobAddress( $address ) {
        if ( !preg_match( '/^(\w+):(\w+)(\?(.*))?$/', $address, $m ) ) {
            throw new InvalidArgumentException( "Bad blob address: $address" );
        }
 
        $schema = strtolower( $m[1] );
        $id = $m[2];
        $parameters = isset( $m[4] ) ? wfCgiToArray( $m[4] ) : [];
 
        return [ $schema, $id, $parameters ];
    }
 
    public function isReadOnly() {
        if ( $this->useExternalStore && ExternalStore::defaultStoresAreReadOnly() ) {
            return true;
        }
 
        return ( $this->getDBLoadBalancer()->getReadOnlyReason() !== false );
    }
}