← All changes | jetpack_vendor/automattic/jetpack-sync/src/class-sender.php +1039 -0 16.2-beta → 16.3 View file →

		@@ -1,0 +1,1039 @@
	1	+<?php
	2	+/**
	3	+ * Sync sender.
	4	+ *
	5	+ * @package automattic/jetpack-sync
	6	+ */
	7	+
	8	+namespace Automattic\Jetpack\Sync;
	9	+
	10	+use Automattic\Jetpack\Connection\Manager;
	11	+use Automattic\Jetpack\Constants;
	12	+use WP_Error;
	13	+
	14	+/**
	15	+ * This class grabs pending actions from the queue and sends them
	16	+ */
	17	+class Sender {
	18	+ /**
	19	+ * Name of the option that stores the time of the next sync.
	20	+ *
	21	+ * @access public
	22	+ *
	23	+ * @var string
	24	+ */
	25	+ const NEXT_SYNC_TIME_OPTION_NAME = 'jetpack_next_sync_time';
	26	+
	27	+ /**
	28	+ * Name of the transient responsible for temprorarily disabling Sync sending during Pulls.
	29	+ *
	30	+ * @access public
	31	+ *
	32	+ * @var string
	33	+ */
	34	+ const TEMP_SYNC_DISABLE_TRANSIENT_NAME = 'jetpack_disable_sync_sending';
	35	+
	36	+ /**
	37	+ * Expiry of the transient responsible for temprorarily disabling Sync sending during Pulls.
	38	+ *
	39	+ * @access public
	40	+ *
	41	+ * @var int
	42	+ */
	43	+ const TEMP_SYNC_DISABLE_TRANSIENT_EXPIRY = MINUTE_IN_SECONDS;
	44	+
	45	+ /**
	46	+ * Sync timeout after a WPCOM error.
	47	+ *
	48	+ * @access public
	49	+ *
	50	+ * @var int
	51	+ */
	52	+ const WPCOM_ERROR_SYNC_DELAY = 60;
	53	+
	54	+ /**
	55	+ * Sync timeout after a queue has been locked.
	56	+ *
	57	+ * @access public
	58	+ *
	59	+ * @var int
	60	+ */
	61	+ const QUEUE_LOCKED_SYNC_DELAY = 10;
	62	+
	63	+ /**
	64	+ * Maximum bytes to checkout without exceeding the memory limit.
	65	+ *
	66	+ * @access private
	67	+ *
	68	+ * @var int
	69	+ */
	70	+ private $dequeue_max_bytes;
	71	+
	72	+ /**
	73	+ * Maximum bytes in a single encoded item.
	74	+ *
	75	+ * @access private
	76	+ *
	77	+ * @var int
	78	+ */
	79	+ private $upload_max_bytes;
	80	+
	81	+ /**
	82	+ * Maximum number of sync items in a single action.
	83	+ *
	84	+ * @access private
	85	+ *
	86	+ * @var int
	87	+ */
	88	+ private $upload_max_rows;
	89	+
	90	+ /**
	91	+ * Maximum time for perfirming a checkout of items from the queue (in seconds).
	92	+ *
	93	+ * @access private
	94	+ *
	95	+ * @var int
	96	+ */
	97	+ private $max_dequeue_time;
	98	+
	99	+ /**
	100	+ * How many seconds to wait after sending sync items after exceeding the sync wait threshold (in seconds).
	101	+ *
	102	+ * @access private
	103	+ *
	104	+ * @var int
	105	+ */
	106	+ private $sync_wait_time;
	107	+
	108	+ /**
	109	+ * How much maximum time to wait for the checkout to finish (in seconds).
	110	+ *
	111	+ * @access private
	112	+ *
	113	+ * @var int
	114	+ */
	115	+ private $sync_wait_threshold;
	116	+
	117	+ /**
	118	+ * How much maximum time to wait for the sync items to be queued for sending (in seconds).
	119	+ *
	120	+ * @access private
	121	+ *
	122	+ * @var int
	123	+ */
	124	+ private $enqueue_wait_time;
	125	+
	126	+ /**
	127	+ * Incremental sync queue object.
	128	+ *
	129	+ * @access private
	130	+ *
	131	+ * @var \Automattic\Jetpack\Sync\Queue
	132	+ */
	133	+ private $sync_queue;
	134	+
	135	+ /**
	136	+ * Full sync queue object.
	137	+ *
	138	+ * @access private
	139	+ *
	140	+ * @var \Automattic\Jetpack\Sync\Queue
	141	+ */
	142	+ private $full_sync_queue;
	143	+
	144	+ /**
	145	+ * Codec object for encoding and decoding sync items.
	146	+ *
	147	+ * @access private
	148	+ *
	149	+ * @var \Automattic\Jetpack\Sync\Codec_Interface
	150	+ */
	151	+ private $codec;
	152	+
	153	+ /**
	154	+ * The current user before we change or clear it.
	155	+ *
	156	+ * @access private
	157	+ *
	158	+ * @var \WP_User
	159	+ */
	160	+ private $old_user;
	161	+
	162	+ /**
	163	+ * Container for the singleton instance of this class.
	164	+ *
	165	+ * @access private
	166	+ * @static
	167	+ *
	168	+ * @var \Automattic\Jetpack\Sync\Sender
	169	+ */
	170	+ private static $instance;
	171	+
	172	+ /**
	173	+ * Retrieve the singleton instance of this class.
	174	+ *
	175	+ * @access public
	176	+ * @static
	177	+ *
	178	+ * @return Sender
	179	+ */
	180	+ public static function get_instance() {
	181	+ if ( null === self::$instance ) {
	182	+ self::$instance = new self();
	183	+ }
	184	+
	185	+ return self::$instance;
	186	+ }
	187	+
	188	+ /**
	189	+ * Constructor.
	190	+ * This is necessary because you can't use "new" when you declare instance properties >:(
	191	+ *
	192	+ * @access protected
	193	+ * @static
	194	+ */
	195	+ protected function __construct() {
	196	+ $this->set_defaults();
	197	+ $this->init();
	198	+ }
	199	+
	200	+ /**
	201	+ * Initialize the sender.
	202	+ * Prepares the current user and initializes all sync modules.
	203	+ *
	204	+ * @access private
	205	+ */
	206	+ private function init() {
	207	+ add_action( 'jetpack_sync_before_send_queue_sync', array( $this, 'maybe_set_user_from_token' ), 1 );
	208	+ add_action( 'jetpack_sync_before_send_queue_sync', array( $this, 'maybe_clear_user_from_token' ), 20 );
	209	+ add_filter( 'jetpack_xmlrpc_unauthenticated_methods', array( $this, 'register_jetpack_xmlrpc_methods' ) );
	210	+ foreach ( Modules::get_modules() as $module ) {
	211	+ $module->init_before_send();
	212	+ }
	213	+ }
	214	+
	215	+ /**
	216	+ * Detect if this is a XMLRPC request with a valid signature.
	217	+ * If so, changes the user to the new one.
	218	+ *
	219	+ * @access public
	220	+ */
	221	+ public function maybe_set_user_from_token() {
	222	+ $connection = new Manager();
	223	+ $verified_user = $connection->verify_xml_rpc_signature();
	224	+ if ( Constants::is_true( 'XMLRPC_REQUEST' ) &&
	225	+ ! is_wp_error( $verified_user )
	226	+ && $verified_user
	227	+ ) {
	228	+ $old_user = wp_get_current_user();
	229	+ $this->old_user = $old_user->ID ?? 0;
	230	+ wp_set_current_user( $verified_user['user_id'] );
	231	+ }
	232	+ }
	233	+
	234	+ /**
	235	+ * If we used to have a previous current user, revert back to it.
	236	+ *
	237	+ * @access public
	238	+ */
	239	+ public function maybe_clear_user_from_token() {
	240	+ if ( isset( $this->old_user ) ) {
	241	+ wp_set_current_user( $this->old_user );
	242	+ }
	243	+ }
	244	+
	245	+ /**
	246	+ * Retrieve the next sync time.
	247	+ *
	248	+ * Update @since 1.43.2
	249	+ * Sometimes when we process Sync requests in Jetpack, the server clock can be a
	250	+ * bit in the future and this can lock Sync to not send stuff for a while.
	251	+ * We are introducing an extra check, to make sure to limit the next_sync_time
	252	+ * to be at most one hour in the future from the current time.
	253	+ *
	254	+ * @access public
	255	+ *
	256	+ * @param string $queue_name Name of the queue.
	257	+ * @return float Timestamp of the next sync.
	258	+ */
	259	+ public function get_next_sync_time( $queue_name ) {
	260	+ $option_name = self::NEXT_SYNC_TIME_OPTION_NAME . '_' . $queue_name;
	261	+ $next_sync_time = (float) get_option( $option_name, 0 );
	262	+
	263	+ $is_more_than_one_hour = ( $next_sync_time - microtime( true ) ) >= HOUR_IN_SECONDS;
	264	+
	265	+ if ( $is_more_than_one_hour ) {
	266	+ delete_option( $option_name );
	267	+ $next_sync_time = 0;
	268	+ }
	269	+
	270	+ return $next_sync_time;
	271	+ }
	272	+
	273	+ /**
	274	+ * Set the next sync time.
	275	+ *
	276	+ * @access public
	277	+ *
	278	+ * @param int $time Timestamp of the next sync.
	279	+ * @param string $queue_name Name of the queue.
	280	+ * @return boolean True if update was successful, false otherwise.
	281	+ */
	282	+ public function set_next_sync_time( $time, $queue_name ) {
	283	+ return update_option( self::NEXT_SYNC_TIME_OPTION_NAME . '_' . $queue_name, $time, true );
	284	+ }
	285	+
	286	+ /**
	287	+ * Trigger a full sync.
	288	+ *
	289	+ * @access public
	290	+ *
	291	+ * @return boolean|WP_Error True if this sync sending was successful, error object otherwise.
	292	+ */
	293	+ public function do_full_sync() {
	294	+ $sync_module = Modules::get_module( 'full-sync' );
	295	+ '@phan-var Modules\Full_Sync_Immediately|Modules\Full_Sync $sync_module';
	296	+ if ( ! $sync_module ) {
	297	+ return;
	298	+ }
	299	+ // Full Sync Disabled.
	300	+ if ( ! Settings::get_setting( 'full_sync_sender_enabled' ) ) {
	301	+ return;
	302	+ }
	303	+
	304	+ // Don't sync if request is marked as read only.
	305	+ if ( Constants::is_true( 'JETPACK_SYNC_READ_ONLY' ) ) {
	306	+ return new WP_Error( 'jetpack_sync_read_only' );
	307	+ }
	308	+
	309	+ // Sync not started or Sync finished.
	310	+ $status = $sync_module->get_status();
	311	+ if ( false === $status['started'] || ( ! empty( $status['started'] ) && ! empty( $status['finished'] ) ) ) {
	312	+ return false;
	313	+ }
	314	+
	315	+ $this->continue_full_sync_enqueue();
	316	+ // immediate full sync sends data in continue_full_sync_enqueue.
	317	+ if ( ! $sync_module instanceof Modules\Full_Sync_Immediately ) {
	318	+ return $this->do_sync_and_set_delays( $this->full_sync_queue );
	319	+ } else {
	320	+ $status = $sync_module->get_status();
	321	+ // Sync not started or Sync finished.
	322	+ if ( false === $status['started'] || ( ! empty( $status['started'] ) && ! empty( $status['finished'] ) ) ) {
	323	+ return false;
	324	+ } else {
	325	+ return true;
	326	+ }
	327	+ }
	328	+ }
	329	+
	330	+ /**
	331	+ * Enqueue the next sync items for sending.
	332	+ * Will not be done if the current request is a WP import one.
	333	+ * Will be delayed until the next sync time comes.
	334	+ *
	335	+ * @access private
	336	+ */
	337	+ private function continue_full_sync_enqueue() {
	338	+ if ( defined( 'WP_IMPORTING' ) && WP_IMPORTING ) {
	339	+ return false;
	340	+ }
	341	+
	342	+ if ( $this->get_next_sync_time( 'full-sync-enqueue' ) > microtime( true ) ) {
	343	+ return false;
	344	+ }
	345	+
	346	+ $full_sync_module = Modules::get_module( 'full-sync' );
	347	+ '@phan-var Modules\Full_Sync_Immediately|Modules\Full_Sync $full_sync_module';
	348	+ $full_sync_module->continue_enqueuing();
	349	+
	350	+ $this->set_next_sync_time( time() + $this->get_enqueue_wait_time(), 'full-sync-enqueue' );
	351	+ }
	352	+
	353	+ /**
	354	+ * Trigger incremental sync.
	355	+ *
	356	+ * @access public
	357	+ *
	358	+ * @return boolean|WP_Error True if this sync sending was successful, error object otherwise.
	359	+ */
	360	+ public function do_sync() {
	361	+ // Sync directly during cron. We are doing this because otherwise
	362	+ // the dedicated sync flow would be spawning HTTP requests during cron shutdown,
	363	+ // which can be unreliable and cause sync lag for time-sensitive events like updates.
	364	+ if ( ! Settings::is_dedicated_sync_enabled() || Settings::is_doing_cron() ) {
	365	+ $result = $this->do_sync_and_set_delays( $this->sync_queue );
	366	+ } else {
	367	+ $result = Dedicated_Sender::spawn_sync( $this->sync_queue );
	368	+ }
	369	+
	370	+ return $result;
	371	+ }
	372	+
	373	+ /**
	374	+ * Trigger incremental sync and early exit on Dedicated Sync request.
	375	+ *
	376	+ * @access public
	377	+ *
	378	+ * @param bool $do_real_exit If we should exit at the end of the request. We should by default.
	379	+ * In the context of running this in the REST API, we actually want to return an error.
	380	+ *
	381	+ * @return void|WP_Error
	382	+ */
	383	+ public function do_dedicated_sync_and_exit( $do_real_exit = true ) {
	384	+ nocache_headers();
	385	+
	386	+ if ( ! Settings::is_dedicated_sync_enabled() ) {
	387	+ return new WP_Error( 'dedicated_sync_disabled', 'Dedicated Sync flow is disabled.' );
	388	+ }
	389	+
	390	+ if ( ! Dedicated_Sender::is_dedicated_sync_request() ) {
	391	+ return new WP_Error( 'non_dedicated_sync_request', 'Not a Dedicated Sync request.' );
	392	+ }
	393	+
	394	+ /**
	395	+ * Output an `OK` to show that Dedicated Sync is enabled and we can process events.
	396	+ * This is used to test the feature is working.
	397	+ *
	398	+ * @see \Automattic\Jetpack\Sync\Dedicated_Sender::can_spawn_dedicated_sync_request
	399	+ */
	400	+ // phpcs:ignore WordPress.Security.EscapeOutput.OutputNotEscaped -- This is just a constant string used for Validation.
	401	+ echo Dedicated_Sender::DEDICATED_SYNC_VALIDATION_STRING;
	402	+
	403	+ // Try to disconnect the request as quickly as possible and process things in the background.
	404	+ $this->fastcgi_finish_request();
	405	+
	406	+ /**
	407	+ * Close the PHP session to free up the server threads to handle other requests while we
	408	+ * send sync data with Dedicated Sync.
	409	+ *
	410	+ * When we spawn Dedicated Sync, we send `$_COOKIES` with the request to help out with any
	411	+ * firewall and/or caching functionality that might prevent us to ping the site directly.
	412	+ *
	413	+ * This will cause Dedicated Sync to reuse the visitor's PHP session and lock it until the
	414	+ * request finishes, which can take anywhere from 1 to 30+ seconds, depending on the server
	415	+ * `max_execution_time` configuration.
	416	+ *
	417	+ * By closing the session we're freeing up the session, so other requests can acquire the
	418	+ * lock and proceed with their own tasks.
	419	+ */
	420	+ if ( session_status() === PHP_SESSION_ACTIVE ) {
	421	+ session_write_close();
	422	+ }
	423	+
	424	+ // Actually try to send Sync events.
	425	+ $result = $this->do_sync_and_set_delays( $this->sync_queue );
	426	+
	427	+ // Output not used right now. Try to release dedicated sync lock
	428	+ Dedicated_Sender::try_release_lock_spawn_request();
	429	+
	430	+ // If no errors occurred, re-spawn a dedicated Sync request.
	431	+ if ( true === $result ) {
	432	+ Dedicated_Sender::spawn_sync( $this->sync_queue );
	433	+ }
	434	+
	435	+ if ( $do_real_exit ) {
	436	+ exit( 0 );
	437	+ }
	438	+ }
	439	+
	440	+ /**
	441	+ * Trigger sync for a certain sync queue.
	442	+ * Responsible for setting next sync time.
	443	+ * Will not be delayed if the current request is a WP import one.
	444	+ * Will be delayed until the next sync time comes.
	445	+ *
	446	+ * @access public
	447	+ *
	448	+ * @param \Automattic\Jetpack\Sync\Queue $queue Queue object.
	449	+ *
	450	+ * @return boolean|WP_Error True if this sync sending was successful, error object otherwise.
	451	+ */
	452	+ public function do_sync_and_set_delays( $queue ) {
	453	+ // Don't sync if importing.
	454	+ if ( defined( 'WP_IMPORTING' ) && WP_IMPORTING ) {
	455	+ return new WP_Error( 'is_importing' );
	456	+ }
	457	+
	458	+ // Don't sync if request is marked as read only.
	459	+ if ( Constants::is_true( 'JETPACK_SYNC_READ_ONLY' ) ) {
	460	+ return new WP_Error( 'jetpack_sync_read_only' );
	461	+ }
	462	+
	463	+ if ( ! Settings::is_sender_enabled( $queue->id ) ) {
	464	+ return new WP_Error( 'sender_disabled_for_queue_' . $queue->id );
	465	+ }
	466	+
	467	+ if ( get_transient( self::TEMP_SYNC_DISABLE_TRANSIENT_NAME ) ) {
	468	+ return new WP_Error( 'sender_temporarily_disabled_while_pulling' );
	469	+ }
	470	+
	471	+ // Return early if we've gotten a retry-after header response.
	472	+ $retry_time = get_option( Actions::RETRY_AFTER_PREFIX . $queue->id );
	473	+ if ( $retry_time ) {
	474	+ // If expired update to false but don't send. Send will occurr in new request to avoid race conditions.
	475	+ if ( microtime( true ) > $retry_time ) {
	476	+ update_option( Actions::RETRY_AFTER_PREFIX . $queue->id, false, false );
	477	+ }
	478	+ return new WP_Error( 'retry_after' );
	479	+ }
	480	+
	481	+ // Don't sync if we are throttled.
	482	+ if ( $this->get_next_sync_time( $queue->id ) > microtime( true ) ) {
	483	+ return new WP_Error( 'sync_throttled' );
	484	+ }
	485	+
	486	+ $start_time = microtime( true );
	487	+
	488	+ Settings::set_is_syncing( true );
	489	+
	490	+ $sync_result = $this->do_sync_for_queue( $queue );
	491	+
	492	+ Settings::set_is_syncing( false );
	493	+
	494	+ $exceeded_sync_wait_threshold = ( microtime( true ) - $start_time ) > (float) $this->get_sync_wait_threshold();
	495	+
	496	+ if ( is_wp_error( $sync_result ) ) {
	497	+ if ( 'unclosed_buffer' === $sync_result->get_error_code() ) {
	498	+ $this->set_next_sync_time( time() + self::QUEUE_LOCKED_SYNC_DELAY, $queue->id );
	499	+ }
	500	+ if ( 'wpcom_error' === $sync_result->get_error_code() ) {
	501	+ $this->set_next_sync_time( time() + self::WPCOM_ERROR_SYNC_DELAY, $queue->id );
	502	+ }
	503	+ } elseif ( $exceeded_sync_wait_threshold && ! Settings::is_doing_cron() ) {
	504	+ // If a send was slow, briefly pause before the next one.
	505	+ // Applies only to Dedicated/Normal Sync to avoid impacting user traffic;
	506	+ // cron jobs are exempt.
	507	+ $this->set_next_sync_time( time() + $this->get_sync_wait_time(), $queue->id );
	508	+ }
	509	+
	510	+ return $sync_result;
	511	+ }
	512	+
	513	+ /**
	514	+ * Retrieve the next sync items to send.
	515	+ *
	516	+ * @access public
	517	+ *
	518	+ * @param (array|\Automattic\Jetpack\Sync\Queue_Buffer) $buffer_or_items Queue buffer or array of objects.
	519	+ * @param boolean $encode Whether to encode the items.
	520	+ * @return array Sync items to send.
	521	+ */
	522	+ public function get_items_to_send( $buffer_or_items, $encode = true ) {
	523	+ // Track how long we've been processing so we can avoid request timeouts.
	524	+ $start_time = microtime( true );
	525	+ $upload_size = 0;
	526	+ $items_to_send = array();
	527	+ $items = is_array( $buffer_or_items ) ? $buffer_or_items : $buffer_or_items->get_items();
	528	+ if ( ! is_array( $items ) ) {
	529	+ $items = array();
	530	+ }
	531	+
	532	+ // Set up current screen to avoid errors rendering content.
	533	+ require_once ABSPATH . 'wp-admin/includes/class-wp-screen.php';
	534	+ require_once ABSPATH . 'wp-admin/includes/screen.php';
	535	+ set_current_screen( 'sync' );
	536	+ $skipped_items_ids = array();
	537	+ /**
	538	+ * We estimate the total encoded size as we go by encoding each item individually.
	539	+ * This is expensive, but the only way to really know :/
	540	+ */
	541	+ foreach ( $items as $key => $item ) {
	542	+ if ( ! is_array( $item ) ) {
	543	+ $skipped_items_ids[] = $key;
	544	+ continue;
	545	+ }
	546	+
	547	+ // Suspending cache addition help prevent overloading in memory cache of large sites.
	548	+ wp_suspend_cache_addition( true );
	549	+ /**
	550	+ * Modify the data within an action before it is serialized and sent to the server
	551	+ * For example, during full sync this expands Post ID's into full Post objects,
	552	+ * so that we don't have to serialize the whole object into the queue.
	553	+ *
	554	+ * @since 1.6.3
	555	+ * @since-jetpack 4.2.0
	556	+ *
	557	+ * @param array The action parameters
	558	+ * @param int The ID of the user who triggered the action
	559	+ */
	560	+ $item[1] = apply_filters( 'jetpack_sync_before_send_' . $item[0], $item[1], $item[2] );
	561	+ wp_suspend_cache_addition( false );
	562	+ // Serialization usage can lead to empty, null or false action_name. Lets skip as there is no information to send.
	563	+ if ( empty( $item[0] ) || false === $item[1] ) {
	564	+ $skipped_items_ids[] = $key;
	565	+ continue;
	566	+ }
	567	+ $encoded_item = $this->codec->encode( $item );
	568	+ $upload_size += strlen( $encoded_item );
	569	+ if ( $upload_size > $this->upload_max_bytes && array() !== $items_to_send ) {
	570	+ break;
	571	+ }
	572	+ $items_to_send[ $key ] = $encode ? $encoded_item : $item;
	573	+ if ( microtime( true ) - $start_time > $this->max_dequeue_time ) {
	574	+ break;
	575	+ }
	576	+ }
	577	+
	578	+ return array( $items_to_send, $skipped_items_ids, $items, microtime( true ) - $start_time );
	579	+ }
	580	+
	581	+ /**
	582	+ * If supported, flush all response data to the client and finish the request.
	583	+ * This allows for time consuming tasks to be performed without leaving the connection open.
	584	+ *
	585	+ * @access private
	586	+ */
	587	+ private function fastcgi_finish_request() {
	588	+ if ( function_exists( 'fastcgi_finish_request' ) ) {
	589	+ fastcgi_finish_request();
	590	+ }
	591	+ }
	592	+
	593	+ /**
	594	+ * Perform sync for a certain sync queue.
	595	+ *
	596	+ * @access public
	597	+ *
	598	+ * @param \Automattic\Jetpack\Sync\Queue $queue Queue object.
	599	+ *
	600	+ * @return boolean|WP_Error True if this sync sending was successful, error object otherwise.
	601	+ */
	602	+ public function do_sync_for_queue( $queue ) {
	603	+ do_action( 'jetpack_sync_before_send_queue_' . $queue->id );
	604	+ if ( $queue->size() === 0 ) {
	605	+ return new WP_Error( 'empty_queue_' . $queue->id );
	606	+ }
	607	+
	608	+ /**
	609	+ * Now that we're sure we are about to sync, try to ignore user abort
	610	+ * so we can avoid getting into a bad state.
	611	+ */
	612	+ // https://plugins.trac.wordpress.org/ticket/2041
	613	+ if ( function_exists( 'ignore_user_abort' ) ) {
	614	+ ignore_user_abort( true );
	615	+ }
	616	+
	617	+ /* Don't make the request block till we finish, if possible. */
	618	+ if ( Constants::is_true( 'REST_REQUEST' ) || Constants::is_true( 'XMLRPC_REQUEST' ) ) {
	619	+ $this->fastcgi_finish_request();
	620	+ }
	621	+
	622	+ $checkout_start_time = microtime( true );
	623	+
	624	+ $buffer = $queue->checkout_with_memory_limit( $this->dequeue_max_bytes, $this->upload_max_rows );
	625	+
	626	+ if ( ! $buffer ) {
	627	+ // Buffer has no items.
	628	+ return new WP_Error( 'empty_buffer' );
	629	+ }
	630	+
	631	+ if ( is_wp_error( $buffer ) ) {
	632	+ return $buffer;
	633	+ }
	634	+
	635	+ $checkout_duration = microtime( true ) - $checkout_start_time;
	636	+
	637	+ list( $items_to_send, $skipped_items_ids, $items, $preprocess_duration ) = $this->get_items_to_send( $buffer, true );
	638	+ if ( ! empty( $items_to_send ) ) {
	639	+ /**
	640	+ * Fires when data is ready to send to the server.
	641	+ * Return false or WP_Error to abort the sync (e.g. if there's an error)
	642	+ * The items will be automatically re-sent later
	643	+ *
	644	+ * @since 1.6.3
	645	+ * @since-jetpack 4.2.0
	646	+ *
	647	+ * @param array $data The action buffer
	648	+ * @param string $codec The codec name used to encode the data
	649	+ * @param double $time The current time
	650	+ * @param string $queue The queue used to send ('sync' or 'full_sync')
	651	+ * @param float $checkout_duration The duration of the checkout operation.
	652	+ * @param float $preprocess_duration The duration of the pre-process operation.
	653	+ * @param int $queue_size The size of the sync queue at the time of processing.
	654	+ */
	655	+ Settings::set_is_sending( true );
	656	+ $processed_item_ids = apply_filters( 'jetpack_sync_send_data', $items_to_send, $this->codec->name(), microtime( true ), $queue->id, $checkout_duration, $preprocess_duration, $queue->size(), $buffer->id );
	657	+ Settings::set_is_sending( false );
	658	+ } else {
	659	+ $processed_item_ids = $skipped_items_ids;
	660	+ $skipped_items_ids = array();
	661	+ }
	662	+
	663	+ if ( 'non-blocking' !== $processed_item_ids ) {
	664	+ if ( ! $processed_item_ids || is_wp_error( $processed_item_ids ) ) {
	665	+ $checked_in_item_ids = $queue->checkin( $buffer );
	666	+ if ( is_wp_error( $checked_in_item_ids ) ) {
	667	+ // phpcs:ignore WordPress.PHP.DevelopmentFunctions.error_log_error_log
	668	+ error_log( 'Error checking in buffer: ' . $checked_in_item_ids->get_error_message() );
	669	+ $queue->force_checkin();
	670	+ }
	671	+ if ( is_wp_error( $processed_item_ids ) ) {
	672	+ return new WP_Error( 'wpcom_error', $processed_item_ids->get_error_code() );
	673	+ }
	674	+
	675	+ // Returning a wpcom_error is a sign to the caller that we should wait a while before syncing again.
	676	+ return new WP_Error( 'wpcom_error', 'jetpack_sync_send_data_false' );
	677	+ } else {
	678	+ // Detect if the last item ID was an error.
	679	+ $had_wp_error = is_wp_error( end( $processed_item_ids ) );
	680	+ $wp_error = $had_wp_error ? array_pop( $processed_item_ids ) : null;
	681	+ // Also checkin any items that were skipped.
	682	+ if ( array() !== $skipped_items_ids ) {
	683	+ $processed_item_ids = array_merge( $processed_item_ids, $skipped_items_ids );
	684	+ }
	685	+ $processed_items = array_intersect_key( $items, array_flip( $processed_item_ids ) );
	686	+ /**
	687	+ * Allows us to keep track of all the actions that have been sent.
	688	+ * Allows us to calculate the progress of specific actions.
	689	+ *
	690	+ * @since 1.6.3
	691	+ * @since-jetpack 4.2.0
	692	+ *
	693	+ * @param array $processed_actions The actions that we send successfully.
	694	+ */
	695	+ do_action( 'jetpack_sync_processed_actions', $processed_items );
	696	+ $queue->close( $buffer, $processed_item_ids );
	697	+ // Returning a WP_Error is a sign to the caller that we should wait a while before syncing again.
	698	+ if ( $had_wp_error ) {
	699	+ return new WP_Error( 'wpcom_error', $wp_error->get_error_code() );
	700	+ }
	701	+ }
	702	+ }
	703	+
	704	+ return true;
	705	+ }
	706	+
	707	+ /**
	708	+ * Immediately sends a single item without firing or enqueuing it
	709	+ *
	710	+ * @param string $action_name The action.
	711	+ * @param array $data The data associated with the action.
	712	+ * @param string $key The key to use for the action.
	713	+ *
	714	+ * @return array Items processed. TODO: this doesn't make much sense anymore, it should probably be just a bool.
	715	+ */
	716	+ public function send_action( $action_name, $data = null, $key = null ) {
	717	+ if ( ! Settings::is_sender_enabled( 'full_sync' ) ) {
	718	+ return array();
	719	+ }
	720	+
	721	+ // Compose the data to be sent.
	722	+ $action_to_send = $this->create_action_to_send( $action_name, $data, $key );
	723	+
	724	+ list( $items_to_send, $skipped_items_ids, $items, $preprocess_duration ) = $this->get_items_to_send( $action_to_send, true ); // phpcs:ignore VariableAnalysis.CodeAnalysis.VariableAnalysis.UnusedVariable
	725	+ Settings::set_is_sending( true );
	726	+ $processed_item_ids = apply_filters( 'jetpack_sync_send_data', $items_to_send, $this->get_codec()->name(), microtime( true ), 'immediate-send', 0, $preprocess_duration );
	727	+ Settings::set_is_sending( false );
	728	+
	729	+ /**
	730	+ * Allows us to keep track of all the actions that have been sent.
	731	+ * Allows us to calculate the progress of specific actions.
	732	+ *
	733	+ * @param array $processed_actions The actions that we send successfully.
	734	+ *
	735	+ * @since 1.6.3
	736	+ * @since-jetpack 4.2.0
	737	+ */
	738	+ do_action( 'jetpack_sync_processed_actions', $action_to_send );
	739	+
	740	+ return $processed_item_ids;
	741	+ }
	742	+
	743	+ /**
	744	+ * Create an synthetic action for direct sending to WPCOM during full sync (for example)
	745	+ *
	746	+ * @access private
	747	+ *
	748	+ * @param string $action_name The action.
	749	+ * @param array $data The data associated with the action.
	750	+ * @param string $key The key to use for the action.
	751	+ * @return array An array of synthetic sync actions keyed by current microtime(true)
	752	+ */
	753	+ private function create_action_to_send( $action_name, $data, $key = null ) {
	754	+ return array(
	755	+ $key ?? (string) microtime( true ) => array(
	756	+ $action_name,
	757	+ $data,
	758	+ get_current_user_id(),
	759	+ microtime( true ),
	760	+ Settings::is_importing(),
	761	+ ),
	762	+ );
	763	+ }
	764	+
	765	+ /**
	766	+ * Returns any object that is able to be synced.
	767	+ *
	768	+ * @access public
	769	+ *
	770	+ * @param array $args the synchronized object parameters.
	771	+ * @return string Encoded sync object.
	772	+ */
	773	+ public function sync_object( $args ) {
	774	+ // For example: posts, post, 5.
	775	+ list( $module_name, $object_type, $id ) = $args;
	776	+
	777	+ $sync_module = Modules::get_module( $module_name );
	778	+ $codec = $this->get_codec();
	779	+
	780	+ return $codec->encode( $sync_module->get_object_by_id( $object_type, $id ) );
	781	+ }
	782	+
	783	+ /**
	784	+ * Register additional sync XML-RPC methods available to Jetpack for authenticated users.
	785	+ *
	786	+ * @access public
	787	+ * @since 1.6.3
	788	+ * @since-jetpack 7.8.0
	789	+ *
	790	+ * @param array $jetpack_methods XML-RPC methods available to the Jetpack Server.
	791	+ * @return array Filtered XML-RPC methods.
	792	+ */
	793	+ public function register_jetpack_xmlrpc_methods( $jetpack_methods ) {
	794	+ $jetpack_methods['jetpack.syncObject'] = array( $this, 'sync_object' );
	795	+ return $jetpack_methods;
	796	+ }
	797	+
	798	+ /**
	799	+ * Get the incremental sync queue object.
	800	+ *
	801	+ * @access public
	802	+ *
	803	+ * @return \Automattic\Jetpack\Sync\Queue Queue object.
	804	+ */
	805	+ public function get_sync_queue() {
	806	+ return $this->sync_queue;
	807	+ }
	808	+
	809	+ /**
	810	+ * Get the full sync queue object.
	811	+ *
	812	+ * @access public
	813	+ *
	814	+ * @return \Automattic\Jetpack\Sync\Queue Queue object.
	815	+ */
	816	+ public function get_full_sync_queue() {
	817	+ return $this->full_sync_queue;
	818	+ }
	819	+
	820	+ /**
	821	+ * Get the codec object.
	822	+ *
	823	+ * @access public
	824	+ *
	825	+ * @return \Automattic\Jetpack\Sync\Codec_Interface Codec object.
	826	+ */
	827	+ public function get_codec() {
	828	+ return $this->codec;
	829	+ }
	830	+
	831	+ /**
	832	+ * Determine the codec object.
	833	+ * Use gzip deflate if supported.
	834	+ *
	835	+ * @access public
	836	+ */
	837	+ public function set_codec() {
	838	+ if ( function_exists( 'gzinflate' ) ) {
	839	+ $this->codec = new JSON_Deflate_Array_Codec();
	840	+ } else {
	841	+ $this->codec = new Simple_Codec();
	842	+ }
	843	+ }
	844	+
	845	+ /**
	846	+ * Compute and send all the checksums.
	847	+ *
	848	+ * @access public
	849	+ */
	850	+ public function send_checksum() {
	851	+ $store = new Replicastore();
	852	+ do_action( 'jetpack_sync_checksum', $store->checksum_all() );
	853	+ }
	854	+
	855	+ /**
	856	+ * Reset the incremental sync queue.
	857	+ *
	858	+ * @access public
	859	+ */
	860	+ public function reset_sync_queue() {
	861	+ $this->sync_queue->reset();
	862	+ }
	863	+
	864	+ /**
	865	+ * Reset the full sync queue.
	866	+ *
	867	+ * @access public
	868	+ */
	869	+ public function reset_full_sync_queue() {
	870	+ $this->full_sync_queue->reset();
	871	+ }
	872	+
	873	+ /**
	874	+ * Set the maximum bytes to checkout without exceeding the memory limit.
	875	+ *
	876	+ * @access public
	877	+ *
	878	+ * @param int $size Maximum bytes to checkout.
	879	+ */
	880	+ public function set_dequeue_max_bytes( $size ) {
	881	+ $this->dequeue_max_bytes = $size;
	882	+ }
	883	+
	884	+ /**
	885	+ * Set the maximum bytes in a single encoded item.
	886	+ *
	887	+ * @access public
	888	+ *
	889	+ * @param int $max_bytes Maximum bytes in a single encoded item.
	890	+ */
	891	+ public function set_upload_max_bytes( $max_bytes ) {
	892	+ $this->upload_max_bytes = $max_bytes;
	893	+ }
	894	+
	895	+ /**
	896	+ * Set the maximum number of sync items in a single action.
	897	+ *
	898	+ * @access public
	899	+ *
	900	+ * @param int $max_rows Maximum number of sync items.
	901	+ */
	902	+ public function set_upload_max_rows( $max_rows ) {
	903	+ $this->upload_max_rows = $max_rows;
	904	+ }
	905	+
	906	+ /**
	907	+ * Set the sync wait time (in seconds).
	908	+ *
	909	+ * @access public
	910	+ *
	911	+ * @param int $seconds Sync wait time.
	912	+ */
	913	+ public function set_sync_wait_time( $seconds ) {
	914	+ $this->sync_wait_time = $seconds;
	915	+ }
	916	+
	917	+ /**
	918	+ * Get current sync wait time (in seconds).
	919	+ *
	920	+ * @access public
	921	+ *
	922	+ * @return int Sync wait time.
	923	+ */
	924	+ public function get_sync_wait_time() {
	925	+ return $this->sync_wait_time;
	926	+ }
	927	+
	928	+ /**
	929	+ * Set the enqueue wait time (in seconds).
	930	+ *
	931	+ * @access public
	932	+ *
	933	+ * @param int $seconds Enqueue wait time.
	934	+ */
	935	+ public function set_enqueue_wait_time( $seconds ) {
	936	+ $this->enqueue_wait_time = $seconds;
	937	+ }
	938	+
	939	+ /**
	940	+ * Get current enqueue wait time (in seconds).
	941	+ *
	942	+ * @access public
	943	+ *
	944	+ * @return int Enqueue wait time.
	945	+ */
	946	+ public function get_enqueue_wait_time() {
	947	+ return $this->enqueue_wait_time;
	948	+ }
	949	+
	950	+ /**
	951	+ * Set the sync wait threshold (in seconds).
	952	+ *
	953	+ * @access public
	954	+ *
	955	+ * @param int $seconds Sync wait threshold.
	956	+ */
	957	+ public function set_sync_wait_threshold( $seconds ) {
	958	+ $this->sync_wait_threshold = $seconds;
	959	+ }
	960	+
	961	+ /**
	962	+ * Get current sync wait threshold (in seconds).
	963	+ *
	964	+ * @access public
	965	+ *
	966	+ * @return int Sync wait threshold.
	967	+ */
	968	+ public function get_sync_wait_threshold() {
	969	+ return $this->sync_wait_threshold;
	970	+ }
	971	+
	972	+ /**
	973	+ * Set the maximum time for perfirming a checkout of items from the queue (in seconds).
	974	+ *
	975	+ * @access public
	976	+ *
	977	+ * @param int $seconds Maximum dequeue time.
	978	+ */
	979	+ public function set_max_dequeue_time( $seconds ) {
	980	+ $this->max_dequeue_time = $seconds;
	981	+ }
	982	+
	983	+ /**
	984	+ * Initialize the sync queues, codec and set the default settings.
	985	+ *
	986	+ * @access public
	987	+ */
	988	+ public function set_defaults() {
	989	+ $this->sync_queue = new Queue( 'sync' );
	990	+ $this->full_sync_queue = new Queue( 'full_sync' );
	991	+ $this->set_codec();
	992	+
	993	+ // Saved settings.
	994	+ Settings::set_importing( null );
	995	+ $settings = Settings::get_settings();
	996	+ $this->set_dequeue_max_bytes( $settings['dequeue_max_bytes'] );
	997	+ $this->set_upload_max_bytes( $settings['upload_max_bytes'] );
	998	+ $this->set_upload_max_rows( $settings['upload_max_rows'] );
	999	+ $this->set_sync_wait_time( $settings['sync_wait_time'] );
	1000	+ $this->set_enqueue_wait_time( $settings['enqueue_wait_time'] );
	1001	+ $this->set_sync_wait_threshold( $settings['sync_wait_threshold'] );
	1002	+ $this->set_max_dequeue_time( Defaults::get_max_sync_execution_time() );
	1003	+ }
	1004	+
	1005	+ /**
	1006	+ * Reset sync queues, modules and settings.
	1007	+ *
	1008	+ * @access public
	1009	+ */
	1010	+ public function reset_data() {
	1011	+ $this->reset_sync_queue();
	1012	+ $this->reset_full_sync_queue();
	1013	+
	1014	+ foreach ( Modules::get_modules() as $module ) {
	1015	+ $module->reset_data();
	1016	+ }
	1017	+ // Reset Sync locks without unlocking queues since we already reset those.
	1018	+ Actions::reset_sync_locks( false );
	1019	+
	1020	+ Settings::reset_data();
	1021	+ }
	1022	+
	1023	+ /**
	1024	+ * Perform cleanup at the event of plugin uninstallation.
	1025	+ *
	1026	+ * @access public
	1027	+ */
	1028	+ public function uninstall() {
	1029	+ // Lets delete all the other fun stuff like transient and option and the sync queue.
	1030	+ $this->reset_data();
	1031	+
	1032	+ // Delete the full sync status.
	1033	+ delete_option( 'jetpack_full_sync_status' );
	1034	+
	1035	+ // Clear the sync cron.
	1036	+ wp_clear_scheduled_hook( 'jetpack_sync_cron' );
	1037	+ wp_clear_scheduled_hook( 'jetpack_sync_full_cron' );
	1038	+ }
	1039	+}