Merge "Stash WatchedItem changes so that the jobs run from the queue"

diff --git a/includes/ServiceWiring.php b/includes/ServiceWiring.php
index 46dd913..e5f891e 100644 (file)
--- a/includes/ServiceWiring.php
+++ b/includes/ServiceWiring.php
@@ -598,13 +598,16 @@ return [
                return new WatchedItemQueryService(
                        $services->getDBLoadBalancer(),
                        $services->getCommentStore(),
-                       $services->getActorMigration()
+                       $services->getActorMigration(),
+                       $services->getWatchedItemStore()
                );
        },
 
        'WatchedItemStore' => function ( MediaWikiServices $services ) : WatchedItemStore {
                $store = new WatchedItemStore(
                        $services->getDBLoadBalancerFactory(),
+                       JobQueueGroup::singleton(),
+                       $services->getMainObjectStash(),
                        new HashBagOStuff( [ 'maxKeys' => 100 ] ),
                        $services->getReadOnlyMode(),
                        $services->getMainConfig()->get( 'UpdateRowsPerQuery' )

diff --git a/includes/watcheditem/NoWriteWatchedItemStore.php b/includes/watcheditem/NoWriteWatchedItemStore.php
index 2801207..39d7a5d 100644 (file)
--- a/includes/watcheditem/NoWriteWatchedItemStore.php
+++ b/includes/watcheditem/NoWriteWatchedItemStore.php
@@ -152,4 +152,7 @@ class NoWriteWatchedItemStore implements WatchedItemStoreInterface {
                throw new DBReadOnlyError( null, self::DB_READONLY_ERROR );
        }
 
+       public function getLatestNotificationTimestamp( $timestamp, User $user, LinkTarget $target ) {
+               return wfTimestampOrNull( TS_MW, $timestamp );
+       }
 }

diff --git a/includes/watcheditem/WatchedItemQueryService.php b/includes/watcheditem/WatchedItemQueryService.php
index 8b54a95..3ebc94a 100644 (file)
--- a/includes/watcheditem/WatchedItemQueryService.php
+++ b/includes/watcheditem/WatchedItemQueryService.php
@@ -65,14 +65,19 @@ class WatchedItemQueryService {
        /** @var ActorMigration */
        private $actorMigration;
 
+       /** @var WatchedItemStoreInterface */
+       private $watchedItemStore;
+
        public function __construct(
                LoadBalancer $loadBalancer,
                CommentStore $commentStore,
-               ActorMigration $actorMigration
+               ActorMigration $actorMigration,
+               WatchedItemStoreInterface $watchedItemStore
        ) {
                $this->loadBalancer = $loadBalancer;
                $this->commentStore = $commentStore;
                $this->actorMigration = $actorMigration;
+               $this->watchedItemStore = $watchedItemStore;
        }
 
        /**
@@ -228,11 +233,14 @@ class WatchedItemQueryService {
                                break;
                        }
 
+                       $target = new TitleValue( (int)$row->rc_namespace, $row->rc_title );
                        $items[] = [
                                new WatchedItem(
                                        $user,
-                                       new TitleValue( (int)$row->rc_namespace, $row->rc_title ),
-                                       $row->wl_notificationtimestamp
+                                       $target,
+                                       $this->watchedItemStore->getLatestNotificationTimestamp(
+                                               $row->wl_notificationtimestamp, $user, $target
+                                       )
                                ),
                                $this->getRecentChangeFieldsFromRow( $row )
                        ];
@@ -307,11 +315,14 @@ class WatchedItemQueryService {
 
                $watchedItems = [];
                foreach ( $res as $row ) {
+                       $target = new TitleValue( (int)$row->wl_namespace, $row->wl_title );
                        // todo these could all be cached at some point?
                        $watchedItems[] = new WatchedItem(
                                $user,
-                               new TitleValue( (int)$row->wl_namespace, $row->wl_title ),
-                               $row->wl_notificationtimestamp
+                               $target,
+                               $this->watchedItemStore->getLatestNotificationTimestamp(
+                                       $row->wl_notificationtimestamp, $user, $target
+                               )
                        );
                }
 

diff --git a/includes/watcheditem/WatchedItemQueryServiceExtension.php b/includes/watcheditem/WatchedItemQueryServiceExtension.php
index 873ae2d..a0e64c5 100644 (file)
--- a/includes/watcheditem/WatchedItemQueryServiceExtension.php
+++ b/includes/watcheditem/WatchedItemQueryServiceExtension.php
@@ -1,6 +1,6 @@
 <?php
 
-use Wikimedia\Rdbms\ResultWrapper;
+use Wikimedia\Rdbms\IResultWrapper;
 use Wikimedia\Rdbms\IDatabase;
 
 /**
@@ -45,7 +45,7 @@ interface WatchedItemQueryServiceExtension {
         * @param IDatabase $db Database connection being used for the query
         * @param array &$items array of pairs ( WatchedItem $watchedItem, string[] $recentChangeInfo ).
         *  May be truncated if necessary, in which case $startFrom must be updated.
-        * @param ResultWrapper|bool $res Database query result
+        * @param IResultWrapper|bool $res Database query result
         * @param array|null &$startFrom Continuation value. If you truncate $items, set this to
         *  [ $recentChangeInfo['rc_timestamp'], $recentChangeInfo['rc_id'] ] from the first item
         *  removed.

diff --git a/includes/watcheditem/WatchedItemStore.php b/includes/watcheditem/WatchedItemStore.php
index 1c33754..274a35d 100644 (file)
--- a/includes/watcheditem/WatchedItemStore.php
+++ b/includes/watcheditem/WatchedItemStore.php
@@ -28,6 +28,16 @@ class WatchedItemStore implements WatchedItemStoreInterface, StatsdAwareInterfac
         */
        private $loadBalancer;
 
+       /**
+        * @var JobQueueGroup
+        */
+       private $queueGroup;
+
+       /**
+        * @var BagOStuff
+        */
+       private $stash;
+
        /**
         * @var ReadOnlyMode
         */
@@ -38,6 +48,11 @@ class WatchedItemStore implements WatchedItemStoreInterface, StatsdAwareInterfac
         */
        private $cache;
 
+       /**
+        * @var HashBagOStuff
+        */
+       private $latestUpdateCache;
+
        /**
         * @var array[] Looks like $cacheIndex[Namespace ID][Target DB Key][User Id] => 'key'
         * The index is needed so that on mass changes all relevant items can be un-cached.
@@ -68,18 +83,24 @@ class WatchedItemStore implements WatchedItemStoreInterface, StatsdAwareInterfac
 
        /**
         * @param ILBFactory $lbFactory
+        * @param JobQueueGroup $queueGroup
+        * @param BagOStuff $stash
         * @param HashBagOStuff $cache
         * @param ReadOnlyMode $readOnlyMode
         * @param int $updateRowsPerQuery
         */
        public function __construct(
                ILBFactory $lbFactory,
+               JobQueueGroup $queueGroup,
+               BagOStuff $stash,
                HashBagOStuff $cache,
                ReadOnlyMode $readOnlyMode,
                $updateRowsPerQuery
        ) {
                $this->lbFactory = $lbFactory;
                $this->loadBalancer = $lbFactory->getMainLB();
+               $this->queueGroup = $queueGroup;
+               $this->stash = $stash;
                $this->cache = $cache;
                $this->readOnlyMode = $readOnlyMode;
                $this->stats = new NullStatsdDataFactory();
@@ -88,6 +109,8 @@ class WatchedItemStore implements WatchedItemStoreInterface, StatsdAwareInterfac
                $this->revisionGetTimestampFromIdCallback =
                        [ Revision::class, 'getTimestampFromId' ];
                $this->updateRowsPerQuery = $updateRowsPerQuery;
+
+               $this->latestUpdateCache = new HashBagOStuff( [ 'maxKeys' => 3 ] );
        }
 
        /**
@@ -286,8 +309,7 @@ class WatchedItemStore implements WatchedItemStoreInterface, StatsdAwareInterfac
         */
        public function clearUserWatchedItemsUsingJobQueue( User $user ) {
                $job = ClearUserWatchlistJob::newForUser( $user, $this->getMaxId() );
-               // TODO inject me.
-               JobQueueGroup::singleton()->push( $job );
+               $this->queueGroup->push( $job );
        }
 
        /**
@@ -568,6 +590,7 @@ class WatchedItemStore implements WatchedItemStoreInterface, StatsdAwareInterfac
                }
 
                $dbr = $this->getConnectionRef( DB_REPLICA );
+
                $row = $dbr->selectRow(
                        'watchlist',
                        'wl_notificationtimestamp',
@@ -582,7 +605,7 @@ class WatchedItemStore implements WatchedItemStoreInterface, StatsdAwareInterfac
                $item = new WatchedItem(
                        $user,
                        $target,
-                       wfTimestampOrNull( TS_MW, $row->wl_notificationtimestamp )
+                       $this->getLatestNotificationTimestamp( $row->wl_notificationtimestamp, $user, $target )
                );
                $this->cache( $item );
 
@@ -622,11 +645,13 @@ class WatchedItemStore implements WatchedItemStoreInterface, StatsdAwareInterfac
 
                $watchedItems = [];
                foreach ( $res as $row ) {
+                       $target = new TitleValue( (int)$row->wl_namespace, $row->wl_title );
                        // @todo: Should we add these to the process cache?
                        $watchedItems[] = new WatchedItem(
                                $user,
                                new TitleValue( (int)$row->wl_namespace, $row->wl_title ),
-                               $row->wl_notificationtimestamp
+                               $this->getLatestNotificationTimestamp(
+                                       $row->wl_notificationtimestamp, $user, $target )
                        );
                }
 
@@ -688,8 +713,10 @@ class WatchedItemStore implements WatchedItemStoreInterface, StatsdAwareInterfac
                );
 
                foreach ( $res as $row ) {
+                       $target = new TitleValue( (int)$row->wl_namespace, $row->wl_title );
                        $timestamps[$row->wl_namespace][$row->wl_title] =
-                               wfTimestampOrNull( TS_MW, $row->wl_notificationtimestamp );
+                               $this->getLatestNotificationTimestamp(
+                                       $row->wl_notificationtimestamp, $user, $target );
                }
 
                return $timestamps;
@@ -802,7 +829,7 @@ class WatchedItemStore implements WatchedItemStoreInterface, StatsdAwareInterfac
                        $timestamp = $dbw->timestamp( $timestamp );
                }
 
-               $success = $dbw->update(
+               $dbw->update(
                        'watchlist',
                        [ 'wl_notificationtimestamp' => $timestamp ],
                        $conds,
@@ -811,7 +838,25 @@ class WatchedItemStore implements WatchedItemStoreInterface, StatsdAwareInterfac
 
                $this->uncacheUser( $user );
 
-               return $success;
+               return true;
+       }
+
+       public function getLatestNotificationTimestamp( $timestamp, User $user, LinkTarget $target ) {
+               $timestamp = wfTimestampOrNull( TS_MW, $timestamp );
+               if ( $timestamp === null ) {
+                       return null; // no notification
+               }
+
+               $seenTimestamps = $this->getPageSeenTimestamps( $user );
+               if (
+                       $seenTimestamps &&
+                       $seenTimestamps->get( $this->getPageSeenKey( $target ) ) >= $timestamp
+               ) {
+                       // If a reset job did not yet run, then the "seen" timestamp will be higher
+                       return null;
+               }
+
+               return $timestamp;
        }
 
        public function resetAllNotificationTimestampsForUser( User $user ) {
@@ -902,6 +947,8 @@ class WatchedItemStore implements WatchedItemStoreInterface, StatsdAwareInterfac
         * @return bool
         */
        public function resetNotificationTimestamp( User $user, Title $title, $force = '', $oldid = 0 ) {
+               $time = time();
+
                // Only loggedin user can have a watchlist
                if ( $this->readOnlyMode->isReadOnly() || $user->isAnon() ) {
                        return false;
@@ -919,6 +966,20 @@ class WatchedItemStore implements WatchedItemStoreInterface, StatsdAwareInterfac
                        }
                }
 
+               // Mark the item as read immediately in lightweight storage
+               $this->stash->merge(
+                       $this->getPageSeenTimestampsKey( $user ),
+                       function ( $cache, $key, $current ) use ( $time, $title ) {
+                               $value = $current ?: new MapCacheLRU( 300 );
+                               $value->set( $this->getPageSeenKey( $title ), wfTimestamp( TS_MW, $time ) );
+
+                               $this->latestUpdateCache->set( $key, $value, IExpiringStore::TTL_PROC_LONG );
+
+                               return $value;
+                       },
+                       IExpiringStore::TTL_HOUR
+               );
+
                // If the page is watched by the user (or may be watched), update the timestamp
                $job = new ActivityUpdateJob(
                        $title,
@@ -926,22 +987,51 @@ class WatchedItemStore implements WatchedItemStoreInterface, StatsdAwareInterfac
                                'type'      => 'updateWatchlistNotification',
                                'userid'    => $user->getId(),
                                'notifTime' => $this->getNotificationTimestamp( $user, $title, $item, $force, $oldid ),
-                               'curTime'   => time()
+                               'curTime'   => $time
                        ]
                );
+               // Try to enqueue this post-send
+               $this->queueGroup->lazyPush( $job );
 
-               // Try to run this post-send
-               // Calls DeferredUpdates::addCallableUpdate in normal operation
-               call_user_func(
-                       $this->deferredUpdatesAddCallableUpdateCallback,
-                       function () use ( $job ) {
-                               $job->run();
+               $this->uncache( $user, $title );
+
+               return true;
+       }
+
+       /**
+        * @param User $user
+        * @return MapCacheLRU|null
+        */
+       private function getPageSeenTimestamps( User $user ) {
+               $key = $this->getPageSeenTimestampsKey( $user );
+
+               return $this->latestUpdateCache->getWithSetCallback(
+                       $key,
+                       IExpiringStore::TTL_PROC_LONG,
+                       function () use ( $key ) {
+                               return $this->stash->get( $key ) ?: null;
                        }
                );
+       }
 
-               $this->uncache( $user, $title );
+       /**
+        * @param User $user
+        * @return string
+        */
+       private function getPageSeenTimestampsKey( User $user ) {
+               return $this->stash->makeGlobalKey(
+                       'watchlist-recent-updates',
+                       $this->lbFactory->getLocalDomainID(),
+                       $user->getId()
+               );
+       }
 
-               return true;
+       /**
+        * @param LinkTarget $target
+        * @return string
+        */
+       private function getPageSeenKey( LinkTarget $target ) {
+               return "{$target->getNamespace()}:{$target->getDBkey()}";
        }
 
        private function getNotificationTimestamp( User $user, Title $title, $item, $force, $oldid ) {
@@ -998,25 +1088,22 @@ class WatchedItemStore implements WatchedItemStoreInterface, StatsdAwareInterfac
         * @return int|bool
         */
        public function countUnreadNotifications( User $user, $unreadLimit = null ) {
+               $dbr = $this->getConnectionRef( DB_REPLICA );
+
                $queryOptions = [];
                if ( $unreadLimit !== null ) {
                        $unreadLimit = (int)$unreadLimit;
                        $queryOptions['LIMIT'] = $unreadLimit;
                }
 
-               $dbr = $this->getConnectionRef( DB_REPLICA );
-               $rowCount = $dbr->selectRowCount(
-                       'watchlist',
-                       '1',
-                       [
-                               'wl_user' => $user->getId(),
-                               'wl_notificationtimestamp IS NOT NULL',
-                       ],
-                       __METHOD__,
-                       $queryOptions
-               );
+               $conds = [
+                       'wl_user' => $user->getId(),
+                       'wl_notificationtimestamp IS NOT NULL'
+               ];
+
+               $rowCount = $dbr->selectRowCount( 'watchlist', '1', $conds, __METHOD__, $queryOptions );
 
-               if ( !isset( $unreadLimit ) ) {
+               if ( $unreadLimit === null ) {
                        return $rowCount;
                }
 

diff --git a/includes/watcheditem/WatchedItemStoreInterface.php b/includes/watcheditem/WatchedItemStoreInterface.php
index 274d3f4..349d98a 100644 (file)
--- a/includes/watcheditem/WatchedItemStoreInterface.php
+++ b/includes/watcheditem/WatchedItemStoreInterface.php
@@ -326,4 +326,18 @@ interface WatchedItemStoreInterface {
         */
        public function removeWatchBatchForUser( User $user, array $targets );
 
+       /**
+        * Convert $timestamp to TS_MW or return null if the page was visited since then by $user
+        *
+        * Use this only on single-user methods (having higher read-after-write expectations)
+        * and not in places involving arbitrary batches of different users
+        *
+        * Usage of this method should be limited to WatchedItem* classes
+        *
+        * @param string|null $timestamp Value of wl_notificationtimestamp from the DB
+        * @param User $user
+        * @param LinkTarget $target
+        * @return string TS_MW timestamp or null
+        */
+       public function getLatestNotificationTimestamp( $timestamp, User $user, LinkTarget $target );
 }

diff --git a/tests/phpunit/includes/watcheditem/WatchedItemQueryServiceUnitTest.php b/tests/phpunit/includes/watcheditem/WatchedItemQueryServiceUnitTest.php
index e814a57..16f2367 100644 (file)
--- a/tests/phpunit/includes/watcheditem/WatchedItemQueryServiceUnitTest.php
+++ b/tests/phpunit/includes/watcheditem/WatchedItemQueryServiceUnitTest.php
@@ -72,7 +72,8 @@ class WatchedItemQueryServiceUnitTest extends MediaWikiTestCase {
                return new WatchedItemQueryService(
                        $this->getMockLoadBalancer( $mockDb ),
                        $this->getMockCommentStore(),
-                       $this->getMockActorMigration()
+                       $this->getMockActorMigration(),
+                       $this->getMockWatchedItemStore()
                );
        }
 
@@ -139,6 +140,22 @@ class WatchedItemQueryServiceUnitTest extends MediaWikiTestCase {
                return $mock;
        }
 
+       /**
+        * @param PHPUnit_Framework_MockObject_MockObject|Database $mockDb
+        * @return PHPUnit_Framework_MockObject_MockObject|WatchedItemStore
+        */
+       private function getMockWatchedItemStore() {
+               $mock = $this->getMockBuilder( WatchedItemStore::class )
+                       ->disableOriginalConstructor()
+                       ->getMock();
+               $mock->expects( $this->any() )
+                       ->method( 'getLatestNotificationTimestamp' )
+                       ->will( $this->returnCallback( function ( $timestamp ) {
+                               return $timestamp;
+                       } ) );
+               return $mock;
+       }
+
        /**
         * @param int $id
         * @return PHPUnit_Framework_MockObject_MockObject|User

diff --git a/tests/phpunit/includes/watcheditem/WatchedItemStoreIntegrationTest.php b/tests/phpunit/includes/watcheditem/WatchedItemStoreIntegrationTest.php
index 3102929..6a383a2 100644 (file)
--- a/tests/phpunit/includes/watcheditem/WatchedItemStoreIntegrationTest.php
+++ b/tests/phpunit/includes/watcheditem/WatchedItemStoreIntegrationTest.php
@@ -167,6 +167,13 @@ class WatchedItemStoreIntegrationTest extends MediaWikiTestCase {
                        [ $title->getNamespace() => [ $title->getDBkey() => null ] ],
                        $store->getNotificationTimestampsBatch( $user, [ $title ] )
                );
+
+               // Run the job queue
+               JobQueueGroup::destroySingletons();
+               $jobs = new RunJobs;
+               $jobs->loadParamsAndArgs( null, [ 'quiet' => true ], null );
+               $jobs->execute();
+
                $this->assertEquals(
                        $initialVisitingWatchers,
                        $store->countVisitingWatchers( $title, '20150202020202' )

diff --git a/tests/phpunit/includes/watcheditem/WatchedItemStoreUnitTest.php b/tests/phpunit/includes/watcheditem/WatchedItemStoreUnitTest.php
index 240b3f5..280ad90 100644 (file)
--- a/tests/phpunit/includes/watcheditem/WatchedItemStoreUnitTest.php
+++ b/tests/phpunit/includes/watcheditem/WatchedItemStoreUnitTest.php
@@ -59,6 +59,26 @@ class WatchedItemStoreUnitTest extends MediaWikiTestCase {
                return $mock;
        }
 
+       /**
+        * @return PHPUnit_Framework_MockObject_MockObject|JobQueueGroup
+        */
+       private function getMockJobQueueGroup() {
+               $mock = $this->getMockBuilder( JobQueueGroup::class )
+                       ->disableOriginalConstructor()
+                       ->getMock();
+               $mock->expects( $this->any() )
+                       ->method( 'push' )
+                       ->will( $this->returnCallback( function ( Job $job ) {
+                               $job->run();
+                       } ) );
+               $mock->expects( $this->any() )
+                       ->method( 'lazyPush' )
+                       ->will( $this->returnCallback( function ( Job $job ) {
+                               $job->run();
+                       } ) );
+               return $mock;
+       }
+
        /**
         * @return PHPUnit_Framework_MockObject_MockObject|HashBagOStuff
         */
@@ -118,11 +138,16 @@ class WatchedItemStoreUnitTest extends MediaWikiTestCase {
                return $fakeRow;
        }
 
-       private function newWatchedItemStore( LBFactory $lbFactory, HashBagOStuff $cache,
+       private function newWatchedItemStore(
+               LBFactory $lbFactory,
+               JobQueueGroup $queueGroup,
+               HashBagOStuff $cache,
                ReadOnlyMode $readOnlyMode
        ) {
                return new WatchedItemStore(
                        $lbFactory,
+                       $queueGroup,
+                       new HashBagOStuff(),
                        $cache,
                        $readOnlyMode,
                        1000
@@ -161,6 +186,7 @@ class WatchedItemStoreUnitTest extends MediaWikiTestCase {
 
                $store = $this->newWatchedItemStore(
                        $this->getMockLBFactory( $mockDb ),
+                       $this->getMockJobQueueGroup(),
                        $mockCache,
                        $this->getMockReadOnlyMode()
                );
@@ -193,6 +219,7 @@ class WatchedItemStoreUnitTest extends MediaWikiTestCase {
 
                $store = $this->newWatchedItemStore(
                        $this->getMockLBFactory( $mockDb ),
+                       $this->getMockJobQueueGroup(),
                        $mockCache,
                        $this->getMockReadOnlyMode()
                );
@@ -223,6 +250,7 @@ class WatchedItemStoreUnitTest extends MediaWikiTestCase {
 
                $store = $this->newWatchedItemStore(
                        $this->getMockLBFactory( $mockDb ),
+                       $this->getMockJobQueueGroup(),
                        $mockCache,
                        $this->getMockReadOnlyMode()
                );
@@ -254,6 +282,7 @@ class WatchedItemStoreUnitTest extends MediaWikiTestCase {
 
                $store = $this->newWatchedItemStore(
                        $this->getMockLBFactory( $mockDb ),
+                       $this->getMockJobQueueGroup(),
                        $mockCache,
                        $this->getMockReadOnlyMode()
                );
@@ -306,6 +335,7 @@ class WatchedItemStoreUnitTest extends MediaWikiTestCase {
 
                $store = $this->newWatchedItemStore(
                        $this->getMockLBFactory( $mockDb ),
+                       $this->getMockJobQueueGroup(),
                        $mockCache,
                        $this->getMockReadOnlyMode()
                );
@@ -373,6 +403,7 @@ class WatchedItemStoreUnitTest extends MediaWikiTestCase {
 
                $store = $this->newWatchedItemStore(
                        $this->getMockLBFactory( $mockDb ),
+                       $this->getMockJobQueueGroup(),
                        $mockCache,
                        $this->getMockReadOnlyMode()
                );
@@ -422,6 +453,7 @@ class WatchedItemStoreUnitTest extends MediaWikiTestCase {
 
                $store = $this->newWatchedItemStore(
                        $this->getMockLBFactory( $mockDb ),
+                       $this->getMockJobQueueGroup(),
                        $mockCache,
                        $this->getMockReadOnlyMode()
                );
@@ -504,6 +536,7 @@ class WatchedItemStoreUnitTest extends MediaWikiTestCase {
 
                $store = $this->newWatchedItemStore(
                        $this->getMockLBFactory( $mockDb ),
+                       $this->getMockJobQueueGroup(),
                        $mockCache,
                        $this->getMockReadOnlyMode()
                );
@@ -609,6 +642,7 @@ class WatchedItemStoreUnitTest extends MediaWikiTestCase {
 
                $store = $this->newWatchedItemStore(
                        $this->getMockLBFactory( $mockDb ),
+                       $this->getMockJobQueueGroup(),
                        $mockCache,
                        $this->getMockReadOnlyMode()
                );
@@ -663,6 +697,7 @@ class WatchedItemStoreUnitTest extends MediaWikiTestCase {
 
                $store = $this->newWatchedItemStore(
                        $this->getMockLBFactory( $mockDb ),
+                       $this->getMockJobQueueGroup(),
                        $mockCache,
                        $this->getMockReadOnlyMode()
                );
@@ -701,6 +736,7 @@ class WatchedItemStoreUnitTest extends MediaWikiTestCase {
 
                $store = $this->newWatchedItemStore(
                        $this->getMockLBFactory( $mockDb ),
+                       $this->getMockJobQueueGroup(),
                        $mockCache,
                        $this->getMockReadOnlyMode()
                );
@@ -736,6 +772,7 @@ class WatchedItemStoreUnitTest extends MediaWikiTestCase {
 
                $store = $this->newWatchedItemStore(
                        $this->getMockLBFactory( $mockDb ),
+                       $this->getMockJobQueueGroup(),
                        $mockCache,
                        $this->getMockReadOnlyMode()
                );
@@ -774,6 +811,7 @@ class WatchedItemStoreUnitTest extends MediaWikiTestCase {
 
                $store = $this->newWatchedItemStore(
                        $this->getMockLBFactory( $mockDb ),
+                       $this->getMockJobQueueGroup(),
                        $mockCache,
                        $this->getMockReadOnlyMode()
                );
@@ -805,6 +843,7 @@ class WatchedItemStoreUnitTest extends MediaWikiTestCase {
 
                $store = $this->newWatchedItemStore(
                        $this->getMockLBFactory( $mockDb ),
+                       $this->getMockJobQueueGroup(),
                        $this->getMockCache(),
                        $this->getMockReadOnlyMode()
                );
@@ -864,6 +903,7 @@ class WatchedItemStoreUnitTest extends MediaWikiTestCase {
 
                $store = $this->newWatchedItemStore(
                        $this->getMockLBFactory( $mockDb ),
+                       $this->getMockJobQueueGroup(),
                        $mockCache,
                        $this->getMockReadOnlyMode()
                );
@@ -911,6 +951,7 @@ class WatchedItemStoreUnitTest extends MediaWikiTestCase {
 
                $store = $this->newWatchedItemStore(
                        $this->getMockLBFactory( $mockDb ),
+                       $this->getMockJobQueueGroup(),
                        $mockCache,
                        $this->getMockReadOnlyMode()
                );
@@ -1005,6 +1046,7 @@ class WatchedItemStoreUnitTest extends MediaWikiTestCase {
 
                $store = $this->newWatchedItemStore(
                        $this->getMockLBFactory( $mockDb ),
+                       $this->getMockJobQueueGroup(),
                        $mockCache,
                        $this->getMockReadOnlyMode()
                );
@@ -1038,6 +1080,7 @@ class WatchedItemStoreUnitTest extends MediaWikiTestCase {
 
                $store = $this->newWatchedItemStore(
                        $this->getMockLBFactory( $mockDb ),
+                       $this->getMockJobQueueGroup(),
                        $mockCache,
                        $this->getMockReadOnlyMode()
                );
@@ -1059,6 +1102,7 @@ class WatchedItemStoreUnitTest extends MediaWikiTestCase {
 
                $store = $this->newWatchedItemStore(
                        $this->getMockLBFactory( $mockDb ),
+                       $this->getMockJobQueueGroup(),
                        $mockCache,
                        $this->getMockReadOnlyMode()
                );
@@ -1072,6 +1116,7 @@ class WatchedItemStoreUnitTest extends MediaWikiTestCase {
        public function testAddWatchBatchForUser_readOnlyDBReturnsFalse() {
                $store = $this->newWatchedItemStore(
                        $this->getMockLBFactory( $this->getMockDb() ),
+                       $this->getMockJobQueueGroup(),
                        $this->getMockCache(),
                        $this->getMockReadOnlyMode( true )
                );
@@ -1122,6 +1167,7 @@ class WatchedItemStoreUnitTest extends MediaWikiTestCase {
 
                $store = $this->newWatchedItemStore(
                        $this->getMockLBFactory( $mockDb ),
+                       $this->getMockJobQueueGroup(),
                        $mockCache,
                        $this->getMockReadOnlyMode()
                );
@@ -1147,6 +1193,7 @@ class WatchedItemStoreUnitTest extends MediaWikiTestCase {
 
                $store = $this->newWatchedItemStore(
                        $this->getMockLBFactory( $mockDb ),
+                       $this->getMockJobQueueGroup(),
                        $mockCache,
                        $this->getMockReadOnlyMode()
                );
@@ -1171,6 +1218,7 @@ class WatchedItemStoreUnitTest extends MediaWikiTestCase {
 
                $store = $this->newWatchedItemStore(
                        $this->getMockLBFactory( $mockDb ),
+                       $this->getMockJobQueueGroup(),
                        $mockCache,
                        $this->getMockReadOnlyMode()
                );
@@ -1206,6 +1254,7 @@ class WatchedItemStoreUnitTest extends MediaWikiTestCase {
 
                $store = $this->newWatchedItemStore(
                        $this->getMockLBFactory( $mockDb ),
+                       $this->getMockJobQueueGroup(),
                        $mockCache,
                        $this->getMockReadOnlyMode()
                );
@@ -1241,6 +1290,7 @@ class WatchedItemStoreUnitTest extends MediaWikiTestCase {
 
                $store = $this->newWatchedItemStore(
                        $this->getMockLBFactory( $mockDb ),
+                       $this->getMockJobQueueGroup(),
                        $mockCache,
                        $this->getMockReadOnlyMode()
                );
@@ -1264,6 +1314,7 @@ class WatchedItemStoreUnitTest extends MediaWikiTestCase {
 
                $store = $this->newWatchedItemStore(
                        $this->getMockLBFactory( $mockDb ),
+                       $this->getMockJobQueueGroup(),
                        $mockCache,
                        $this->getMockReadOnlyMode()
                );
@@ -1313,6 +1364,7 @@ class WatchedItemStoreUnitTest extends MediaWikiTestCase {
 
                $store = $this->newWatchedItemStore(
                        $this->getMockLBFactory( $mockDb ),
+                       $this->getMockJobQueueGroup(),
                        $mockCache,
                        $this->getMockReadOnlyMode()
                );
@@ -1364,6 +1416,7 @@ class WatchedItemStoreUnitTest extends MediaWikiTestCase {
 
                $store = $this->newWatchedItemStore(
                        $this->getMockLBFactory( $mockDb ),
+                       $this->getMockJobQueueGroup(),
                        $mockCache,
                        $this->getMockReadOnlyMode()
                );
@@ -1389,6 +1442,7 @@ class WatchedItemStoreUnitTest extends MediaWikiTestCase {
 
                $store = $this->newWatchedItemStore(
                        $this->getMockLBFactory( $mockDb ),
+                       $this->getMockJobQueueGroup(),
                        $mockCache,
                        $this->getMockReadOnlyMode()
                );
@@ -1434,6 +1488,7 @@ class WatchedItemStoreUnitTest extends MediaWikiTestCase {
 
                $store = $this->newWatchedItemStore(
                        $this->getMockLBFactory( $mockDb ),
+                       $this->getMockJobQueueGroup(),
                        $mockCache,
                        $this->getMockReadOnlyMode()
                );
@@ -1469,6 +1524,7 @@ class WatchedItemStoreUnitTest extends MediaWikiTestCase {
 
                $store = $this->newWatchedItemStore(
                        $this->getMockLBFactory( $mockDb ),
+                       $this->getMockJobQueueGroup(),
                        $mockCache,
                        $this->getMockReadOnlyMode()
                );
@@ -1507,6 +1563,7 @@ class WatchedItemStoreUnitTest extends MediaWikiTestCase {
 
                $store = $this->newWatchedItemStore(
                        $this->getMockLBFactory( $mockDb ),
+                       $this->getMockJobQueueGroup(),
                        $mockCache,
                        $this->getMockReadOnlyMode()
                );
@@ -1531,6 +1588,7 @@ class WatchedItemStoreUnitTest extends MediaWikiTestCase {
 
                $store = $this->newWatchedItemStore(
                        $this->getMockLBFactory( $mockDb ),
+                       $this->getMockJobQueueGroup(),
                        $mockCache,
                        $this->getMockReadOnlyMode()
                );
@@ -1572,6 +1630,7 @@ class WatchedItemStoreUnitTest extends MediaWikiTestCase {
 
                $store = $this->newWatchedItemStore(
                        $this->getMockLBFactory( $mockDb ),
+                       $this->getMockJobQueueGroup(),
                        $mockCache,
                        $this->getMockReadOnlyMode()
                );
@@ -1623,6 +1682,7 @@ class WatchedItemStoreUnitTest extends MediaWikiTestCase {
 
                $store = $this->newWatchedItemStore(
                        $mockLoadBalancer,
+                       $this->getMockJobQueueGroup(),
                        $mockCache,
                        $this->getMockReadOnlyMode()
                );
@@ -1637,6 +1697,7 @@ class WatchedItemStoreUnitTest extends MediaWikiTestCase {
        public function testGetWatchedItemsForUser_badSortOptionThrowsException() {
                $store = $this->newWatchedItemStore(
                        $this->getMockLBFactory( $this->getMockDb() ),
+                       $this->getMockJobQueueGroup(),
                        $this->getMockCache(),
                        $this->getMockReadOnlyMode()
                );
@@ -1679,6 +1740,7 @@ class WatchedItemStoreUnitTest extends MediaWikiTestCase {
 
                $store = $this->newWatchedItemStore(
                        $this->getMockLBFactory( $mockDb ),
+                       $this->getMockJobQueueGroup(),
                        $mockCache,
                        $this->getMockReadOnlyMode()
                );
@@ -1716,6 +1778,7 @@ class WatchedItemStoreUnitTest extends MediaWikiTestCase {
 
                $store = $this->newWatchedItemStore(
                        $this->getMockLBFactory( $mockDb ),
+                       $this->getMockJobQueueGroup(),
                        $mockCache,
                        $this->getMockReadOnlyMode()
                );
@@ -1740,6 +1803,7 @@ class WatchedItemStoreUnitTest extends MediaWikiTestCase {
 
                $store = $this->newWatchedItemStore(
                        $this->getMockLBFactory( $mockDb ),
+                       $this->getMockJobQueueGroup(),
                        $mockCache,
                        $this->getMockReadOnlyMode()
                );
@@ -1808,6 +1872,7 @@ class WatchedItemStoreUnitTest extends MediaWikiTestCase {
 
                $store = $this->newWatchedItemStore(
                        $this->getMockLBFactory( $mockDb ),
+                       $this->getMockJobQueueGroup(),
                        $mockCache,
                        $this->getMockReadOnlyMode()
                );
@@ -1859,6 +1924,7 @@ class WatchedItemStoreUnitTest extends MediaWikiTestCase {
 
                $store = $this->newWatchedItemStore(
                        $this->getMockLBFactory( $mockDb ),
+                       $this->getMockJobQueueGroup(),
                        $mockCache,
                        $this->getMockReadOnlyMode()
                );
@@ -1921,6 +1987,7 @@ class WatchedItemStoreUnitTest extends MediaWikiTestCase {
 
                $store = $this->newWatchedItemStore(
                        $this->getMockLBFactory( $mockDb ),
+                       $this->getMockJobQueueGroup(),
                        $mockCache,
                        $this->getMockReadOnlyMode()
                );
@@ -1962,6 +2029,7 @@ class WatchedItemStoreUnitTest extends MediaWikiTestCase {
 
                $store = $this->newWatchedItemStore(
                        $this->getMockLBFactory( $mockDb ),
+                       $this->getMockJobQueueGroup(),
                        $mockCache,
                        $this->getMockReadOnlyMode()
                );
@@ -1989,6 +2057,7 @@ class WatchedItemStoreUnitTest extends MediaWikiTestCase {
 
                $store = $this->newWatchedItemStore(
                        $this->getMockLBFactory( $mockDb ),
+                       $this->getMockJobQueueGroup(),
                        $mockCache,
                        $this->getMockReadOnlyMode()
                );
@@ -2014,6 +2083,7 @@ class WatchedItemStoreUnitTest extends MediaWikiTestCase {
 
                $store = $this->newWatchedItemStore(
                        $this->getMockLBFactory( $mockDb ),
+                       $this->getMockJobQueueGroup(),
                        $mockCache,
                        $this->getMockReadOnlyMode()
                );
@@ -2048,6 +2118,7 @@ class WatchedItemStoreUnitTest extends MediaWikiTestCase {
 
                $store = $this->newWatchedItemStore(
                        $this->getMockLBFactory( $mockDb ),
+                       $this->getMockJobQueueGroup(),
                        $mockCache,
                        $this->getMockReadOnlyMode()
                );
@@ -2092,29 +2163,26 @@ class WatchedItemStoreUnitTest extends MediaWikiTestCase {
                        ->method( 'delete' )
                        ->with( '0:SomeDbKey:1' );
 
+               $mockQueueGroup = $this->getMockJobQueueGroup();
+               $mockQueueGroup->expects( $this->once() )
+                       ->method( 'lazyPush' )
+                       ->willReturnCallback( function ( ActivityUpdateJob $job ) {
+                               // don't run
+                       } );
+
                $store = $this->newWatchedItemStore(
                        $this->getMockLBFactory( $mockDb ),
+                       $mockQueueGroup,
                        $mockCache,
                        $this->getMockReadOnlyMode()
                );
 
-               // Note: This does not actually assert the job is correct
-               $callableCallCounter = 0;
-               $mockCallback = function ( $callable ) use ( &$callableCallCounter ) {
-                       $callableCallCounter++;
-                       $this->assertInternalType( 'callable', $callable );
-               };
-               $scopedOverride = $store->overrideDeferredUpdatesAddCallableUpdateCallback( $mockCallback );
-
                $this->assertTrue(
                        $store->resetNotificationTimestamp(
                                $user,
                                $title
                        )
                );
-               $this->assertEquals( 1, $callableCallCounter );
-
-               ScopedCallback::consume( $scopedOverride );
        }
 
        public function testResetNotificationTimestamp_noItemForced() {
@@ -2132,19 +2200,19 @@ class WatchedItemStoreUnitTest extends MediaWikiTestCase {
                        ->method( 'delete' )
                        ->with( '0:SomeDbKey:1' );
 
+               $mockQueueGroup = $this->getMockJobQueueGroup();
                $store = $this->newWatchedItemStore(
                        $this->getMockLBFactory( $mockDb ),
+                       $mockQueueGroup,
                        $mockCache,
                        $this->getMockReadOnlyMode()
                );
 
-               // Note: This does not actually assert the job is correct
-               $callableCallCounter = 0;
-               $mockCallback = function ( $callable ) use ( &$callableCallCounter ) {
-                       $callableCallCounter++;
-                       $this->assertInternalType( 'callable', $callable );
-               };
-               $scopedOverride = $store->overrideDeferredUpdatesAddCallableUpdateCallback( $mockCallback );
+               $mockQueueGroup->expects( $this->any() )
+                       ->method( 'lazyPush' )
+                       ->will( $this->returnCallback( function ( ActivityUpdateJob $job ) {
+                               // don't run
+                       } ) );
 
                $this->assertTrue(
                        $store->resetNotificationTimestamp(
@@ -2153,9 +2221,6 @@ class WatchedItemStoreUnitTest extends MediaWikiTestCase {
                                'force'
                        )
                );
-               $this->assertEquals( 1, $callableCallCounter );
-
-               ScopedCallback::consume( $scopedOverride );
        }
 
        /**
@@ -2179,20 +2244,11 @@ class WatchedItemStoreUnitTest extends MediaWikiTestCase {
        }
 
        private function verifyCallbackJob(
-               $callback,
+               ActivityUpdateJob $job,
                LinkTarget $expectedTitle,
                $expectedUserId,
                callable $notificationTimestampCondition
        ) {
-               $this->assertInternalType( 'callable', $callback );
-
-               $callbackReflector = new ReflectionFunction( $callback );
-               $vars = $callbackReflector->getStaticVariables();
-               $this->assertArrayHasKey( 'job', $vars );
-               $this->assertInstanceOf( ActivityUpdateJob::class, $vars['job'] );
-
-               /** @var ActivityUpdateJob $job */
-               $job = $vars['job'];
                $this->assertEquals( $expectedTitle->getDBkey(), $job->getTitle()->getDBkey() );
                $this->assertEquals( $expectedTitle->getNamespace(), $job->getTitle()->getNamespace() );
 
@@ -2225,26 +2281,28 @@ class WatchedItemStoreUnitTest extends MediaWikiTestCase {
                        ->method( 'delete' )
                        ->with( '0:SomeTitle:1' );
 
+               $mockQueueGroup = $this->getMockJobQueueGroup();
                $store = $this->newWatchedItemStore(
                        $this->getMockLBFactory( $mockDb ),
+                       $mockQueueGroup,
                        $mockCache,
                        $this->getMockReadOnlyMode()
                );
 
-               $callableCallCounter = 0;
-               $scopedOverride = $store->overrideDeferredUpdatesAddCallableUpdateCallback(
-                       function ( $callable ) use ( &$callableCallCounter, $title, $user ) {
-                               $callableCallCounter++;
-                               $this->verifyCallbackJob(
-                                       $callable,
-                                       $title,
-                                       $user->getId(),
-                                       function ( $time ) {
-                                               return $time === null;
-                                       }
-                               );
-                       }
-               );
+               $mockQueueGroup->expects( $this->any() )
+                       ->method( 'lazyPush' )
+                       ->will( $this->returnCallback(
+                               function ( ActivityUpdateJob $job ) use ( $title, $user ) {
+                                       $this->verifyCallbackJob(
+                                               $job,
+                                               $title,
+                                               $user->getId(),
+                                               function ( $time ) {
+                                                       return $time === null;
+                                               }
+                                       );
+                               }
+                       ) );
 
                $this->assertTrue(
                        $store->resetNotificationTimestamp(
@@ -2254,9 +2312,6 @@ class WatchedItemStoreUnitTest extends MediaWikiTestCase {
                                $oldid
                        )
                );
-               $this->assertEquals( 1, $callableCallCounter );
-
-               ScopedCallback::consume( $scopedOverride );
        }
 
        public function testResetNotificationTimestamp_oldidSpecifiedNotLatestRevisionForced() {
@@ -2293,26 +2348,28 @@ class WatchedItemStoreUnitTest extends MediaWikiTestCase {
                        ->method( 'delete' )
                        ->with( '0:SomeDbKey:1' );
 
+               $mockQueueGroup = $this->getMockJobQueueGroup();
                $store = $this->newWatchedItemStore(
                        $this->getMockLBFactory( $mockDb ),
+                       $mockQueueGroup,
                        $mockCache,
                        $this->getMockReadOnlyMode()
                );
 
-               $addUpdateCallCounter = 0;
-               $scopedOverrideDeferred = $store->overrideDeferredUpdatesAddCallableUpdateCallback(
-                       function ( $callable ) use ( &$addUpdateCallCounter, $title, $user ) {
-                               $addUpdateCallCounter++;
-                               $this->verifyCallbackJob(
-                                       $callable,
-                                       $title,
-                                       $user->getId(),
-                                       function ( $time ) {
-                                               return $time !== null && $time > '20151212010101';
-                                       }
-                               );
-                       }
-               );
+               $mockQueueGroup->expects( $this->any() )
+                       ->method( 'lazyPush' )
+                       ->will( $this->returnCallback(
+                               function ( ActivityUpdateJob $job ) use ( $title, $user ) {
+                                       $this->verifyCallbackJob(
+                                               $job,
+                                               $title,
+                                               $user->getId(),
+                                               function ( $time ) {
+                                                       return $time !== null && $time > '20151212010101';
+                                               }
+                                       );
+                               }
+                       ) );
 
                $getTimestampCallCounter = 0;
                $scopedOverrideRevision = $store->overrideRevisionGetTimestampFromIdCallback(
@@ -2331,10 +2388,8 @@ class WatchedItemStoreUnitTest extends MediaWikiTestCase {
                                $oldid
                        )
                );
-               $this->assertEquals( 1, $addUpdateCallCounter );
                $this->assertEquals( 1, $getTimestampCallCounter );
 
-               ScopedCallback::consume( $scopedOverrideDeferred );
                ScopedCallback::consume( $scopedOverrideRevision );
        }
 
@@ -2368,26 +2423,28 @@ class WatchedItemStoreUnitTest extends MediaWikiTestCase {
                        ->method( 'delete' )
                        ->with( '0:SomeDbKey:1' );
 
+               $mockQueueGroup = $this->getMockJobQueueGroup();
                $store = $this->newWatchedItemStore(
                        $this->getMockLBFactory( $mockDb ),
+                       $mockQueueGroup,
                        $mockCache,
                        $this->getMockReadOnlyMode()
                );
 
-               $callableCallCounter = 0;
-               $scopedOverride = $store->overrideDeferredUpdatesAddCallableUpdateCallback(
-                       function ( $callable ) use ( &$callableCallCounter, $title, $user ) {
-                               $callableCallCounter++;
-                               $this->verifyCallbackJob(
-                                       $callable,
-                                       $title,
-                                       $user->getId(),
-                                       function ( $time ) {
-                                               return $time === null;
-                                       }
-                               );
-                       }
-               );
+               $mockQueueGroup->expects( $this->any() )
+                       ->method( 'lazyPush' )
+                       ->will( $this->returnCallback(
+                               function ( ActivityUpdateJob $job ) use ( $title, $user ) {
+                                       $this->verifyCallbackJob(
+                                               $job,
+                                               $title,
+                                               $user->getId(),
+                                               function ( $time ) {
+                                                       return $time === null;
+                                               }
+                                       );
+                               }
+                       ) );
 
                $this->assertTrue(
                        $store->resetNotificationTimestamp(
@@ -2397,9 +2454,6 @@ class WatchedItemStoreUnitTest extends MediaWikiTestCase {
                                $oldid
                        )
                );
-               $this->assertEquals( 1, $callableCallCounter );
-
-               ScopedCallback::consume( $scopedOverride );
        }
 
        public function testResetNotificationTimestamp_futureNotificationTimestampForced() {
@@ -2436,26 +2490,28 @@ class WatchedItemStoreUnitTest extends MediaWikiTestCase {
                        ->method( 'delete' )
                        ->with( '0:SomeDbKey:1' );
 
+               $mockQueueGroup = $this->getMockJobQueueGroup();
                $store = $this->newWatchedItemStore(
                        $this->getMockLBFactory( $mockDb ),
+                       $mockQueueGroup,
                        $mockCache,
                        $this->getMockReadOnlyMode()
                );
 
-               $addUpdateCallCounter = 0;
-               $scopedOverrideDeferred = $store->overrideDeferredUpdatesAddCallableUpdateCallback(
-                       function ( $callable ) use ( &$addUpdateCallCounter, $title, $user ) {
-                               $addUpdateCallCounter++;
-                               $this->verifyCallbackJob(
-                                       $callable,
-                                       $title,
-                                       $user->getId(),
-                                       function ( $time ) {
-                                               return $time === '30151212010101';
-                                       }
-                               );
-                       }
-               );
+               $mockQueueGroup->expects( $this->any() )
+                       ->method( 'lazyPush' )
+                       ->will( $this->returnCallback(
+                               function ( ActivityUpdateJob $job ) use ( $title, $user ) {
+                                       $this->verifyCallbackJob(
+                                               $job,
+                                               $title,
+                                               $user->getId(),
+                                               function ( $time ) {
+                                                       return $time === '30151212010101';
+                                               }
+                                       );
+                               }
+                       ) );
 
                $getTimestampCallCounter = 0;
                $scopedOverrideRevision = $store->overrideRevisionGetTimestampFromIdCallback(
@@ -2474,10 +2530,8 @@ class WatchedItemStoreUnitTest extends MediaWikiTestCase {
                                $oldid
                        )
                );
-               $this->assertEquals( 1, $addUpdateCallCounter );
                $this->assertEquals( 1, $getTimestampCallCounter );
 
-               ScopedCallback::consume( $scopedOverrideDeferred );
                ScopedCallback::consume( $scopedOverrideRevision );
        }
 
@@ -2515,26 +2569,28 @@ class WatchedItemStoreUnitTest extends MediaWikiTestCase {
                        ->method( 'delete' )
                        ->with( '0:SomeDbKey:1' );
 
+               $mockQueueGroup = $this->getMockJobQueueGroup();
                $store = $this->newWatchedItemStore(
                        $this->getMockLBFactory( $mockDb ),
+                       $mockQueueGroup,
                        $mockCache,
                        $this->getMockReadOnlyMode()
                );
 
-               $addUpdateCallCounter = 0;
-               $scopedOverrideDeferred = $store->overrideDeferredUpdatesAddCallableUpdateCallback(
-                       function ( $callable ) use ( &$addUpdateCallCounter, $title, $user ) {
-                               $addUpdateCallCounter++;
-                               $this->verifyCallbackJob(
-                                       $callable,
-                                       $title,
-                                       $user->getId(),
-                                       function ( $time ) {
-                                               return $time === false;
-                                       }
-                               );
-                       }
-               );
+               $mockQueueGroup->expects( $this->any() )
+                       ->method( 'lazyPush' )
+                       ->will( $this->returnCallback(
+                               function ( ActivityUpdateJob $job ) use ( $title, $user ) {
+                                       $this->verifyCallbackJob(
+                                               $job,
+                                               $title,
+                                               $user->getId(),
+                                               function ( $time ) {
+                                                       return $time === false;
+                                               }
+                                       );
+                               }
+                       ) );
 
                $getTimestampCallCounter = 0;
                $scopedOverrideRevision = $store->overrideRevisionGetTimestampFromIdCallback(
@@ -2553,16 +2609,15 @@ class WatchedItemStoreUnitTest extends MediaWikiTestCase {
                                $oldid
                        )
                );
-               $this->assertEquals( 1, $addUpdateCallCounter );
                $this->assertEquals( 1, $getTimestampCallCounter );
 
-               ScopedCallback::consume( $scopedOverrideDeferred );
                ScopedCallback::consume( $scopedOverrideRevision );
        }
 
        public function testSetNotificationTimestampsForUser_anonUser() {
                $store = $this->newWatchedItemStore(
                        $this->getMockLBFactory( $this->getMockDb() ),
+                       $this->getMockJobQueueGroup(),
                        $this->getMockCache(),
                        $this->getMockReadOnlyMode()
                );
@@ -2590,6 +2645,7 @@ class WatchedItemStoreUnitTest extends MediaWikiTestCase {
 
                $store = $this->newWatchedItemStore(
                        $this->getMockLBFactory( $mockDb ),
+                       $this->getMockJobQueueGroup(),
                        $this->getMockCache(),
                        $this->getMockReadOnlyMode()
                );
@@ -2620,6 +2676,7 @@ class WatchedItemStoreUnitTest extends MediaWikiTestCase {
 
                $store = $this->newWatchedItemStore(
                        $this->getMockLBFactory( $mockDb ),
+                       $this->getMockJobQueueGroup(),
                        $this->getMockCache(),
                        $this->getMockReadOnlyMode()
                );
@@ -2659,6 +2716,7 @@ class WatchedItemStoreUnitTest extends MediaWikiTestCase {
 
                $store = $this->newWatchedItemStore(
                        $this->getMockLBFactory( $mockDb ),
+                       $this->getMockJobQueueGroup(),
                        $this->getMockCache(),
                        $this->getMockReadOnlyMode()
                );
@@ -2702,6 +2760,7 @@ class WatchedItemStoreUnitTest extends MediaWikiTestCase {
 
                $store = $this->newWatchedItemStore(
                        $this->getMockLBFactory( $mockDb ),
+                       $this->getMockJobQueueGroup(),
                        $mockCache,
                        $this->getMockReadOnlyMode()
                );
@@ -2743,6 +2802,7 @@ class WatchedItemStoreUnitTest extends MediaWikiTestCase {
 
                $store = $this->newWatchedItemStore(
                        $this->getMockLBFactory( $mockDb ),
+                       $this->getMockJobQueueGroup(),
                        $mockCache,
                        $this->getMockReadOnlyMode()
                );
@@ -2787,6 +2847,7 @@ class WatchedItemStoreUnitTest extends MediaWikiTestCase {
 
                $store = $this->newWatchedItemStore(
                        $this->getMockLBFactory( $mockDb ),
+                       $this->getMockJobQueueGroup(),
                        $mockCache,
                        $this->getMockReadOnlyMode()
                );