3 * This program is free software; you can redistribute it and/or modify
4 * it under the terms of the GNU General Public License as published by
5 * the Free Software Foundation; either version 2 of the License, or
6 * (at your option) any later version.
8 * This program is distributed in the hope that it will be useful,
9 * but WITHOUT ANY WARRANTY; without even the implied warranty of
10 * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
11 * GNU General Public License for more details.
13 * You should have received a copy of the GNU General Public License along
14 * with this program; if not, write to the Free Software Foundation, Inc.,
15 * 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301, USA.
16 * http://www.gnu.org/copyleft/gpl.html
22 use Psr\Log\LoggerAwareInterface
;
23 use Psr\Log\LoggerInterface
;
24 use Psr\Log\NullLogger
;
25 use Wikimedia\ScopedCallback
;
28 * Class for scanning through chronological, log-structured data or change logs
29 * and locally purging cache keys related to entities that appear in this data.
31 * This is useful for repairing cache when purges are missed by using a reliable
32 * stream, such as Kafka or a replicated MySQL table. Purge loss between datacenters
33 * is expected to be more common than within them.
37 class WANObjectCacheReaper
implements LoggerAwareInterface
{
38 /** @var WANObjectCache */
43 protected $logChunkCallback;
45 protected $keyListCallback;
46 /** @var LoggerInterface */
52 protected $initialStartWindow;
55 * @param WANObjectCache $cache Cache to reap bad keys from
56 * @param BagOStuff $store Cache to store positions use for locking
57 * @param callable $logCallback Callback taking arguments:
58 * - The starting position as a UNIX timestamp
59 * - The starting unique ID used for breaking timestamp collisions or null
60 * - The ending position as a UNIX timestamp
61 * - The maximum number of results to return
62 * It returns a list of maps of (key: cache key, pos: UNIX timestamp, id: unique ID)
63 * for each key affected, with the corrosponding event timestamp/ID information.
64 * The events should be in ascending order, by (timestamp,id).
65 * @param callable $keyCallback Callback taking arguments:
66 * - The WANObjectCache instance
67 * - An object from the event log
68 * It should return a list of WAN cache keys.
69 * The callback must fully duck-type test the object, since can be any model class.
70 * @param array $params Additional options:
71 * - channel: the name of the update event stream.
72 * Default: WANObjectCache::DEFAULT_PURGE_CHANNEL.
73 * - initialStartWindow: seconds back in time to start if the position is lost.
75 * - logger: an SPL monolog instance [optional]
77 public function __construct(
78 WANObjectCache
$cache,
80 callable
$logCallback,
81 callable
$keyCallback,
84 $this->cache
= $cache;
85 $this->store
= $store;
87 $this->logChunkCallback
= $logCallback;
88 $this->keyListCallback
= $keyCallback;
89 if ( isset( $params['channel'] ) ) {
90 $this->channel
= $params['channel'];
92 throw new UnexpectedValueException( "No channel specified." );
95 $this->initialStartWindow
= $params['initialStartWindow'] ??
3600;
96 $this->logger
= $params['logger'] ??
new NullLogger();
99 public function setLogger( LoggerInterface
$logger ) {
100 $this->logger
= $logger;
104 * Check and reap stale keys based on a chunk of events
106 * @param int $n Number of events
107 * @return int Number of keys checked
109 final public function invoke( $n = 100 ) {
110 $posKey = $this->store
->makeGlobalKey( 'WANCache', 'reaper', $this->channel
);
111 $scopeLock = $this->store
->getScopedLock( "$posKey:busy", 0 );
117 $status = $this->store
->get( $posKey );
119 $status = [ 'pos' => $now - $this->initialStartWindow
, 'id' => null ];
122 // Get events for entities who's keys tombstones/hold-off should have expired by now
123 $events = call_user_func_array(
124 $this->logChunkCallback
,
125 [ $status['pos'], $status['id'], $now - WANObjectCache
::HOLDOFF_TTL
- 1, $n ]
130 foreach ( $events as $event ) {
131 $keys = call_user_func_array(
132 $this->keyListCallback
,
133 [ $this->cache
, $event['item'] ]
135 foreach ( $keys as $key ) {
136 unset( $keyEvents[$key] ); // use only the latest per key
138 'pos' => $event['pos'],
146 foreach ( $keyEvents as $key => $keyEvent ) {
147 if ( !$this->cache
->reap( $key, $keyEvent['pos'] ) ) {
151 $lastOkEvent = $event;
154 if ( $lastOkEvent ) {
155 $ok = $this->store
->merge(
157 function ( $bag, $key, $curValue ) use ( $lastOkEvent ) {
161 $curCoord = [ $curValue['pos'], $curValue['id'] ];
162 $newCoord = [ $lastOkEvent['pos'], $lastOkEvent['id'] ];
163 if ( $newCoord < $curCoord ) {
164 // Keep prior position instead of rolling it back
170 'pos' => $lastOkEvent['pos'],
171 'id' => $lastOkEvent['id'],
172 'ctime' => $curValue ?
$curValue['ctime'] : date( 'c' )
175 IExpiringStore
::TTL_INDEFINITE
178 $pos = $lastOkEvent['pos'];
179 $id = $lastOkEvent['id'];
181 $this->logger
->info( "Updated cache reap position ($pos, $id)." );
183 $this->logger
->error( "Could not update cache reap position ($pos, $id)." );
187 ScopedCallback
::consume( $scopeLock );
193 * @return array|bool Returns (pos, id) map or false if not set
195 public function getState() {
196 $posKey = $this->store
->makeGlobalKey( 'WANCache', 'reaper', $this->channel
);
198 return $this->store
->get( $posKey );