Add findBadBlobs script.

This script scans for content blobs that can't be loaded due to
database corruption, and can change their entry in the content table
to an address starting with "bad:". Such addresses cause the content
to be read as empty, with no log entry. This is useful to avoid
errors and log spam due to known bad revisions.

The script is designed to scan a limited number of revisions from a
given start date. The assumption is that database corruption is
generally caused by an intermedia bug or system failure which will
affect many revisions over a short period of time.

Bug: T205936
Change-Id: I6f513133e90701bee89d63efa618afc3f91c2d2b

1 files changed, 371 insertions, 0 deletions

diff --git a/maintenance/findBadBlobs.php b/maintenance/findBadBlobs.php
new file mode 100644
index 000000000000..eab4db8f145e
--- /dev/null
+++ b/maintenance/findBadBlobs.php
@@ -0,0 +1,371 @@
+<?php
+/**
+ * This program is free software; you can redistribute it and/or modify
+ * it under the terms of the GNU General Public License as published by
+ * the Free Software Foundation; either version 2 of the License, or
+ * (at your option) any later version.
+ *
+ * This program is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
+ * GNU General Public License for more details.
+ *
+ * You should have received a copy of the GNU General Public License along
+ * with this program; if not, write to the Free Software Foundation, Inc.,
+ * 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301, USA.
+ * http://www.gnu.org/copyleft/gpl.html
+ *
+ * @file
+ * @ingroup Maintenance
+ */
+
+use MediaWiki\MediaWikiServices;
+use MediaWiki\Revision\RevisionArchiveRecord;
+use MediaWiki\Revision\RevisionRecord;
+use MediaWiki\Revision\RevisionStore;
+use MediaWiki\Revision\RevisionStoreRecord;
+use MediaWiki\Revision\SlotRecord;
+use MediaWiki\Storage\BlobAccessException;
+use MediaWiki\Storage\BlobStore;
+use Wikimedia\Rdbms\LBFactory;
+use Wikimedia\Rdbms\LoadBalancer;
+
+require_once __DIR__ . '/cleanupTable.inc';
+
+/**
+ * Maintenance script for finding and marking bad content blobs.
+ *
+ * @ingroup Maintenance
+ */
+class FindBadBlobs extends Maintenance {
+
+	/**
+	 * @var RevisionStore|null
+	 */
+	private $revisionStore;
+
+	/**
+	 * @var BlobStore|null
+	 */
+	private $blobStore;
+
+	/**
+	 * @var LoadBalancer|null
+	 */
+	private $loadBalancer;
+
+	/**
+	 * @var LBFactory
+	 */
+	private $lbFactory;
+
+	public function __construct() {
+		parent::__construct();
+
+		$this->setBatchSize( 1000 );
+		$this->addDescription( 'Scan for bad content blobs' );
+		$this->addOption( 'from-date', 'Start scanning revisions at the given date. '
+			. 'Format: Anything supported by MediaWiki, e.g. YYYYMMDDHHMMSS or YYYY-MM-DD_HH:MM:SS',
+			true, true );
+		$this->addOption( 'limit', 'Maximum number of revisions to scan. Default: 1000', false, true );
+		$this->addOption( 'mark', 'Mark the blob as "known bad", to avoid errors when '
+			. 'attempting to read it. The value given is the reason for marking the blob as bad, '
+			. 'typically a ticket ID', false, true );
+	}
+
+	public function initializeServices(
+		?RevisionStore $revisionStore = null,
+		?BlobStore $blobStore = null,
+		?LoadBalancer $loadBalancer = null,
+		?LBFactory $lbFactory = null
+	) {
+		$services = MediaWikiServices::getInstance();
+
+		$this->revisionStore = $revisionStore ?? $this->revisionStore ?? $services->getRevisionStore();
+		$this->blobStore = $blobStore ?? $this->blobStore ?? $services->getBlobStore();
+		$this->loadBalancer = $loadBalancer ?? $this->loadBalancer ?? $services->getDBLoadBalancer();
+		$this->lbFactory = $lbFactory ?? $this->lbFactory ?? $services->getDBLoadBalancerFactory();
+	}
+
+	/**
+	 * @return string
+	 */
+	private function getStartTimestamp() {
+		$tsOpt = $this->getOption( 'from-date' );
+		if ( strlen( $tsOpt ) < 14 ) {
+			$this->fatalError( 'Bad timestamp: ' . $tsOpt
+				. ', please provide time and date down to the second.' );
+		}
+
+		$ts = wfTimestamp( TS_MW, $tsOpt );
+		if ( !$ts ) {
+			$this->fatalError( 'Bad timestamp: ' . $tsOpt );
+		}
+
+		return $ts;
+	}
+
+	/**
+	 * @inheritDoc
+	 */
+	public function execute() {
+		$this->initializeServices();
+
+		$fromTimestamp = $this->getStartTimestamp();
+		$total = $this->getOption( 'limit', 1000 );
+
+		$this->scanRevisionsByTimestamp( $fromTimestamp, $total );
+	}
+
+	/**
+	 * @param string $fromTimestamp
+	 * @param int $total
+	 *
+	 * @return int
+	 */
+	private function scanRevisionsByTimestamp( $fromTimestamp, $total ) {
+		$count = 0;
+		$lastRevId = 0;
+		$firstRevId = 0;
+		$lastTimestamp = $fromTimestamp;
+		$revisionRowsScanned = 0;
+		$archiveRowsScanned = 0;
+
+		$this->output( "Scanning revisions table, "
+			. "$total rows starting at rev_timestamp $fromTimestamp\n" );
+
+		while ( $revisionRowsScanned < $total ) {
+			$batchSize = min( $total - $revisionRowsScanned, $this->getBatchSize() );
+			$revisions = $this->loadRevisionsByTimestamp( $lastRevId, $lastTimestamp, $batchSize );
+			if ( !$revisions ) {
+				break;
+			}
+
+			foreach ( $revisions as $rev ) {
+				// we are sorting by timestamp, so we may encounter revision IDs out of sequence
+				$firstRevId = $firstRevId ? min( $firstRevId, $rev->getId() ) : $rev->getId();
+				$lastRevId = max( $lastRevId, $rev->getId() );
+
+				$count += $this->checkRevision( $rev );
+			}
+
+			$lastTimestamp = $rev->getTimestamp();
+			$batchSize = count( $revisions );
+			$revisionRowsScanned += $batchSize;
+			$this->output(
+				"\t- Scanned a batch of $batchSize revisions, "
+				. "up to revision $lastRevId ($lastTimestamp)\n"
+			);
+
+			$this->waitForReplication();
+		}
+
+		// NOTE: the archive table isn't indexed by timestamp, so the best we can do is use the
+		// revision ID just before the first revision ID we found above as the starting point
+		// of the scan, and scan up to on revision after the last revision ID we found above.
+		// If $firstRevId is 0, the loop body above didn't execute,
+		// so we should skip the one below as well.
+		$fromArchived = $this->getNextRevision( $firstRevId, '<', 'DESC' );
+		$maxArchived = $this->getNextRevision( $lastRevId, '>', 'ASC' );
+		$maxArchived = $maxArchived ?: PHP_INT_MAX;
+
+		$this->output( "Scanning archive table by ar_rev_id, $fromArchived to $maxArchived\n" );
+		while ( $firstRevId > 0 && $fromArchived < $maxArchived ) {
+			$batchSize = min( $total - $archiveRowsScanned, $this->getBatchSize() );
+			$revisions = $this->loadArchiveByRevisionId( $fromArchived, $maxArchived, $batchSize );
+			if ( !$revisions ) {
+				break;
+			}
+			/** @var RevisionRecord $rev */
+			foreach ( $revisions as $rev ) {
+				$count += $this->checkRevision( $rev );
+			}
+			$fromArchived = $rev->getId();
+			$batchSize = count( $revisions );
+			$archiveRowsScanned += $batchSize;
+			$this->output(
+				"\t- Scanned a batch of $batchSize archived revisions, "
+				. "up to revision $fromArchived ($lastTimestamp)\n"
+			);
+
+			$this->waitForReplication();
+		}
+
+		if ( $this->hasOption( 'mark' ) ) {
+			$this->output( "Marked $count bad revisions\n" );
+		} else {
+			$this->output( "Found $count bad revisions\n" );
+		}
+
+		$this->output( "The range of archive rows scanned is based on the range of revision IDs "
+			. "scanned in the revision table.\n" );
+
+		return $count;
+	}
+
+	/**
+	 * @param int $afterId
+	 * @param string $fromTimestamp
+	 * @param int $batchSize
+	 *
+	 * @return RevisionStoreRecord[]
+	 */
+	private function loadRevisionsByTimestamp( int $afterId, string $fromTimestamp, $batchSize ) {
+		$db = $this->loadBalancer->getConnectionRef( DB_REPLICA );
+		$queryInfo = $this->revisionStore->getQueryInfo();
+		$quotedTimestamp = $db->addQuotes( $fromTimestamp );
+		$rows = $db->select(
+			$queryInfo['tables'],
+			$queryInfo['fields'],
+			"rev_timestamp > $quotedTimestamp OR "
+			. "(rev_timestamp = $quotedTimestamp AND rev_id > $afterId )",
+			__METHOD__,
+			[ 'LIMIT' => $batchSize, 'ORDER BY' => 'rev_timestamp, rev_id' ],
+			$queryInfo['joins']
+		);
+		$result = $this->revisionStore->newRevisionsFromBatch( $rows, [ 'slots' => true ] );
+		if ( !$result->isOK() ) {
+			$this->fatalError( Status::wrap( $result )->getMessage( false, false, 'en' )->text() );
+		}
+		return $result->value;
+	}
+
+	/**
+	 * @param int $afterId
+	 * @param int $uptoId
+	 * @param int $batchSize
+	 *
+	 * @return RevisionArchiveRecord[]
+	 */
+	private function loadArchiveByRevisionId( int $afterId, int $uptoId, $batchSize ) {
+		$db = $this->loadBalancer->getConnectionRef( DB_REPLICA );
+		$queryInfo = $this->revisionStore->getArchiveQueryInfo();
+		$rows = $db->select(
+			$queryInfo['tables'],
+			$queryInfo['fields'],
+			[ "ar_rev_id > $afterId", "ar_rev_id <= $uptoId" ],
+			__METHOD__,
+			[ 'LIMIT' => $batchSize, 'ORDER BY' => 'ar_rev_id' ],
+			$queryInfo['joins']
+		);
+		$result = $this->revisionStore->newRevisionsFromBatch(
+			$rows,
+			[ 'archive' => true, 'slots' => true ]
+		);
+		if ( !$result->isOK() ) {
+			$this->fatalError( Status::wrap( $result )->getMessage( false, false, 'en' )->text() );
+		}
+		return $result->value;
+	}
+
+	/**
+	 * Returns the revision ID next to $revId, according to $comp and $dir
+	 *
+	 * @param int $revId
+	 * @param string $comp the comparator, either '<' or '>', to go with $dir
+	 * @param string $dir the sort direction to go with $comp, either 'ARC' or 'DESC'
+	 *
+	 * @return int
+	 */
+	private function getNextRevision( int $revId, string $comp, string $dir ) {
+		$db = $this->loadBalancer->getConnectionRef( DB_REPLICA );
+		$next = $db->selectField(
+			'revision',
+			'rev_id',
+			"rev_id $comp $revId",
+			__METHOD__,
+			[ 'ORDER BY' => "rev_id $dir" ]
+		);
+		return (int)$next;
+	}
+
+	/**
+	 * @param RevisionRecord $rev
+	 *
+	 * @return int
+	 */
+	private function checkRevision( RevisionRecord $rev ) {
+		$count = 0;
+		foreach ( $rev->getSlots()->getSlots() as $slot ) {
+			$count += $this->checkSlot( $rev, $slot );
+		}
+
+		return $count;
+	}
+
+	/**
+	 * @param RevisionRecord $rev
+	 * @param SlotRecord $slot
+	 *
+	 * @return int
+	 */
+	private function checkSlot( RevisionRecord $rev, SlotRecord $slot ) {
+		$address = $slot->getAddress();
+		$error = null;
+
+		try {
+			$this->blobStore->getBlob( $address );
+			return 0; // nothing to do
+		} catch ( BlobAccessException $ex ) {
+			$error = $ex->getMessage();
+		} catch ( ExternalStoreException $ex ) {
+			$error = $ex->getMessage();
+		}
+
+		$this->output( "\t! Found bad blob on revision {$rev->getId()} ({$slot->getRole()} slot): "
+			. "content_id={$slot->getContentId()}, address=<{$slot->getAddress()}>, error='$error'\n" );
+
+		if ( $this->hasOption( 'mark' ) ) {
+			$newAddress = $this->markBlob( $rev, $slot, $error );
+			$this->output( "\tChanged address to <$newAddress>\n" );
+		}
+
+		return 1;
+	}
+
+	/**
+	 * @param RevisionRecord $rev
+	 * @param SlotRecord $slot
+	 * @param string|null $error
+	 *
+	 * @return false|string
+	 */
+	private function markBlob( RevisionRecord $rev, SlotRecord $slot, string $error = null ) {
+		$args = [];
+
+		if ( $this->hasOption( 'mark' ) ) {
+			$args['reason'] = $this->getOption( 'mark' );
+		}
+
+		if ( $error ) {
+			$args['error'] = $error;
+		}
+
+		$address = $slot->getAddress() ?: 'empty';
+		$badAddress = 'bad:' . urlencode( $address );
+
+		if ( $args ) {
+			$badAddress .= '?' . wfArrayToCgi( $args );
+		}
+
+		$badAddress = substr( $badAddress, 0, 255 );
+
+		$dbw = $this->loadBalancer->getConnectionRef( DB_MASTER );
+		$dbw->update(
+			'content',
+			[ 'content_address' => $badAddress ],
+			[ 'content_id' => $slot->getContentId() ],
+			__METHOD__
+		);
+
+		return $badAddress;
+	}
+
+	private function waitForReplication() {
+		return $this->lbFactory->waitForReplication();
+	}
+
+}
+
+$maintClass = FindBadBlobs::class;
+require_once RUN_MAINTENANCE_IF_MAIN;