jetpack / jetpack_vendor / automattic / jetpack-sync / src / class-queue.php

class-queue.php in Jetpack – WP Security, Backup, Speed, & Growth 16.3-a.1, at jetpack_vendor/automattic/jetpack-sync/src/class-queue.php

726 lines 17.1 KB

Raw Download Zip

1	<?php
2	/**
3	* The class that describes the Queue for the sync package.
4	*
5	* @package automattic/jetpack-sync
6	*/
7
8	namespace Automattic\Jetpack\Sync;
9
10	use Automattic\Jetpack\Sync\Queue\Queue_Storage_Options;
11	use Automattic\Jetpack\Sync\Queue\Queue_Storage_Table;
12	use WP_Error;
13
14	/**
15	* A persistent queue that can be flushed in increments of N items,
16	* and which blocks reads until checked-out buffers are checked in or
17	* closed. This uses raw SQL for two reasons: speed, and not triggering
18	* tons of added_option callbacks.
19	*/
20	class Queue {
21	/**
22	* The queue id.
23	*
24	* @var string
25	*/
26	public $id;
27
28	/**
29	* Keeps track of the rows.
30	*
31	* @var int
32	*/
33	private $row_iterator;
34
35	/**
36	* Random number.
37	*
38	* @var int
39	*/
40	public $random_int;
41
42	/**
43	* Queue Storage instance where we'll store the queue items.
44	*
45	* For now, it's only the Options table. To be updated to include the Custom table in future updates.
46	*
47	* @var Queue_Storage_Options\|Queue_Storage_Table\|null
48	*/
49	public $queue_storage = null;
50
51	/**
52	* Queue constructor.
53	*
54	* @param string $id Name of the queue.
55	*/
56	public function __construct( $id ) {
57	$this->id = str_replace( '-', '_', $id ); // Necessary to ensure we don't have ID collisions in the SQL.
58	$this->row_iterator = 0;
59	$this->random_int = wp_rand( 1, 1000000 );
60
61	/**
62	* If the Custom queue table is enabled - let's use it as a backend. Otherwise, fall back to the Options table.
63	*/
64	if ( Settings::is_custom_queue_table_enabled() ) {
65	$this->queue_storage = new Queue_Storage_Table( $this->id );
66	} else {
67	// Initialize the storage with the Options table backend. To be changed in subsequent updates to include the logic to switch to Custom Table.
68	$this->queue_storage = new Queue_Storage_Options( $this->id );
69	}
70	}
71
72	/**
73	* Add a single item to the queue.
74	*
75	* @param object $item Event object to add to queue.
76	*
77	* @return bool\|WP_Error
78	*/
79	public function add( $item ) {
80	$added = false;
81
82	// If empty, don't add.
83	if ( empty( $item ) ) {
84	return false;
85	}
86
87	// Attempt to serialize data, if an exception (closures) return early.
88	try {
89	$item = serialize( $item ); // phpcs:ignore WordPress.PHP.DiscouragedPHPFunctions.serialize_serialize
90	} catch ( \Exception $ex ) {
91	return new WP_Error( 'queue_unable_to_serialize', 'Unable to serialize item' );
92	}
93
94	// This basically tries to add the option until enough time has elapsed that
95	// it has a unique (microtime-based) option key.
96	while ( ! $added ) {
97	$added = $this->queue_storage->insert_item( $this->get_next_data_row_option_name(), $item );
98	}
99
100	return $added;
101	}
102
103	/**
104	* Insert all the items in a single SQL query. May be subject to query size limits!
105	*
106	* @param array $items Array of events to add to the queue.
107	*
108	* @return bool\|\WP_Error
109	*/
110	public function add_all( $items ) {
111	// TODO check and figure out if it's used at all and if we can optimize it.
112	$base_option_name = $this->get_next_data_row_option_name();
113
114	$rows_added = $this->queue_storage->add_all( $items, $base_option_name );
115
116	if ( count( $items ) !== $rows_added ) {
117	return new WP_Error( 'row_count_mismatch', "The number of rows inserted didn't match the size of the input array" );
118	}
119
120	return true;
121	}
122
123	/**
124	* Get the front-most items on the queue without checking them out.
125	*
126	* @param int $count Number of items to return when looking at the items.
127	*
128	* @return array
129	*/
130	public function peek( $count = 1 ) {
131	$items = $this->fetch_items( $count );
132	if ( $items ) {
133	return Utils::get_item_values( $items );
134	}
135
136	return array();
137	}
138
139	/**
140	* Get the last-added items on the queue without checking them out.
141	*
142	* @param int $count Number of items to return when looking at the items.
143	*
144	* @return array
145	*/
146	public function peek_newest( $count = 1 ) {
147	$items = $this->fetch_items( $count, 'DESC' );
148	if ( $items ) {
149	return Utils::get_item_values( $items );
150	}
151
152	return array();
153	}
154
155	/**
156	* Gets items with particular IDs.
157	*
158	* @param array $item_ids Array of item IDs to retrieve.
159	*
160	* @return array
161	*/
162	public function peek_by_id( $item_ids ) {
163	$items = $this->fetch_items_by_id( $item_ids );
164	if ( $items ) {
165	return Utils::get_item_values( $items );
166	}
167
168	return array();
169	}
170
171	/**
172	* Gets the queue lag.
173	* Lag is the difference in time between the age of the oldest item
174	* (aka first or frontmost item) and the current time.
175	*
176	* @param float $now The current time in microtime.
177	*
178	* @return float
179	*/
180	public function lag( $now = null ) {
181	return (float) $this->queue_storage->get_lag( $now );
182	}
183
184	/**
185	* Resets the queue.
186	*/
187	public function reset() {
188	$this->delete_checkout_id();
189
190	$this->queue_storage->clear_queue();
191	}
192
193	/**
194	* Return the size of the queue.
195	*
196	* @return int
197	*/
198	public function size() {
199	return $this->queue_storage->get_item_count();
200	}
201
202	/**
203	* Lets you know if there is any items in the queue.
204	*
205	* We use this peculiar implementation because it's much faster than count(*).
206	*
207	* @return bool
208	*/
209	public function has_any_items() {
210	return $this->size() > 0;
211	}
212
213	/**
214	* Used to checkout the queue.
215	*
216	* @param int $buffer_size Size of the buffer to checkout.
217	*
218	* @return \Automattic\Jetpack\Sync\Queue_Buffer\|bool\|int\|\WP_Error
219	*/
220	public function checkout( $buffer_size ) {
221	if ( $this->get_checkout_id() ) {
222	return new WP_Error( 'unclosed_buffer', 'There is an unclosed buffer' );
223	}
224
225	// TODO check if adding a prefix is going to be a problem
226	$buffer_id = uniqid( '', true );
227
228	$result = $this->set_checkout_id( $buffer_id );
229
230	if ( ! $result \|\| is_wp_error( $result ) ) {
231	return $result;
232	}
233
234	$items = $this->fetch_items( $buffer_size );
235
236	if ( ! is_countable( $items ) ) {
237	return false;
238	}
239
240	if ( count( $items ) === 0 ) {
241	return false;
242	}
243
244	return new Queue_Buffer( $buffer_id, array_slice( $items, 0, $buffer_size ) );
245	}
246
247	/**
248	* Given a list of items return the items ids.
249	*
250	* @param array $items List of item objects.
251	*
252	* @return array Ids of the items.
253	*/
254	public function get_ids( $items ) {
255	return array_map(
256	function ( $item ) {
257	return $item->id;
258	},
259	$items
260	);
261	}
262
263	/**
264	* Remove the oldest items from the queue.
265	*
266	* @param int $limit Number of items to remove from the queue.
267	*
268	* @return array\|object\|null
269	*/
270	public function pop( $limit ) {
271	$items = $this->fetch_items( $limit );
272
273	if ( ! $items ) {
274	return array();
275	}
276
277	$ids = $this->get_ids( $items );
278
279	$this->delete( $ids );
280
281	return $items;
282	}
283
284	/**
285	* Remove the newest items from the queue.
286	*
287	* @param int $limit Number of items to remove from the queue.
288	*
289	* @return array\|object\|null
290	*/
291	public function pop_newest( $limit ) {
292	$items = $this->fetch_items( $limit, 'DESC' );
293
294	if ( ! $items ) {
295	return array();
296	}
297
298	$ids = $this->get_ids( $items );
299
300	$this->delete( $ids );
301
302	return $items;
303	}
304
305	/**
306	* Get the items from the queue with a memory limit.
307	*
308	* This checks out rows until it either empties the queue or hits a certain memory limit
309	* it loads the sizes from the DB first so that it doesn't accidentally
310	* load more data into memory than it needs to.
311	* The only way it will load more items than $max_size is if a single queue item
312	* exceeds the memory limit, but in that case it will send that item by itself.
313	*
314	* @param int $max_memory (bytes) Maximum memory threshold.
315	* @param int $max_buffer_size Maximum buffer size (number of items).
316	*
317	* @return \Automattic\Jetpack\Sync\Queue_Buffer\|bool\|int\|\WP_Error
318	*/
319	public function checkout_with_memory_limit( $max_memory, $max_buffer_size = 500 ) {
320	if ( $this->get_checkout_id() ) {
321	return new WP_Error( 'unclosed_buffer', 'There is an unclosed buffer' );
322	}
323
324	$buffer_id = uniqid( '', true );
325
326	$result = $this->set_checkout_id( $buffer_id );
327
328	if ( ! $result \|\| is_wp_error( $result ) ) {
329	return $result;
330	}
331
332	// How much memory is currently being used by the items.
333	$total_memory = 0;
334
335	// Store the items to return
336	$items = array();
337
338	$current_items_ids = $this->queue_storage->get_items_ids_with_size( $max_buffer_size - count( $items ) );
339
340	// If no valid items are returned or no items are returned, continue.
341	if ( ! is_countable( $current_items_ids ) \|\| count( $current_items_ids ) === 0 ) {
342	return false;
343	}
344
345	$item_ids_to_fetch = array();
346
347	foreach ( $current_items_ids as $id => $item_with_size ) {
348	$total_memory += $item_with_size->value_size;
349
350	// If this is the first item and it exceeds memory, allow loop to continue
351	// we will exit on the next iteration instead.
352	if ( $total_memory > $max_memory && $id > 0 ) {
353	break;
354	}
355
356	$item_ids_to_fetch[] = $item_with_size->id;
357	}
358
359	$current_items = $this->queue_storage->fetch_items_by_ids( $item_ids_to_fetch );
360
361	$items_count = is_countable( $current_items ) ? count( $current_items ) : 0;
362
363	if ( $items_count > 0 ) {
364	/**
365	* Save some memory by moving things one by one to the array of items being returned, instead of
366	* unserializing all and then merging them with other items.
367	*
368	* PHPCS ignore is because this is the expected behavior - we're assigning a variable in the condition part of the loop.
369	*/
370	// phpcs:ignore Generic.CodeAnalysis.AssignmentInCondition.FoundInWhileCondition
371	while ( ( $current_item = array_shift( $current_items ) ) !== null ) {
372	// @codingStandardsIgnoreStart
373	$current_item->value = @unserialize( $current_item->value );
374	// @codingStandardsIgnoreEnd
375	$items[] = $current_item;
376	}
377	}
378
379	if ( count( $items ) === 0 ) {
380	$this->delete_checkout_id();
381
382	return false;
383	}
384
385	return new Queue_Buffer( $buffer_id, $items );
386	}
387
388	/**
389	* Check in the queue.
390	*
391	* @param \Automattic\Jetpack\Sync\Queue_Buffer $buffer Queue_Buffer object.
392	*
393	* @return bool\|\WP_Error
394	*/
395	public function checkin( $buffer ) {
396	$is_valid = $this->validate_checkout( $buffer );
397
398	if ( is_wp_error( $is_valid ) ) {
399	return $is_valid;
400	}
401
402	$this->delete_checkout_id();
403
404	return true;
405	}
406
407	/**
408	* Close the buffer.
409	*
410	* @param \Automattic\Jetpack\Sync\Queue_Buffer $buffer Queue_Buffer object.
411	* @param null\|array $ids_to_remove Ids to remove from the queue.
412	*
413	* @return bool\|\WP_Error
414	*/
415	public function close( $buffer, $ids_to_remove = null ) {
416	$is_valid = $this->validate_checkout( $buffer );
417
418	if ( is_wp_error( $is_valid ) ) {
419	// Always delete ids_to_remove even when buffer is no longer checked-out.
420	// They were processed by WP.com so safe to remove from queue.
421	if ( $ids_to_remove !== null ) {
422	$this->delete( $ids_to_remove );
423	}
424	return $is_valid;
425	}
426
427	$this->delete_checkout_id();
428
429	// By default clear all items in the buffer.
430	if ( $ids_to_remove === null ) {
431	$ids_to_remove = $buffer->get_item_ids();
432	}
433
434	$this->delete( $ids_to_remove );
435
436	return true;
437	}
438
439	/**
440	* Delete elements from the queue.
441	*
442	* @param array $ids Ids to delete.
443	*
444	* @return bool\|int
445	*/
446	private function delete( $ids ) {
447	if ( array() === $ids ) {
448	return 0;
449	}
450
451	$this->queue_storage->delete_items_by_ids( $ids );
452
453	return true;
454	}
455
456	/**
457	* Flushes all items from the queue.
458	*
459	* @return array
460	*/
461	public function flush_all() {
462	$items = Utils::get_item_values( $this->fetch_items() );
463	$this->reset();
464
465	return $items;
466	}
467
468	/**
469	* Get all the items from the queue.
470	*
471	* @return array\|object\|null
472	*/
473	public function get_all() {
474	return $this->fetch_items();
475	}
476
477	/**
478	* Forces Checkin of the queue.
479	* Use with caution, this could allow multiple processes to delete
480	* and send from the queue at the same time
481	*/
482	public function force_checkin() {
483	$this->delete_checkout_id();
484	}
485
486	/**
487	* Checks if the queue is locked.
488	*
489	* @return bool
490	*/
491	public function is_locked() {
492	return (bool) $this->get_checkout_id();
493	}
494
495	/**
496	* Locks checkouts from the queue
497	* tries to wait up to $timeout seconds for the queue to be empty.
498	*
499	* @param int $timeout The wait time in seconds for the queue to be empty.
500	*
501	* @return bool\|int\|\WP_Error
502	*/
503	public function lock( $timeout = 30 ) {
504	$tries = 0;
505
506	while ( $this->has_any_items() && $tries < $timeout ) {
507	sleep( 1 );
508	++$tries;
509	}
510
511	if ( 30 === $tries ) {
512	return new WP_Error( 'lock_timeout', 'Timeout waiting for sync queue to empty' );
513	}
514
515	if ( $this->get_checkout_id() ) {
516	return new WP_Error( 'unclosed_buffer', 'There is an unclosed buffer' );
517	}
518
519	// Hopefully this means we can acquire a checkout?
520	$result = $this->set_checkout_id( 'lock' );
521
522	if ( ! $result \|\| is_wp_error( $result ) ) {
523	return $result;
524	}
525
526	return true;
527	}
528
529	/**
530	* Unlocks the queue.
531	*
532	* @return bool\|int
533	*/
534	public function unlock() {
535	return $this->delete_checkout_id();
536	}
537
538	/**
539	* This option is specifically chosen to, as much as possible, preserve time order
540	* and minimise the possibility of collisions between multiple processes working
541	* at the same time.
542	*
543	* @return string
544	*/
545	protected function generate_option_name_timestamp() {
546	return sprintf( '%.6f', microtime( true ) );
547	}
548
549	/**
550	* Gets the checkout ID.
551	*
552	* @return bool\|string
553	*/
554	private function get_checkout_id() {
555	global $wpdb;
556	$checkout_value = $wpdb->get_var(
557	$wpdb->prepare(
558	"SELECT option_value FROM $wpdb->options WHERE option_name = %s",
559	$this->get_lock_option_name()
560	)
561	);
562
563	if ( $checkout_value ) {
564	$parts = explode( ':', $checkout_value, 2 );
565	if ( count( $parts ) === 2 ) {
566	list( $checkout_id, $timestamp ) = $parts;
567	if ( (int) $timestamp > time() ) {
568	return $checkout_id;
569	}
570	}
571	}
572
573	return false;
574	}
575
576	/**
577	* Sets the checkout id.
578	*
579	* @param string $checkout_id The ID of the checkout.
580	*
581	* @return bool\|int
582	*/
583	private function set_checkout_id( $checkout_id ) {
584	global $wpdb;
585
586	$expires = time() + Defaults::$default_sync_queue_lock_timeout;
587	$updated_num = $wpdb->query(
588	$wpdb->prepare(
589	"UPDATE $wpdb->options SET option_value = %s WHERE option_name = %s",
590	"$checkout_id:$expires",
591	$this->get_lock_option_name()
592	)
593	);
594
595	if ( ! $updated_num ) {
596	$updated_num = $wpdb->query(
597	$wpdb->prepare(
598	"INSERT INTO $wpdb->options ( option_name, option_value, autoload ) VALUES ( %s, %s, 'no' )",
599	$this->get_lock_option_name(),
600	"$checkout_id:$expires"
601	)
602	);
603	}
604
605	return $updated_num;
606	}
607
608	/**
609	* Deletes the checkout ID.
610	*
611	* @return bool\|int
612	*/
613	private function delete_checkout_id() {
614	global $wpdb;
615	// Rather than delete, which causes fragmentation, we update in place.
616	return $wpdb->query(
617	$wpdb->prepare(
618	"UPDATE $wpdb->options SET option_value = %s WHERE option_name = %s",
619	'0:0',
620	$this->get_lock_option_name()
621	)
622	);
623	}
624
625	/**
626	* Return the lock option name.
627	*
628	* @return string
629	*/
630	private function get_lock_option_name() {
631	return "jpsq_{$this->id}_checkout";
632	}
633
634	/**
635	* Return the next data row option name.
636	*
637	* @return string
638	*/
639	private function get_next_data_row_option_name() {
640	$timestamp = $this->generate_option_name_timestamp();
641
642	// Row iterator is used to avoid collisions where we're writing data waaay fast in a single process.
643	if ( PHP_INT_MAX === $this->row_iterator ) {
644	$this->row_iterator = 0;
645	} else {
646	$this->row_iterator += 1;
647	}
648
649	return 'jpsq_' . $this->id . '-' . $timestamp . '-' . $this->random_int . '-' . $this->row_iterator;
650	}
651
652	/**
653	* Return the items in the queue.
654	*
655	* @param null\|int $limit Limit to the number of items we fetch at once.
656	* @param string $order Sort direction for the items. Accepts 'ASC' or 'DESC'.
657	* Any other value will be treated as 'ASC'.
658	*
659	* @return array\|object\|null
660	*/
661	private function fetch_items( $limit = null, $order = 'ASC' ) {
662	$order = 'DESC' === $order ? 'DESC' : 'ASC';
663
664	$items = $this->queue_storage->fetch_items( $limit, $order );
665
666	return $this->unserialize_values( $items );
667	}
668
669	/**
670	* Return items with specific ids.
671	*
672	* @param array $items_ids Array of event ids.
673	*
674	* @return array\|object\|null
675	*/
676	private function fetch_items_by_id( $items_ids ) {
677	return $this->unserialize_values( $this->queue_storage->fetch_items_by_ids( $items_ids ) );
678	}
679
680	/**
681	* Unserialize item values.
682	*
683	* @param array $items Events from the Queue to be unserialized.
684	*
685	* @return mixed
686	*/
687	private function unserialize_values( $items ) {
688	array_walk(
689	$items,
690	function ( $item ) {
691	// @codingStandardsIgnoreStart
692	$item->value = @unserialize( $item->value );
693	// @codingStandardsIgnoreEnd
694	}
695	);
696
697	return $items;
698	}
699
700	/**
701	* Return true if the buffer is still valid or an Error other wise.
702	*
703	* @param \Automattic\Jetpack\Sync\Queue_Buffer $buffer The Queue_Buffer.
704	*
705	* @return bool\|WP_Error
706	*/
707	private function validate_checkout( $buffer ) {
708	if ( ! $buffer instanceof Queue_Buffer ) {
709	return new WP_Error( 'not_a_buffer', 'You must checkin an instance of Automattic\\Jetpack\\Sync\\Queue_Buffer' );
710	}
711
712	$checkout_id = $this->get_checkout_id();
713
714	if ( ! $checkout_id ) {
715	return new WP_Error( 'buffer_not_checked_out', 'There are no checked out buffers' );
716	}
717
718	// TODO: change to strict comparison.
719	if ( $checkout_id != $buffer->id ) { // phpcs:ignore Universal.Operators.StrictComparisons.LooseNotEqual
720	return new WP_Error( 'buffer_mismatch', 'The buffer you checked in was not checked out' );
721	}
722
723	return true;
724	}
725	}
726