Code Coverage
 
Lines
Functions and Methods
Classes and Traits
Total
7.27% covered (danger)
7.27%
16 / 220
6.25% covered (danger)
6.25%
1 / 16
CRAP
0.00% covered (danger)
0.00%
0 / 1
Updater
7.27% covered (danger)
7.27%
16 / 220
6.25% covered (danger)
6.25%
1 / 16
2124.78
0.00% covered (danger)
0.00%
0 / 1
 __construct
100.00% covered (success)
100.00%
2 / 2
100.00% covered (success)
100.00%
1 / 1
1
 build
0.00% covered (danger)
0.00%
0 / 3
0.00% covered (danger)
0.00%
0 / 1
2
 updateFromTitle
56.25% covered (warning)
56.25%
9 / 16
0.00% covered (danger)
0.00%
0 / 1
7.09
 updateRedirectDocument
83.33% covered (warning)
83.33%
5 / 6
0.00% covered (danger)
0.00%
0 / 1
3.04
 traceRedirects
0.00% covered (danger)
0.00%
0 / 29
0.00% covered (danger)
0.00%
0 / 1
90
 updatePages
0.00% covered (danger)
0.00%
0 / 44
0.00% covered (danger)
0.00%
0 / 1
20
 updateWeightedTags
0.00% covered (danger)
0.00%
0 / 23
0.00% covered (danger)
0.00%
0 / 1
2
 resetWeightedTags
0.00% covered (danger)
0.00%
0 / 7
0.00% covered (danger)
0.00%
0 / 1
6
 deletePages
0.00% covered (danger)
0.00%
0 / 16
0.00% covered (danger)
0.00%
0 / 1
2
 archivePages
0.00% covered (danger)
0.00%
0 / 16
0.00% covered (danger)
0.00%
0 / 1
6
 buildArchiveDocuments
0.00% covered (danger)
0.00%
0 / 15
0.00% covered (danger)
0.00%
0 / 1
12
 updateLinkedArticles
0.00% covered (danger)
0.00%
0 / 21
0.00% covered (danger)
0.00%
0 / 1
132
 pagesToTitles
0.00% covered (danger)
0.00%
0 / 4
0.00% covered (danger)
0.00%
0 / 1
6
 pushElasticaWriteJobs
0.00% covered (danger)
0.00%
0 / 6
0.00% covered (danger)
0.00%
0 / 1
12
 elasticaWriteClusters
0.00% covered (danger)
0.00%
0 / 6
0.00% covered (danger)
0.00%
0 / 1
6
 newLog
