MediaWiki 1.42.0
SqlBlobStore.php
Go to the documentation of this file.
1<?php
26namespace MediaWiki\Storage;
27
28use AppendIterator;
34use InvalidArgumentException;
35use StatusValue;
37use Wikimedia\Assert\Assert;
38use Wikimedia\AtEase\AtEase;
42
51class SqlBlobStore implements BlobStore {
52
53 // Note: the name has been taken unchanged from the old Revision class.
54 public const TEXT_CACHE_GROUP = 'revisiontext:10';
55
57 public const DEFAULT_TTL = 7 * 24 * 3600; // 7 days
58
62 private $dbLoadBalancer;
63
67 private $extStoreAccess;
68
72 private $cache;
73
77 private $dbDomain;
78
82 private $cacheExpiry = self::DEFAULT_TTL;
83
87 private $compressBlobs = false;
88
92 private $legacyEncoding = false;
93
97 private $useExternalStore = false;
98
110 public function __construct(
111 ILoadBalancer $dbLoadBalancer,
112 ExternalStoreAccess $extStoreAccess,
113 WANObjectCache $cache,
114 $dbDomain = false
115 ) {
116 $this->dbLoadBalancer = $dbLoadBalancer;
117 $this->extStoreAccess = $extStoreAccess;
118 $this->cache = $cache;
119 $this->dbDomain = $dbDomain;
120 }
121
125 public function getCacheExpiry() {
126 return $this->cacheExpiry;
127 }
128
132 public function setCacheExpiry( int $cacheExpiry ) {
133 $this->cacheExpiry = $cacheExpiry;
134 }
135
139 public function getCompressBlobs() {
140 return $this->compressBlobs;
141 }
142
146 public function setCompressBlobs( $compressBlobs ) {
147 $this->compressBlobs = $compressBlobs;
148 }
149
154 public function getLegacyEncoding() {
155 return $this->legacyEncoding;
156 }
157
166 public function setLegacyEncoding( string $legacyEncoding ) {
167 $this->legacyEncoding = $legacyEncoding;
168 }
169
173 public function getUseExternalStore() {
174 return $this->useExternalStore;
175 }
176
180 public function setUseExternalStore( bool $useExternalStore ) {
181 $this->useExternalStore = $useExternalStore;
182 }
183
187 private function getDBLoadBalancer() {
188 return $this->dbLoadBalancer;
189 }
190
196 private function getDBConnection( $index ) {
197 $lb = $this->getDBLoadBalancer();
198 return $lb->getConnectionRef( $index, [], $this->dbDomain );
199 }
200
211 public function storeBlob( $data, $hints = [] ) {
212 $flags = $this->compressData( $data );
213
214 # Write to external storage if required
215 if ( $this->useExternalStore ) {
216 // Store and get the URL
217 try {
218 $data = $this->extStoreAccess->insert( $data, [ 'domain' => $this->dbDomain ] );
219 } catch ( ExternalStoreException $e ) {
220 throw new BlobAccessException( $e->getMessage(), 0, $e );
221 }
222 if ( !$data ) {
223 throw new BlobAccessException( "Failed to store text to external storage" );
224 }
225 if ( $flags ) {
226 $flags .= ',';
227 }
228 $flags .= 'external';
229
230 // TODO: we could also return an address for the external store directly here.
231 // That would mean bypassing the text table entirely when the external store is
232 // used. We'll need to assess expected fallout before doing that.
233 }
234
235 $dbw = $this->getDBConnection( DB_PRIMARY );
236
237 $dbw->newInsertQueryBuilder()
238 ->insertInto( 'text' )
239 ->row( [ 'old_text' => $data, 'old_flags' => $flags ] )
240 ->caller( __METHOD__ )->execute();
241
242 $textId = $dbw->insertId();
243
244 return self::makeAddressFromTextId( $textId );
245 }
246
259 public function getBlob( $blobAddress, $queryFlags = 0 ) {
260 Assert::parameterType( 'string', $blobAddress, '$blobAddress' );
261
262 $error = null;
263 $blob = $this->cache->getWithSetCallback(
264 $this->getCacheKey( $blobAddress ),
265 $this->getCacheTTL(),
266 function ( $unused, &$ttl, &$setOpts ) use ( $blobAddress, $queryFlags, &$error ) {
267 // Ignore $setOpts; blobs are immutable and negatives are not cached
268 [ $result, $errors ] = $this->fetchBlobs( [ $blobAddress ], $queryFlags );
269 // No negative caching; negative hits on text rows may be due to corrupted replica DBs
270 $error = $errors[$blobAddress] ?? null;
271 if ( $error ) {
272 $ttl = WANObjectCache::TTL_UNCACHEABLE;
273 }
274 return $result[$blobAddress];
275 },
276 $this->getCacheOptions()
277 );
278
279 if ( $error ) {
280 if ( $error[0] === 'badrevision' ) {
281 throw new BadBlobException( $error[1] );
282 } else {
283 throw new BlobAccessException( $error[1] );
284 }
285 }
286
287 Assert::postcondition( is_string( $blob ), 'Blob must not be null' );
288 return $blob;
289 }
290
302 public function getBlobBatch( $blobAddresses, $queryFlags = 0 ) {
303 // FIXME: All caching has temporarily been removed in I94c6f9ba7b9caeeb due to T235188.
304 // Caching behavior should be restored by reverting I94c6f9ba7b9caeeb as soon as
305 // the root cause of T235188 has been resolved.
306
307 [ $blobsByAddress, $errors ] = $this->fetchBlobs( $blobAddresses, $queryFlags );
308
309 $blobsByAddress = array_map( static function ( $blob ) {
310 return $blob === false ? null : $blob;
311 }, $blobsByAddress );
312
313 $result = StatusValue::newGood( $blobsByAddress );
314 foreach ( $errors as $error ) {
315 // @phan-suppress-next-line PhanParamTooFewUnpack
316 $result->warning( ...$error );
317 }
318 return $result;
319 }
320
335 private function fetchBlobs( $blobAddresses, $queryFlags ) {
336 $textIdToBlobAddress = [];
337 $result = [];
338 $errors = [];
339 foreach ( $blobAddresses as $blobAddress ) {
340 try {
341 [ $schema, $id ] = self::splitBlobAddress( $blobAddress );
342 } catch ( InvalidArgumentException $ex ) {
343 throw new BlobAccessException(
344 $ex->getMessage() . '. Use findBadBlobs.php to remedy.',
345 0,
346 $ex
347 );
348 }
349
350 // TODO: MCR: also support 'ex' schema with ExternalStore URLs, plus flags encoded in the URL!
351 if ( $schema === 'bad' ) {
352 // Database row was marked as "known bad"
353 wfDebug(
354 __METHOD__
355 . ": loading known-bad content ($blobAddress), returning empty string"
356 );
357 $result[$blobAddress] = '';
358 $errors[$blobAddress] = [
359 'badrevision',
360 'The content of this revision is missing or corrupted (bad schema)'
361 ];
362 } elseif ( $schema === 'tt' ) {
363 $textId = intval( $id );
364
365 if ( $textId < 1 || $id !== (string)$textId ) {
366 $errors[$blobAddress] = [
367 'internalerror',
368 "Bad blob address: $blobAddress. Use findBadBlobs.php to remedy."
369 ];
370 $result[$blobAddress] = false;
371 }
372
373 $textIdToBlobAddress[$textId] = $blobAddress;
374 } else {
375 $errors[$blobAddress] = [
376 'internalerror',
377 "Unknown blob address schema: $schema. Use findBadBlobs.php to remedy."
378 ];
379 $result[$blobAddress] = false;
380 }
381 }
382
383 $textIds = array_keys( $textIdToBlobAddress );
384 if ( !$textIds ) {
385 return [ $result, $errors ];
386 }
387 // Callers doing updates will pass in READ_LATEST as usual. Since the text/blob tables
388 // do not normally get rows changed around, set READ_LATEST_IMMUTABLE in those cases.
389 $queryFlags |= DBAccessObjectUtils::hasFlags( $queryFlags, IDBAccessObject::READ_LATEST )
390 ? IDBAccessObject::READ_LATEST_IMMUTABLE
391 : 0;
392 [ $index, $options, $fallbackIndex, $fallbackOptions ] =
393 self::getDBOptions( $queryFlags );
394 // Text data is immutable; check replica DBs first.
395 $dbConnection = $this->getDBConnection( $index );
396 $rows = $dbConnection->newSelectQueryBuilder()
397 ->select( [ 'old_id', 'old_text', 'old_flags' ] )
398 ->from( 'text' )
399 ->where( [ 'old_id' => $textIds ] )
400 ->options( $options )
401 ->caller( __METHOD__ )->fetchResultSet();
402 $numRows = 0;
403 if ( $rows instanceof IResultWrapper ) {
404 $numRows = $rows->numRows();
405 }
406
407 // Fallback to DB_PRIMARY in some cases if not all the rows were found, using the appropriate
408 // options, such as FOR UPDATE to avoid missing rows due to REPEATABLE-READ.
409 if ( $numRows !== count( $textIds ) && $fallbackIndex !== null ) {
410 $fetchedTextIds = [];
411 foreach ( $rows as $row ) {
412 $fetchedTextIds[] = $row->old_id;
413 }
414 $missingTextIds = array_diff( $textIds, $fetchedTextIds );
415 $dbConnection = $this->getDBConnection( $fallbackIndex );
416 $rowsFromFallback = $dbConnection->newSelectQueryBuilder()
417 ->select( [ 'old_id', 'old_text', 'old_flags' ] )
418 ->from( 'text' )
419 ->where( [ 'old_id' => $missingTextIds ] )
420 ->options( $fallbackOptions )
421 ->caller( __METHOD__ )->fetchResultSet();
422 $appendIterator = new AppendIterator();
423 $appendIterator->append( $rows );
424 $appendIterator->append( $rowsFromFallback );
425 $rows = $appendIterator;
426 }
427
428 foreach ( $rows as $row ) {
429 $blobAddress = $textIdToBlobAddress[$row->old_id];
430 $blob = false;
431 if ( $row->old_text !== null ) {
432 $blob = $this->expandBlob( $row->old_text, $row->old_flags, $blobAddress );
433 }
434 if ( $blob === false ) {
435 $errors[$blobAddress] = [
436 'internalerror',
437 "Bad data in text row {$row->old_id}. Use findBadBlobs.php to remedy."
438 ];
439 }
440 $result[$blobAddress] = $blob;
441 }
442
443 // If we're still missing some of the rows, set errors for missing blobs.
444 if ( count( $result ) !== count( $blobAddresses ) ) {
445 foreach ( $blobAddresses as $blobAddress ) {
446 if ( !isset( $result[$blobAddress ] ) ) {
447 $errors[$blobAddress] = [
448 'internalerror',
449 "Unable to fetch blob at $blobAddress. Use findBadBlobs.php to remedy."
450 ];
451 $result[$blobAddress] = false;
452 }
453 }
454 }
455 return [ $result, $errors ];
456 }
457
458 private static function getDBOptions( $bitfield ) {
459 if ( DBAccessObjectUtils::hasFlags( $bitfield, IDBAccessObject::READ_LATEST_IMMUTABLE ) ) {
460 $index = DB_REPLICA; // override READ_LATEST if set
461 $fallbackIndex = DB_PRIMARY;
462 } elseif ( DBAccessObjectUtils::hasFlags( $bitfield, IDBAccessObject::READ_LATEST ) ) {
463 $index = DB_PRIMARY;
464 $fallbackIndex = null;
465 } else {
466 $index = DB_REPLICA;
467 $fallbackIndex = null;
468 }
469
470 $lockingOptions = [];
471 if ( DBAccessObjectUtils::hasFlags( $bitfield, IDBAccessObject::READ_EXCLUSIVE ) ) {
472 $lockingOptions[] = 'FOR UPDATE';
473 } elseif ( DBAccessObjectUtils::hasFlags( $bitfield, IDBAccessObject::READ_LOCKING ) ) {
474 $lockingOptions[] = 'LOCK IN SHARE MODE';
475 }
476
477 if ( $fallbackIndex !== null ) {
478 $options = []; // locks on DB_REPLICA make no sense
479 $fallbackOptions = $lockingOptions;
480 } else {
481 $options = $lockingOptions;
482 $fallbackOptions = []; // no fallback
483 }
484
485 return [ $index, $options, $fallbackIndex, $fallbackOptions ];
486 }
487
498 private function getCacheKey( $blobAddress ) {
499 return $this->cache->makeGlobalKey(
500 'SqlBlobStore-blob',
501 $this->dbLoadBalancer->resolveDomainID( $this->dbDomain ),
502 $blobAddress
503 );
504 }
505
511 private function getCacheOptions() {
512 return [
513 'pcGroup' => self::TEXT_CACHE_GROUP,
514 'pcTTL' => WANObjectCache::TTL_PROC_LONG,
515 'segmentable' => true
516 ];
517 }
518
539 public function expandBlob( $raw, $flags, $blobAddress = null ) {
540 if ( is_string( $flags ) ) {
541 $flags = self::explodeFlags( $flags );
542 }
543 if ( in_array( 'error', $flags ) ) {
544 throw new BadBlobException(
545 "The content of this revision is missing or corrupted (error flag)"
546 );
547 }
548
549 // Use external methods for external objects, text in table is URL-only then
550 if ( in_array( 'external', $flags ) ) {
551 $url = $raw;
552 $parts = explode( '://', $url, 2 );
553 if ( count( $parts ) == 1 || $parts[1] == '' ) {
554 return false;
555 }
556
557 if ( $blobAddress ) {
558 // The cached value should be decompressed, so handle that and return here.
559 return $this->cache->getWithSetCallback(
560 $this->getCacheKey( $blobAddress ),
561 $this->getCacheTTL(),
562 function () use ( $url, $flags ) {
563 // Ignore $setOpts; blobs are immutable and negatives are not cached
564 $blob = $this->extStoreAccess
565 ->fetchFromURL( $url, [ 'domain' => $this->dbDomain ] );
566
567 return $blob === false ? false : $this->decompressData( $blob, $flags );
568 },
569 $this->getCacheOptions()
570 );
571 } else {
572 $blob = $this->extStoreAccess->fetchFromURL( $url, [ 'domain' => $this->dbDomain ] );
573 return $blob === false ? false : $this->decompressData( $blob, $flags );
574 }
575 } else {
576 return $this->decompressData( $raw, $flags );
577 }
578 }
579
596 public function compressData( &$blob ) {
597 $blobFlags = [];
598
599 // Revisions not marked as UTF-8 will have legacy decoding applied by decompressData().
600 // XXX: if $this->legacyEncoding is not set, we could skip this. That would however be
601 // risky, since $this->legacyEncoding being set in the future would lead to data corruption.
602 $blobFlags[] = 'utf-8';
603
604 if ( $this->compressBlobs ) {
605 if ( function_exists( 'gzdeflate' ) ) {
606 $deflated = gzdeflate( $blob );
607
608 if ( $deflated === false ) {
609 wfLogWarning( __METHOD__ . ': gzdeflate() failed' );
610 } else {
611 $blob = $deflated;
612 $blobFlags[] = 'gzip';
613 }
614 } else {
615 wfDebug( __METHOD__ . " -- no zlib support, not compressing" );
616 }
617 }
618 return implode( ',', $blobFlags );
619 }
620
636 public function decompressData( string $blob, array $blobFlags ) {
637 if ( in_array( 'error', $blobFlags ) ) {
638 // Error row, return false
639 return false;
640 }
641
642 if ( in_array( 'gzip', $blobFlags ) ) {
643 # Deal with optional compression of archived pages.
644 # This can be done periodically via maintenance/compressOld.php, and
645 # as pages are saved if $wgCompressRevisions is set.
646 $blob = gzinflate( $blob );
647
648 if ( $blob === false ) {
649 wfWarn( __METHOD__ . ': gzinflate() failed' );
650 return false;
651 }
652 }
653
654 if ( in_array( 'object', $blobFlags ) ) {
655 # Generic compressed storage
656 $obj = HistoryBlobUtils::unserialize( $blob );
657 if ( !$obj ) {
658 // Invalid object
659 return false;
660 }
661 $blob = $obj->getText();
662 }
663
664 // Needed to support old revisions from before MW 1.5.
665 if ( $blob !== false && $this->legacyEncoding
666 && !in_array( 'utf-8', $blobFlags ) && !in_array( 'utf8', $blobFlags )
667 ) {
668 # Old revisions kept around in a legacy encoding?
669 # Upconvert on demand.
670 # ("utf8" checked for compatibility with some broken
671 # conversion scripts 2008-12-30)
672 # Even with //IGNORE iconv can whine about illegal characters in
673 # *input* string. We just ignore those too.
674 # REF: https://bugs.php.net/bug.php?id=37166
675 # REF: https://phabricator.wikimedia.org/T18885
676 AtEase::suppressWarnings();
677 $blob = iconv( $this->legacyEncoding, 'UTF-8//IGNORE', $blob );
678 AtEase::restoreWarnings();
679 }
680
681 return $blob;
682 }
683
691 private function getCacheTTL() {
692 $cache = $this->cache;
693
694 if ( $cache->getQoS( $cache::ATTR_DURABILITY ) >= $cache::QOS_DURABILITY_RDBMS ) {
695 // Do not cache RDBMs blobs in...the RDBMs store
696 $ttl = $cache::TTL_UNCACHEABLE;
697 } else {
698 $ttl = $this->cacheExpiry ?: $cache::TTL_UNCACHEABLE;
699 }
700
701 return $ttl;
702 }
703
724 public function getTextIdFromAddress( $address ) {
725 [ $schema, $id, ] = self::splitBlobAddress( $address );
726
727 if ( $schema !== 'tt' ) {
728 return null;
729 }
730
731 $textId = intval( $id );
732
733 if ( !$textId || $id !== (string)$textId ) {
734 throw new InvalidArgumentException( "Malformed text_id: $id" );
735 }
736
737 return $textId;
738 }
739
753 public static function makeAddressFromTextId( $id ) {
754 return 'tt:' . $id;
755 }
756
763 public static function explodeFlags( string $flagsString ) {
764 return $flagsString === '' ? [] : explode( ',', $flagsString );
765 }
766
777 public static function splitBlobAddress( $address ) {
778 if ( !preg_match( '/^([-+.\w]+):([^\s?]+)(\?([^\s]*))?$/', $address, $m ) ) {
779 throw new InvalidArgumentException( "Bad blob address: $address" );
780 }
781
782 $schema = strtolower( $m[1] );
783 $id = $m[2];
784 $parameters = wfCgiToArray( $m[4] ?? '' );
785
786 return [ $schema, $id, $parameters ];
787 }
788
789 public function isReadOnly() {
790 if ( $this->useExternalStore && $this->extStoreAccess->isReadOnly() ) {
791 return true;
792 }
793
794 return ( $this->getDBLoadBalancer()->getReadOnlyReason() !== false );
795 }
796}
wfDebug( $text, $dest='all', array $context=[])
Sends a line to the debug log if enabled or, optionally, to a comment in output.
wfWarn( $msg, $callerOffset=1, $level=E_USER_NOTICE)
Send a warning either to the debug log or in a PHP error depending on $wgDevelopmentWarnings.
wfLogWarning( $msg, $callerOffset=1, $level=E_USER_WARNING)
Send a warning as a PHP error and the debug log.
wfCgiToArray( $query)
This is the logical opposite of wfArrayToCgi(): it accepts a query string as its argument and returns...
getCacheKey()
Get the cache key used to store status.
Helper class for DAO classes.
static hasFlags( $bitfield, $flags)
This is the main interface for fetching or inserting objects with ExternalStore.
static unserialize(string $str, bool $allowDouble=false)
Unserialize a HistoryBlob.
Exception thrown when a blob has the "bad" content address schema, or has "error" in its old_flags,...
Exception representing a failure to access a data blob.
Service for storing and loading Content objects representing revision data blobs.
static makeAddressFromTextId( $id)
Returns an address referring to content stored in the text table row with the given ID.
getTextIdFromAddress( $address)
Returns an ID corresponding to the old_id field in the text table, corresponding to the given $addres...
__construct(ILoadBalancer $dbLoadBalancer, ExternalStoreAccess $extStoreAccess, WANObjectCache $cache, $dbDomain=false)
decompressData(string $blob, array $blobFlags)
Re-converts revision text according to its flags.
getBlob( $blobAddress, $queryFlags=0)
Retrieve a blob, given an address.
setLegacyEncoding(string $legacyEncoding)
Set the legacy encoding to assume for blobs that do not have the utf-8 flag set.
compressData(&$blob)
If $wgCompressRevisions is enabled, we will compress data.
static splitBlobAddress( $address)
Splits a blob address into three parts: the schema, the ID, and parameters/flags.
getBlobBatch( $blobAddresses, $queryFlags=0)
A batched version of BlobStore::getBlob.
storeBlob( $data, $hints=[])
Stores an arbitrary blob of data and returns an address that can be used with getBlob() to retrieve t...
setUseExternalStore(bool $useExternalStore)
isReadOnly()
Check if the blob metadata or backing blob data store is read-only.
expandBlob( $raw, $flags, $blobAddress=null)
Expand a raw data blob according to the flags given.
static explodeFlags(string $flagsString)
Split a comma-separated old_flags value into its constituent parts.
Generic operation result class Has warning/error list, boolean status and arbitrary value.
Multi-datacenter aware caching interface.
Interface for database access objects.
Service for loading and storing data blobs.
Definition BlobStore.php:33
Basic database interface for live and lazy-loaded relation database handles.
Definition IDatabase.php:36
This class is a delegate to ILBFactory for a given database cluster.
Result wrapper for grabbing data queried from an IDatabase object.
const DB_REPLICA
Definition defines.php:26
const DB_PRIMARY
Definition defines.php:28