class.jetpack-sync-sender.php 13 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406
  1. <?php
  2. require_once dirname( __FILE__ ) . '/class.jetpack-sync-queue.php';
  3. require_once dirname( __FILE__ ) . '/class.jetpack-sync-defaults.php';
  4. require_once dirname( __FILE__ ) . '/class.jetpack-sync-json-deflate-array-codec.php';
  5. require_once dirname( __FILE__ ) . '/class.jetpack-sync-simple-codec.php';
  6. require_once dirname( __FILE__ ) . '/class.jetpack-sync-modules.php';
  7. require_once dirname( __FILE__ ) . '/class.jetpack-sync-settings.php';
  8. /**
  9. * This class grabs pending actions from the queue and sends them
  10. */
  11. class Jetpack_Sync_Sender {
  12. const NEXT_SYNC_TIME_OPTION_NAME = 'jetpack_next_sync_time';
  13. const WPCOM_ERROR_SYNC_DELAY = 60;
  14. const QUEUE_LOCKED_SYNC_DELAY = 10;
  15. private $dequeue_max_bytes;
  16. private $upload_max_bytes;
  17. private $upload_max_rows;
  18. private $max_dequeue_time;
  19. private $sync_wait_time;
  20. private $sync_wait_threshold;
  21. private $enqueue_wait_time;
  22. private $sync_queue;
  23. private $full_sync_queue;
  24. private $codec;
  25. private $old_user;
  26. // singleton functions
  27. private static $instance;
  28. public static function get_instance() {
  29. if ( null === self::$instance ) {
  30. self::$instance = new self();
  31. }
  32. return self::$instance;
  33. }
  34. // this is necessary because you can't use "new" when you declare instance properties >:(
  35. protected function __construct() {
  36. $this->set_defaults();
  37. $this->init();
  38. }
  39. private function init() {
  40. add_action( 'jetpack_sync_before_send_queue_sync', array( $this, 'maybe_set_user_from_token' ), 1 );
  41. add_action( 'jetpack_sync_before_send_queue_sync', array( $this, 'maybe_clear_user_from_token' ), 20 );
  42. foreach ( Jetpack_Sync_Modules::get_modules() as $module ) {
  43. $module->init_before_send();
  44. }
  45. }
  46. public function maybe_set_user_from_token( ) {
  47. $jetpack = Jetpack::init();
  48. $verified_user = $jetpack->verify_xml_rpc_signature();
  49. if ( Jetpack_Constants::is_true( 'XMLRPC_REQUEST' ) &&
  50. ! is_wp_error( $verified_user )
  51. && $verified_user
  52. ) {
  53. $old_user = wp_get_current_user();
  54. $this->old_user = isset( $old_user->ID ) ? $old_user->ID : 0;
  55. wp_set_current_user( $verified_user['user_id'] );
  56. }
  57. }
  58. public function maybe_clear_user_from_token() {
  59. if ( isset( $this->old_user ) ) {
  60. wp_set_current_user( $this->old_user );
  61. }
  62. }
  63. public function get_next_sync_time( $queue_name ) {
  64. return (double) get_option( self::NEXT_SYNC_TIME_OPTION_NAME . '_' . $queue_name, 0 );
  65. }
  66. public function set_next_sync_time( $time, $queue_name ) {
  67. return update_option( self::NEXT_SYNC_TIME_OPTION_NAME . '_' . $queue_name, $time, true );
  68. }
  69. public function do_full_sync() {
  70. if ( ! Jetpack_Sync_Modules::get_module( 'full-sync' ) ) {
  71. return;
  72. }
  73. $this->continue_full_sync_enqueue();
  74. return $this->do_sync_and_set_delays( $this->full_sync_queue );
  75. }
  76. private function continue_full_sync_enqueue() {
  77. if ( defined( 'WP_IMPORTING' ) && WP_IMPORTING ) {
  78. return false;
  79. }
  80. if ( $this->get_next_sync_time( 'full-sync-enqueue' ) > microtime( true ) ) {
  81. return false;
  82. }
  83. Jetpack_Sync_Modules::get_module( 'full-sync' )->continue_enqueuing();
  84. $this->set_next_sync_time( time() + $this->get_enqueue_wait_time(), 'full-sync-enqueue' );
  85. }
  86. public function do_sync() {
  87. return $this->do_sync_and_set_delays( $this->sync_queue );
  88. }
  89. public function do_sync_and_set_delays( $queue ) {
  90. // don't sync if importing
  91. if ( defined( 'WP_IMPORTING' ) && WP_IMPORTING ) {
  92. return new WP_Error( 'is_importing' );
  93. }
  94. // don't sync if we are throttled
  95. if ( $this->get_next_sync_time( $queue->id ) > microtime( true ) ) {
  96. return new WP_Error( 'sync_throttled' );
  97. }
  98. $start_time = microtime( true );
  99. Jetpack_Sync_Settings::set_is_syncing( true );
  100. $sync_result = $this->do_sync_for_queue( $queue );
  101. Jetpack_Sync_Settings::set_is_syncing( false );
  102. $exceeded_sync_wait_threshold = ( microtime( true ) - $start_time ) > (double) $this->get_sync_wait_threshold();
  103. if ( is_wp_error( $sync_result ) ) {
  104. if ( 'unclosed_buffer' === $sync_result->get_error_code() ) {
  105. $this->set_next_sync_time( time() + self::QUEUE_LOCKED_SYNC_DELAY, $queue->id );
  106. }
  107. if ( 'wpcom_error' === $sync_result->get_error_code() ) {
  108. $this->set_next_sync_time( time() + self::WPCOM_ERROR_SYNC_DELAY, $queue->id );
  109. }
  110. } elseif ( $exceeded_sync_wait_threshold ) {
  111. // if we actually sent data and it took a while, wait before sending again
  112. $this->set_next_sync_time( time() + $this->get_sync_wait_time(), $queue->id );
  113. }
  114. return $sync_result;
  115. }
  116. public function get_items_to_send( $buffer, $encode = true ) {
  117. // track how long we've been processing so we can avoid request timeouts
  118. $start_time = microtime( true );
  119. $upload_size = 0;
  120. $items_to_send = array();
  121. $items = $buffer->get_items();
  122. // set up current screen to avoid errors rendering content
  123. require_once( ABSPATH . 'wp-admin/includes/class-wp-screen.php' );
  124. require_once( ABSPATH . 'wp-admin/includes/screen.php' );
  125. set_current_screen( 'sync' );
  126. $skipped_items_ids = array();
  127. // we estimate the total encoded size as we go by encoding each item individually
  128. // this is expensive, but the only way to really know :/
  129. foreach ( $items as $key => $item ) {
  130. // Suspending cache addition help prevent overloading in memory cache of large sites.
  131. wp_suspend_cache_addition( true );
  132. /**
  133. * Modify the data within an action before it is serialized and sent to the server
  134. * For example, during full sync this expands Post ID's into full Post objects,
  135. * so that we don't have to serialize the whole object into the queue.
  136. *
  137. * @since 4.2.0
  138. *
  139. * @param array The action parameters
  140. * @param int The ID of the user who triggered the action
  141. */
  142. $item[1] = apply_filters( 'jetpack_sync_before_send_' . $item[0], $item[1], $item[2] );
  143. wp_suspend_cache_addition( false );
  144. if ( $item[1] === false ) {
  145. $skipped_items_ids[] = $key;
  146. continue;
  147. }
  148. $encoded_item = $encode ? $this->codec->encode( $item ) : $item;
  149. $upload_size += strlen( $encoded_item );
  150. if ( $upload_size > $this->upload_max_bytes && count( $items_to_send ) > 0 ) {
  151. break;
  152. }
  153. $items_to_send[ $key ] = $encoded_item;
  154. if ( microtime(true) - $start_time > $this->max_dequeue_time ) {
  155. break;
  156. }
  157. }
  158. return array( $items_to_send, $skipped_items_ids, $items, microtime( true ) - $start_time );
  159. }
  160. public function do_sync_for_queue( $queue ) {
  161. do_action( 'jetpack_sync_before_send_queue_' . $queue->id );
  162. if ( $queue->size() === 0 ) {
  163. return new WP_Error( 'empty_queue_' . $queue->id );
  164. }
  165. // now that we're sure we are about to sync, try to
  166. // ignore user abort so we can avoid getting into a
  167. // bad state
  168. if ( function_exists( 'ignore_user_abort' ) ) {
  169. ignore_user_abort( true );
  170. }
  171. $checkout_start_time = microtime( true );
  172. $buffer = $queue->checkout_with_memory_limit( $this->dequeue_max_bytes, $this->upload_max_rows );
  173. if ( ! $buffer ) {
  174. // buffer has no items
  175. return new WP_Error( 'empty_buffer' );
  176. }
  177. if ( is_wp_error( $buffer ) ) {
  178. return $buffer;
  179. }
  180. $checkout_duration = microtime( true ) - $checkout_start_time;
  181. list( $items_to_send, $skipped_items_ids, $items, $preprocess_duration ) = $this->get_items_to_send( $buffer, true );
  182. if ( ! empty( $items_to_send ) ) {
  183. /**
  184. * Fires when data is ready to send to the server.
  185. * Return false or WP_Error to abort the sync (e.g. if there's an error)
  186. * The items will be automatically re-sent later
  187. *
  188. * @since 4.2.0
  189. *
  190. * @param array $data The action buffer
  191. * @param string $codec The codec name used to encode the data
  192. * @param double $time The current time
  193. * @param string $queue The queue used to send ('sync' or 'full_sync')
  194. */
  195. Jetpack_Sync_Settings::set_is_sending( true );
  196. $processed_item_ids = apply_filters( 'jetpack_sync_send_data', $items_to_send, $this->codec->name(), microtime( true ), $queue->id, $checkout_duration, $preprocess_duration );
  197. Jetpack_Sync_Settings::set_is_sending( false );
  198. } else {
  199. $processed_item_ids = $skipped_items_ids;
  200. $skipped_items_ids = array();
  201. }
  202. if ( ! $processed_item_ids || is_wp_error( $processed_item_ids ) ) {
  203. $checked_in_item_ids = $queue->checkin( $buffer );
  204. if ( is_wp_error( $checked_in_item_ids ) ) {
  205. error_log( 'Error checking in buffer: ' . $checked_in_item_ids->get_error_message() );
  206. $queue->force_checkin();
  207. }
  208. if ( is_wp_error( $processed_item_ids ) ) {
  209. return new WP_Error( 'wpcom_error', $processed_item_ids->get_error_code() );
  210. }
  211. // returning a WP_Error('wpcom_error') is a sign to the caller that we should wait a while
  212. // before syncing again
  213. return new WP_Error( 'wpcom_error', 'jetpack_sync_send_data_false' );
  214. } else {
  215. // detect if the last item ID was an error
  216. $had_wp_error = is_wp_error( end( $processed_item_ids ) );
  217. if ( $had_wp_error ) {
  218. $wp_error = array_pop( $processed_item_ids );
  219. }
  220. // also checkin any items that were skipped
  221. if ( count( $skipped_items_ids ) > 0 ) {
  222. $processed_item_ids = array_merge( $processed_item_ids, $skipped_items_ids );
  223. }
  224. $processed_items = array_intersect_key( $items, array_flip( $processed_item_ids ) );
  225. /**
  226. * Allows us to keep track of all the actions that have been sent.
  227. * Allows us to calculate the progress of specific actions.
  228. *
  229. * @since 4.2.0
  230. *
  231. * @param array $processed_actions The actions that we send successfully.
  232. */
  233. do_action( 'jetpack_sync_processed_actions', $processed_items );
  234. $queue->close( $buffer, $processed_item_ids );
  235. // returning a WP_Error is a sign to the caller that we should wait a while
  236. // before syncing again
  237. if ( $had_wp_error ) {
  238. return new WP_Error( 'wpcom_error', $wp_error->get_error_code() );
  239. }
  240. }
  241. return true;
  242. }
  243. function get_sync_queue() {
  244. return $this->sync_queue;
  245. }
  246. function get_full_sync_queue() {
  247. return $this->full_sync_queue;
  248. }
  249. function get_codec() {
  250. return $this->codec;
  251. }
  252. function set_codec() {
  253. if ( function_exists( 'gzinflate' ) ) {
  254. $this->codec = new Jetpack_Sync_JSON_Deflate_Array_Codec();
  255. } else {
  256. $this->codec = new Jetpack_Sync_Simple_Codec();
  257. }
  258. }
  259. function send_checksum() {
  260. require_once 'class.jetpack-sync-wp-replicastore.php';
  261. $store = new Jetpack_Sync_WP_Replicastore();
  262. do_action( 'jetpack_sync_checksum', $store->checksum_all() );
  263. }
  264. function reset_sync_queue() {
  265. $this->sync_queue->reset();
  266. }
  267. function reset_full_sync_queue() {
  268. $this->full_sync_queue->reset();
  269. }
  270. function set_dequeue_max_bytes( $size ) {
  271. $this->dequeue_max_bytes = $size;
  272. }
  273. // in bytes
  274. function set_upload_max_bytes( $max_bytes ) {
  275. $this->upload_max_bytes = $max_bytes;
  276. }
  277. // in rows
  278. function set_upload_max_rows( $max_rows ) {
  279. $this->upload_max_rows = $max_rows;
  280. }
  281. // in seconds
  282. function set_sync_wait_time( $seconds ) {
  283. $this->sync_wait_time = $seconds;
  284. }
  285. function get_sync_wait_time() {
  286. return $this->sync_wait_time;
  287. }
  288. function set_enqueue_wait_time( $seconds ) {
  289. $this->enqueue_wait_time = $seconds;
  290. }
  291. function get_enqueue_wait_time() {
  292. return $this->enqueue_wait_time;
  293. }
  294. // in seconds
  295. function set_sync_wait_threshold( $seconds ) {
  296. $this->sync_wait_threshold = $seconds;
  297. }
  298. function get_sync_wait_threshold() {
  299. return $this->sync_wait_threshold;
  300. }
  301. // in seconds
  302. function set_max_dequeue_time( $seconds ) {
  303. $this->max_dequeue_time = $seconds;
  304. }
  305. function set_defaults() {
  306. $this->sync_queue = new Jetpack_Sync_Queue( 'sync' );
  307. $this->full_sync_queue = new Jetpack_Sync_Queue( 'full_sync' );
  308. $this->set_codec();
  309. // saved settings
  310. Jetpack_Sync_Settings::set_importing( null );
  311. $settings = Jetpack_Sync_Settings::get_settings();
  312. $this->set_dequeue_max_bytes( $settings['dequeue_max_bytes'] );
  313. $this->set_upload_max_bytes( $settings['upload_max_bytes'] );
  314. $this->set_upload_max_rows( $settings['upload_max_rows'] );
  315. $this->set_sync_wait_time( $settings['sync_wait_time'] );
  316. $this->set_enqueue_wait_time( $settings['enqueue_wait_time'] );
  317. $this->set_sync_wait_threshold( $settings['sync_wait_threshold'] );
  318. $this->set_max_dequeue_time( Jetpack_Sync_Defaults::get_max_sync_execution_time() );
  319. }
  320. function reset_data() {
  321. $this->reset_sync_queue();
  322. $this->reset_full_sync_queue();
  323. foreach ( Jetpack_Sync_Modules::get_modules() as $module ) {
  324. $module->reset_data();
  325. }
  326. foreach ( array( 'sync', 'full_sync', 'full-sync-enqueue' ) as $queue_name ) {
  327. delete_option( self::NEXT_SYNC_TIME_OPTION_NAME . '_' . $queue_name );
  328. }
  329. Jetpack_Sync_Settings::reset_data();
  330. }
  331. function uninstall() {
  332. // Lets delete all the other fun stuff like transient and option and the sync queue
  333. $this->reset_data();
  334. // delete the full sync status
  335. delete_option( 'jetpack_full_sync_status' );
  336. // clear the sync cron.
  337. wp_clear_scheduled_hook( 'jetpack_sync_cron' );
  338. wp_clear_scheduled_hook( 'jetpack_sync_full_cron' );
  339. }
  340. }