0.00% covered (danger)
0.00%
0 / 6
0.00% covered (danger)
0.00%
0 / 1
2
1<?php
2
3namespace CirrusSearch;
4
5use CirrusSearch\BuildDocument\BuildDocument;
6use CirrusSearch\BuildDocument\DocumentSizeLimiter;
7use CirrusSearch\Profile\SearchProfileService;
8use CirrusSearch\Search\CirrusIndexField;
9use MediaWiki\Content\TextContent;
10use MediaWiki\Logger\LoggerFactory;
11use MediaWiki\MediaWikiServices;
12use MediaWiki\Page\ProperPageIdentity;
13use MediaWiki\Page\WikiPage;
14use MediaWiki\Title\Title;
15use MediaWiki\WikiMap\WikiMap;
16use Wikimedia\Assert\Assert;
17
18/**
19 * Performs updates and deletes on the Elasticsearch index.  Called by
20 * CirrusSearch.php (our SearchEngine implementation), forceSearchIndex
21 * (for bulk updates), and CirrusSearch's jobs.
22 *
23 * @license GPL-2.0-or-later
24 */
25class Updater extends ElasticsearchIntermediary implements WeightedTagsUpdater {
26    /**
27     * Full title text of pages updated in this process.  Used for deduplication
28     * of updates.
29     * @var string[]
30     */
31    private $updated = [];
32
33    /**
34     * @var string|null Name of cluster to write to, or null if none (write to all)
35     */
36    protected $writeToClusterName;
37
38    /**
39     * @param Connection $readConnection connection used to pull data out of elasticsearch
40     * @param string|null $writeToClusterName
41     */
42    public function __construct( Connection $readConnection, $writeToClusterName = null ) {
43        parent::__construct( $readConnection, null, 0 );
44        $this->writeToClusterName = $writeToClusterName;
45    }
46
47    /**
48     * @param SearchConfig $config
49     * @param string|null $cluster cluster to read from and write to,
50     * null to read from the default cluster and write to all
51     * @return self
52     */
53    public static function build( SearchConfig $config, $cluster ): self {
54        Assert::invariant( self::class === static::class, 'Must be invoked as Updater::build( ... )' );
55        $connection = Connection::getPool( $config, $cluster );
56        return new self( $connection, $cluster );
57    }
58
59    /**
60     * Update a single page.
61     * @param Title $title
62     * @param string|null $updateKind kind of update to perform (used for monitoring)
63     * @param int|null $rootEventTime the time of MW event that caused this update (used for monitoring)
64     */
65    public function updateFromTitle( $title, ?string $updateKind, ?int $rootEventTime ): void {
66        [ $page, $redirects ] = $this->traceRedirects( $title );
67        if ( $page ) {
68            $this->updatePages(
69                [ $page ],
70                BuildDocument::INDEX_EVERYTHING,
71                $updateKind,
72                $rootEventTime
73            );
74        }
75
76        if ( $redirects === [] ) {
77            return;
78        }
79        if ( $this->connection->getConfig()->buildRedirectDocuments() ) {
80            // the redirect pages have their own documents, don't delete them
81            return;
82        }
83        $redirectDocIds = [];
84        foreach ( $redirects as $redirect ) {
85            $redirectDocIds[] = $this->connection->getConfig()->makeId( $redirect->getId() );
86        }
87        $this->deletePages( [], $redirectDocIds );
88    }
89
90    /**
91     * Index a redirect page's own document directly, without tracing to its target. The
92     * write is guarded against the edit-then-delete race: a stale job writes nothing.
93     *
94     * Only meaningful when redirect-document building is enabled; the caller is
95     * responsible for that gating.
96     *
97     * @param Title $title the redirect page to index
98     * @param string|null $updateKind kind of update to perform (used for monitoring)
99     * @param int|null $rootEventTime the time of the MW event that caused this update (monitoring)
100     */
101    public function updateRedirectDocument( $title, ?string $updateKind, ?int $rootEventTime ): void {
102        if ( !$this->connection->getConfig()->buildRedirectDocuments() ) {
103            // Was enabled when building, but not anymore.
104            return;
105        }
106        $page = MediaWikiServices::getInstance()->getWikiPageFactory()->newFromTitle( $title );
107        if ( !$page->exists() ) {
108            return;
109        }
110        $this->updatePages( [ $page ], BuildDocument::INDEX_EVERYTHING, $updateKind, $rootEventTime );
111    }
112
113    /**
114     * Trace redirects from the title to the destination.  Also registers the title in the
115     * memory of titles updated and detects special pages.
116     *
117     * @param Title $title title to trace
118     * @return array with keys: target, redirects
119     *    - target is WikiPage|null wikipage if the $title either isn't a redirect or resolves
120     *    to an updatable page that hasn't been updated yet.  Null if the page has been
121     *    updated, is a special page, or the redirects enter a loop.
122     *    - redirects is an array of WikiPages, one per redirect in the chain.  If title isn't
123     *    a redirect then this will be an empty array
124     */
125    public function traceRedirects( $title ) {
126        // Loop through redirects until we get to the ultimate target
127        $redirects = [];
128        $wikiPageFactory = MediaWikiServices::getInstance()->getWikiPageFactory();
129        while ( true ) {
130            $titleText = $title->getFullText();
131            if ( in_array( $titleText, $this->updated ) ) {
132                // Already indexed this article in this process.  This is mostly useful
133                // to catch self redirects but has a storied history of catching strange
134                // behavior.
135                return [ null, $redirects ];
136            }
137
138            // Don't index special pages, interwiki links, bad namespaces, etc
139            $logger = LoggerFactory::getInstance( LogChannel::DEFAULT );
140            if ( !$title->canExist() ) {
141                $logger->debug( "Ignoring an update for a page that cannot exist: $titleText" );
142                return [ null, $redirects ];
143            }
144
145            $page = $wikiPageFactory->newFromTitle( $title );
146            if ( !$page->exists() ) {
147                $logger->debug( "Ignoring an update for a nonexistent page: $titleText" );
148                return [ null, $redirects ];
149            }
150            $content = $page->getContent();
151            if ( is_string( $content ) ) {
152                $content = new TextContent( $content );
153            }
154            // If the event that the content is _still_ not usable, we have to give up.
155            if ( !is_object( $content ) ) {
156                return [ null, $redirects ];
157            }
158
159            // Add the page to the list of updated pages before we start trying to update to catch redirect loops.
160            $this->updated[] = $titleText;
161            if ( $content->isRedirect() ) {
162                $redirects[] = $page;
163                $target = $content->getRedirectTarget();
164                if ( $target->equals( $page->getTitle() ) ) {
165                    // This doesn't warn about redirect loops longer than one but we'll catch those anyway.
166                    $logger->info( "Title redirecting to itself. Skip indexing" );
167                    return [ null, $redirects ];
168                }
169                $title = $target;
170                continue;
171            } else {
172                return [ $page, $redirects ];
173            }
174        }
175    }
176
177    /**
178     * This updates pages in elasticsearch.
179     *
180     * $flags includes:
181     *   INDEX_EVERYTHING Cirrus will parse the page and count the links and send the document
182     *     to Elasticsearch as an index so if it doesn't exist it'll be created.
183     *   SKIP_PARSE Cirrus will skip parsing the page when building the document.  It makes
184     *     sense to do this when you know the page hasn't changed like when it is newly linked
185     *     from another page.
186     *   SKIP_LINKS Cirrus will skip collecting links information.  It makes sense to do this
187     *     when you know the link counts aren't yet available like during the first phase of
188     *     the two phase index build.
189     *   INDEX_ON_SKIP Cirrus will send an update if SKIP_PARSE or SKIP_LINKS rather than an
190     *     index.  Indexing with any portion of the document skipped is dangerous because it
191     *     can put half created pages in the index.  This is only a good idea during the first
192     *     half of the two phase index build.
193     *
194     * @param WikiPage[] $pages pages to update
195     * @param int $flags Bit field containing instructions about how the document should be built
196     *   and sent to Elasticsearch.
197     * @param string|null $updateKind kind of update to perform (used for monitoring)
198     * @param int|null $rootEventTime the time of MW event that caused this update (used for monitoring)
199     * @return int Number of documents updated
200     */
201    public function updatePages( $pages, $flags, ?string $updateKind = null, ?int $rootEventTime = null ): int {
202        // Don't update the same page twice. We shouldn't, but meh
203        $pageIds = [];
204        $pages = array_filter( $pages, static function ( WikiPage $page ) use ( &$pageIds ) {
205            if ( !in_array( $page->getId(), $pageIds ) ) {
206                $pageIds[] = $page->getId();
207                return true;
208            }
209            return false;
210        } );
211
212        $titles = $this->pagesToTitles( $pages );
213        Job\OtherIndex::queueIfRequired( $this->connection->getConfig(), $titles, $this->writeToClusterName );
214
215        $allDocuments = array_fill_keys( $this->connection->getAllIndexSuffixes(), [] );
216        $services = MediaWikiServices::getInstance();
217        $docSizeLimiter = new DocumentSizeLimiter(
218            $this->connection->getConfig()->getProfileService()->loadProfile( SearchProfileService::DOCUMENT_SIZE_LIMITER ) );
219        $builder = new BuildDocument(
220            $this->connection,
221            $services->getConnectionProvider()->getReplicaDatabase(),
222            $services->getRevisionStore(),
223            $services->getBacklinkCacheFactory(),
224            $docSizeLimiter,
225            $services->getTitleFormatter(),
226            $services->getWikiPageFactory(),
227            $services->getTitleFactory()
228        );
229        foreach ( $builder->initialize( $pages, $flags ) as $document ) {
230            // This isn't really a property of the connection, so it doesn't matter
231            // this is the read cluster and not the write cluster.
232            $suffix = $this->connection->getIndexSuffixForNamespace( $document->get( 'namespace' ) );
233            $allDocuments[$suffix][] = $document;
234        }
235
236        $count = 0;
237        foreach ( $allDocuments as $indexSuffix => $documents ) {
238            $this->pushElasticaWriteJobs(
239                UpdateGroup::PAGE,
240                $documents,
241                static function ( array $chunk, string $cluster ) use ( $indexSuffix, $updateKind, $rootEventTime ) {
242                    return Job\ElasticaWrite::build(
243                        $cluster,
244                        UpdateGroup::PAGE,
245                        'sendData',
246                        [ $indexSuffix, $chunk ],
247                        [],
248                        $updateKind,
249                        $rootEventTime
250                    );
251                } );
252            $count += count( $documents );
253        }
254
255        return $count;
256    }
257
258    /**
259     * @inheritDoc
260     */
261    public function updateWeightedTags(
262        ProperPageIdentity $page,
263        string $tagPrefix,
264        ?array $tagWeights = null,
265        ?string $trigger = null
266    ): void {
267        Assert::precondition( $page->exists(), "page must exist" );
268        $docId = $this->connection->getConfig()->makeId( $page->getId() );
269        $indexSuffix = $this->connection->getIndexSuffixForNamespace( $page->getNamespace() );
270        $this->pushElasticaWriteJobs(
271            UpdateGroup::WEIGHTED_TAGS,
272            [ $docId ],
273            static function ( array $docIds, string $cluster ) use (
274                $docId,
275                $indexSuffix,
276                $tagPrefix,
277                $tagWeights
278            ) {
279                return Job\ElasticaWrite::build(
280                    $cluster,
281                    UpdateGroup::WEIGHTED_TAGS,
282                    'sendWeightedTagsUpdate',
283                    [
284                        $indexSuffix,
285                        $tagPrefix,
286                        [ $docId => $tagWeights ]
287                    ],
288                );
289            } );
290    }
291
292    /**
293     * @inheritDoc
294     */
295    public function resetWeightedTags( ProperPageIdentity $page, array $tagPrefixes, ?string $trigger = null ): void {
296        foreach ( $tagPrefixes as $tagPrefix ) {
297            $this->updateWeightedTags(
298                $page,
299                $tagPrefix,
300                [ CirrusIndexField::MULTILIST_DELETE_GROUPING => null ],
301                $trigger
302            );
303        }
304    }
305
306    /**
307     * Delete pages from the elasticsearch index.  $titles and $docIds must point to the
308     * same pages and should point to them in the same order.
309     *
310     * @param Title[] $titles List of titles to delete.  If empty then skipped other index
311     *      maintenance is skipped.
312     * @param int[]|string[] $docIds List of elasticsearch document ids to delete
313     * @param string|null $indexSuffix index from which to delete.  null means all.
314     * @param array $writeJobParams Parameters passed on to ElasticaWriteJob
315     */
316    public function deletePages( $titles, $docIds, $indexSuffix = null, array $writeJobParams = [] ): void {
317        Job\OtherIndex::queueIfRequired( $this->connection->getConfig(), $titles, $this->writeToClusterName );
318
319        // Deletes are fairly cheap to send, they can be batched in larger
320        // chunks. Unlikely a batch this large ever comes through.
321        $batchSize = 50;
322        $this->pushElasticaWriteJobs(
323            UpdateGroup::PAGE,
324            $docIds,
325            static function ( array $chunk, string $cluster ) use ( $indexSuffix, $writeJobParams ) {
326                return Job\ElasticaWrite::build(
327                    $cluster,
328                    UpdateGroup::PAGE,
329                    'sendDeletes',
330                    [ $chunk, $indexSuffix ],
331                    $writeJobParams
332                );
333            },
334            $batchSize
335        );
336    }
337
338    /**
339     * Add documents to archive index.
340     * @param array $archived
341     * @return bool
342     */
343    public function archivePages( $archived ) {
344        if ( !$this->connection->getConfig()->getElement( CirrusConfigNames::IndexDeletes ) ) {
345            // Disabled by config - don't do anything
346            return true;
347        }
348        $docs = $this->buildArchiveDocuments( $archived );
349        $this->pushElasticaWriteJobs(
350            UpdateGroup::ARCHIVE,
351            $docs,
352            static function ( array $chunk, string $cluster ) {
353                return Job\ElasticaWrite::build(
354                    $cluster,
355                    UpdateGroup::ARCHIVE,
356                    'sendData',
357                    [ Connection::ARCHIVE_INDEX_SUFFIX, $chunk ],
358                    [ 'private_data' => true ],
359                );
360            } );
361
362        return true;
363    }
364
365    /**
366     * Build Elastica documents for archived pages.
367     * @param array $archived
368     * @return \Elastica\Document[]
369     */
370    private function buildArchiveDocuments( array $archived ) {
371        $docs = [];
372        foreach ( $archived as $delete ) {
373            if ( !isset( $delete['title'] ) ) {
374                // These come from pages that still exist, but are redirects.
375                // This is non-obvious and we probably need a better way...
376                continue;
377            }
378            /** @var Title $title */
379            $title = $delete['title'];
380            $doc = new \Elastica\Document( $delete['page'], [
381                'namespace' => $title->getNamespace(),
382                'title' => $title->getText(),
383                'wiki' => WikiMap::getCurrentWikiId(),
384            ] );
385            $doc->setDocAsUpsert( true );
386            $doc->setRetryOnConflict(
387                $this->connection->getConfig()->getElement( CirrusConfigNames::UpdateConflictRetryCount ) );
388
389            $docs[] = $doc;
390        }
391
392        return $docs;
393    }
394
395    /**
396     * Update the search index for newly linked or unlinked articles.
397     * @param Title[] $titles titles to update
398     */
399    public function updateLinkedArticles( $titles ): void {
400        $pages = [];
401        $wikiPageFactory = MediaWikiServices::getInstance()->getWikiPageFactory();
402        foreach ( $titles as $title ) {
403            // Special pages don't get updated, we only index
404            // actual existing pages.
405            if ( !$title || !$title->canExist() ) {
406                continue;
407            }
408
409            $page = $wikiPageFactory->newFromTitle( $title );
410            if ( $page === null || !$page->exists() ) {
411                // Skip link to nonexistent page.
412                continue;
413            }
414            // Resolve one level of redirects because only one level of redirects is scored.
415            if ( $page->isRedirect() ) {
416                $target = $page->getRedirectTarget();
417                if ( $target === null ) {
418                    // Redirect to itself or broken redirect? ignore.
419                    continue;
420                }
421                if ( !$target->exists() ) {
422                    // Skip redirects to nonexistent pages
423                    continue;
424                }
425                $page = $wikiPageFactory->newFromTitle( $target );
426            }
427            if ( $page->isRedirect() ) {
428                // This is a redirect to a redirect which doesn't count in the search score any way.
429                continue;
430            }
431            if ( in_array( $title->getFullText(), $this->updated ) ) {
432                // We've already updated this page in this process so there is no need to update it again.
433                continue;
434            }
435            // Note that we don't add this page to the list of updated pages because this update isn't
436            // a full update (just link counts).
437            $pages[] = $page;
438        }
439        $this->updatePages( $pages, BuildDocument::SKIP_PARSE );
440    }
441
442    /**
443     * Convert an array of pages to an array of their titles.
444     *
445     * @param WikiPage[] $pages
446     * @return Title[]
447     */
448    private function pagesToTitles( $pages ) {
449        $titles = [];
450        foreach ( $pages as $page ) {
451            $titles[] = $page->getTitle();
452        }
453        return $titles;
454    }
455
456    /**
457     * @param string $updateGroup UpdateGroup::* constant
458     * @param mixed[] $items
459     * @param callable $factory
460     * @param int $batchSize
461     */
462    protected function pushElasticaWriteJobs( string $updateGroup, array $items, $factory, int $batchSize = 10 ): void {
463        // Elasticsearch has a queue capacity of 50 so if $documents contains 50 pages it could bump up
464        // against the max.  So we chunk it and do them sequentially.
465        $config = $this->connection->getConfig();
466        $clusters = $this->elasticaWriteClusters( $updateGroup );
467
468        foreach ( array_chunk( $items, $batchSize ) as $chunked ) {
469            foreach ( $clusters as $cluster ) {
470                $job = $factory( $chunked, $cluster );
471                // If the job fails for any reason it will enqueue itself to retry later.
472                $job->run();
473            }
474        }
475    }
476
477    private function elasticaWriteClusters( string $updateGroup ): array {
478        if ( $this->writeToClusterName !== null ) {
479            return [ $this->writeToClusterName ];
480        } else {
481            return $this->connection
482                ->getConfig()
483                ->getClusterAssignment()
484                ->getWritableClusters( $updateGroup );
485        }
486    }
487
488    /**
489     * @param string $description
490     * @param string $queryType
491     * @param string[] $extra
492     * @return SearchRequestLog
493     */
494    protected function newLog( $description, $queryType, array $extra = [] ) {
495        return new SearchRequestLog(
496            $this->connection->getClient(),
497            $description,
498            $queryType,
499            $extra
500        );
501    }
502}