MediaWiki master
CategoryMembershipChangeJob.php
Go to the documentation of this file.
1<?php
22
38
55 private $ticket;
56
57 private const ENQUEUE_FUDGE_SEC = 60;
58
65 public static function newSpec( PageIdentity $page, $revisionTimestamp, bool $forImport ) {
66 return new JobSpecification(
67 'categoryMembershipChange',
68 [
69 'pageId' => $page->getId(),
70 'revTimestamp' => $revisionTimestamp,
71 'forImport' => $forImport,
72 ],
73 [
74 'removeDuplicates' => true,
75 'removeDuplicatesIgnoreParams' => [ 'revTimestamp' ]
76 ],
77 $page
78 );
79 }
80
87 public function __construct( PageIdentity $page, array $params ) {
88 parent::__construct( 'categoryMembershipChange', $page, $params );
89 // Only need one job per page. Note that ENQUEUE_FUDGE_SEC handles races where an
90 // older revision job gets inserted while the newer revision job is de-duplicated.
91 $this->removeDuplicates = true;
92 }
93
94 public function run() {
96 $lbFactory = $services->getDBLoadBalancerFactory();
97 $lb = $lbFactory->getMainLB();
98 $dbw = $lb->getConnection( DB_PRIMARY );
99
100 $this->ticket = $lbFactory->getEmptyTransactionTicket( __METHOD__ );
101
102 $page = $services->getWikiPageFactory()->newFromID( $this->params['pageId'], IDBAccessObject::READ_LATEST );
103 if ( !$page ) {
104 $this->setLastError( "Could not find page #{$this->params['pageId']}" );
105 return false; // deleted?
106 }
107
108 // Cut down on the time spent in waitForPrimaryPos() in the critical section
109 $dbr = $lb->getConnection( DB_REPLICA );
110 if ( !$lb->waitForPrimaryPos( $dbr ) ) {
111 $this->setLastError( "Timed out while pre-waiting for replica DB to catch up" );
112 return false;
113 }
114
115 // Use a named lock so that jobs for this page see each others' changes
116 $lockKey = "{$dbw->getDomainID()}:CategoryMembershipChange:{$page->getId()}"; // per-wiki
117 $scopedLock = $dbw->getScopedLockAndFlush( $lockKey, __METHOD__, 3 );
118 if ( !$scopedLock ) {
119 $this->setLastError( "Could not acquire lock '$lockKey'" );
120 return false;
121 }
122
123 // Wait till replica DB is caught up so that jobs for this page see each others' changes
124 if ( !$lb->waitForPrimaryPos( $dbr ) ) {
125 $this->setLastError( "Timed out while waiting for replica DB to catch up" );
126 return false;
127 }
128 // Clear any stale REPEATABLE-READ snapshot
129 $dbr->flushSnapshot( __METHOD__ );
130
131 $cutoffUnix = wfTimestamp( TS_UNIX, $this->params['revTimestamp'] );
132 // Using ENQUEUE_FUDGE_SEC handles jobs inserted out of revision order due to the delay
133 // between COMMIT and actual enqueueing of the CategoryMembershipChangeJob job.
134 $cutoffUnix -= self::ENQUEUE_FUDGE_SEC;
135
136 // Get the newest page revision that has a SRC_CATEGORIZE row.
137 // Assume that category changes before it were already handled.
138 $subQuery = $dbr->newSelectQueryBuilder()
139 ->select( '1' )
140 ->from( 'recentchanges' )
141 ->where( 'rc_this_oldid = rev_id' )
142 ->andWhere( [ 'rc_source' => RecentChange::SRC_CATEGORIZE ] );
143 $row = $dbr->newSelectQueryBuilder()
144 ->select( [ 'rev_timestamp', 'rev_id' ] )
145 ->from( 'revision' )
146 ->where( [ 'rev_page' => $page->getId() ] )
147 ->andWhere( $dbr->expr( 'rev_timestamp', '>=', $dbr->timestamp( $cutoffUnix ) ) )
148 ->andWhere( new RawSQLExpression( 'EXISTS (' . $subQuery->getSQL() . ')' ) )
149 ->orderBy( [ 'rev_timestamp', 'rev_id' ], SelectQueryBuilder::SORT_DESC )
150 ->caller( __METHOD__ )->fetchRow();
151
152 // Only consider revisions newer than any such revision
153 if ( $row ) {
154 $cutoffUnix = wfTimestamp( TS_UNIX, $row->rev_timestamp );
155 $lastRevId = (int)$row->rev_id;
156 } else {
157 $lastRevId = 0;
158 }
159
160 // Find revisions to this page made around and after this revision which lack category
161 // notifications in recent changes. This lets jobs pick up were the last one left off.
162 $revisionStore = $services->getRevisionStore();
163 $res = $revisionStore->newSelectQueryBuilder( $dbr )
164 ->joinComment()
165 ->where( [
166 'rev_page' => $page->getId(),
167 $dbr->buildComparison( '>', [
168 'rev_timestamp' => $dbr->timestamp( $cutoffUnix ),
169 'rev_id' => $lastRevId,
170 ] )
171 ] )
172 ->orderBy( [ 'rev_timestamp', 'rev_id' ], SelectQueryBuilder::SORT_ASC )
173 ->caller( __METHOD__ )->fetchResultSet();
174
175 // Apply all category updates in revision timestamp order
176 foreach ( $res as $row ) {
177 $this->notifyUpdatesForRevision( $lbFactory, $page, $revisionStore->newRevisionFromRow( $row ) );
178 }
179
180 return true;
181 }
182
188 protected function notifyUpdatesForRevision(
189 LBFactory $lbFactory, WikiPage $page, RevisionRecord $newRev
190 ) {
191 $title = $page->getTitle();
192
193 // Get the new revision
194 if ( $newRev->isDeleted( RevisionRecord::DELETED_TEXT ) ) {
195 return;
196 }
197
198 $services = MediaWikiServices::getInstance();
199 // Get the prior revision (the same for null edits)
200 if ( $newRev->getParentId() ) {
201 $oldRev = $services->getRevisionLookup()
202 ->getRevisionById( $newRev->getParentId(), IDBAccessObject::READ_LATEST );
203 if ( !$oldRev || $oldRev->isDeleted( RevisionRecord::DELETED_TEXT ) ) {
204 return;
205 }
206 } else {
207 $oldRev = null;
208 }
209
210 // Parse the new revision and get the categories
211 $categoryChanges = $this->getExplicitCategoriesChanges( $page, $newRev, $oldRev );
212 [ $categoryInserts, $categoryDeletes ] = $categoryChanges;
213 if ( !$categoryInserts && !$categoryDeletes ) {
214 return; // nothing to do
215 }
216
217 $blc = $services->getBacklinkCacheFactory()->getBacklinkCache( $title );
218 $catMembChange = new CategoryMembershipChange( $title, $blc, $newRev, $this->params['forImport'] ?? false );
219 $catMembChange->checkTemplateLinks();
220
221 $batchSize = $services->getMainConfig()->get( MainConfigNames::UpdateRowsPerQuery );
222 $insertCount = 0;
223
224 foreach ( $categoryInserts as $categoryName ) {
225 $categoryTitle = Title::makeTitle( NS_CATEGORY, $categoryName );
226 $catMembChange->triggerCategoryAddedNotification( $categoryTitle );
227 if ( $insertCount++ && ( $insertCount % $batchSize ) == 0 ) {
228 $lbFactory->commitAndWaitForReplication( __METHOD__, $this->ticket );
229 }
230 }
231
232 foreach ( $categoryDeletes as $categoryName ) {
233 $categoryTitle = Title::makeTitle( NS_CATEGORY, $categoryName );
234 $catMembChange->triggerCategoryRemovedNotification( $categoryTitle );
235 if ( $insertCount++ && ( $insertCount++ % $batchSize ) == 0 ) {
236 $lbFactory->commitAndWaitForReplication( __METHOD__, $this->ticket );
237 }
238 }
239 }
240
241 private function getExplicitCategoriesChanges(
242 WikiPage $page, RevisionRecord $newRev, ?RevisionRecord $oldRev = null
243 ): array {
244 // Inject the same timestamp for both revision parses to avoid seeing category changes
245 // due to time-based parser functions. Inject the same page title for the parses too.
246 // Note that REPEATABLE-READ makes template/file pages appear unchanged between parses.
247 $parseTimestamp = $newRev->getTimestamp();
248 // Parse the old rev and get the categories. Do not use link tables as that
249 // assumes these updates are perfectly FIFO and that link tables are always
250 // up to date, neither of which are true.
251 $oldCategories = $oldRev
252 ? $this->getCategoriesAtRev( $page, $oldRev, $parseTimestamp )
253 : [];
254 // Parse the new revision and get the categories
255 $newCategories = $this->getCategoriesAtRev( $page, $newRev, $parseTimestamp );
256
257 $categoryInserts = array_values( array_diff( $newCategories, $oldCategories ) );
258 $categoryDeletes = array_values( array_diff( $oldCategories, $newCategories ) );
259
260 return [ $categoryInserts, $categoryDeletes ];
261 }
262
270 private function getCategoriesAtRev( WikiPage $page, RevisionRecord $rev, $parseTimestamp ) {
271 $services = MediaWikiServices::getInstance();
272 $options = $page->makeParserOptions( 'canonical' );
273 $options->setTimestamp( $parseTimestamp );
274 $options->setRenderReason( 'CategoryMembershipChangeJob' );
275
276 $output = $rev instanceof RevisionStoreRecord && $rev->isCurrent()
277 ? $services->getParserCache()->get( $page, $options )
278 : null;
279
280 if ( !$output || $output->getCacheRevisionId() !== $rev->getId() ) {
281 $output = $services->getRevisionRenderer()->getRenderedRevision( $rev, $options )
282 ->getRevisionParserOutput();
283 }
284
285 // array keys will cast numeric category names to ints;
286 // ::getCategoryNames() is careful to cast them back to strings
287 // to avoid breaking things!
288 return $output->getCategoryNames();
289 }
290
291 public function getDeduplicationInfo() {
292 $info = parent::getDeduplicationInfo();
293 unset( $info['params']['revTimestamp'] ); // first job wins
294
295 return $info;
296 }
297}
298
300class_alias( CategoryMembershipChangeJob::class, 'CategoryMembershipChangeJob' );
const NS_CATEGORY
Definition Defines.php:79
wfTimestamp( $outputtype=TS_UNIX, $ts=0)
Get a timestamp string in one of various formats.
Job queue task description base code.
Describe and execute a background job.
Definition Job.php:41
array $params
Array of job parameters.
Definition Job.php:46
setLastError( $error)
Definition Job.php:435
Job to add recent change entries mentioning category membership changes.
static newSpec(PageIdentity $page, $revisionTimestamp, bool $forImport)
notifyUpdatesForRevision(LBFactory $lbFactory, WikiPage $page, RevisionRecord $newRev)
__construct(PageIdentity $page, array $params)
Constructor for use by the Job Queue infrastructure.
getDeduplicationInfo()
Subclasses may need to override this to make duplication detection work.
A class containing constants representing the names of configuration variables.
const UpdateRowsPerQuery
Name constant for the UpdateRowsPerQuery setting, for use with Config::get()
Service locator for MediaWiki core services.
static getInstance()
Returns the global default instance of the top level service locator.
Base representation for an editable wiki page.
Definition WikiPage.php:94
getTitle()
Get the title object of the article.
Definition WikiPage.php:262
Helper class for category membership changes.
Utility class for creating and reading rows in the recentchanges table.
Page revision base class.
getParentId( $wikiId=self::LOCAL)
Get parent revision ID (the original previous page revision).
isDeleted( $field)
MCR migration note: this replaced Revision::isDeleted.
A RevisionRecord representing an existing revision persisted in the revision table.
Represents a title within MediaWiki.
Definition Title.php:78
commitAndWaitForReplication( $fname, $ticket, array $opts=[])
Commit primary DB transactions and wait for replication (if $ticket indicates it is safe).
Raw SQL expression to be used in query builders.
Build SELECT queries with a fluent interface.
Interface for objects (potentially) representing an editable wiki page.
getId( $wikiId=self::LOCAL)
Returns the page ID.
Interface for database access objects.
const DB_REPLICA
Definition defines.php:26
const DB_PRIMARY
Definition defines.php:28