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