Completed
Push — fix/language-packs ( c707c6...c0d674 )
by
unknown
214:41 queued 204:01
created

Jetpack_Sync_Sender::do_full_sync()   A

Complexity

Conditions 1
Paths 1

Size

Total Lines 3
Code Lines 2

Duplication

Lines 0
Ratio 0 %

Importance

Changes 0
Metric Value
cc 1
eloc 2
nc 1
nop 0
dl 0
loc 3
rs 10
c 0
b 0
f 0
1
<?php
2
3
require_once dirname( __FILE__ ) . '/class.jetpack-sync-queue.php';
4
require_once dirname( __FILE__ ) . '/class.jetpack-sync-defaults.php';
5
require_once dirname( __FILE__ ) . '/class.jetpack-sync-json-deflate-array-codec.php';
6
require_once dirname( __FILE__ ) . '/class.jetpack-sync-modules.php';
7
require_once dirname( __FILE__ ) . '/class.jetpack-sync-settings.php';
8
9
/**
10
 * This class grabs pending actions from the queue and sends them
11
 */
12
class Jetpack_Sync_Sender {
13
14
	const NEXT_SYNC_TIME_OPTION_NAME = 'jetpack_next_sync_time';
15
	const WPCOM_ERROR_SYNC_DELAY = 60;
16
	const QUEUE_LOCKED_SYNC_DELAY = 10;
17
18
	private $dequeue_max_bytes;
19
	private $upload_max_bytes;
20
	private $upload_max_rows;
21
	private $max_dequeue_time;
22
	private $sync_wait_time;
23
	private $sync_wait_threshold;
24
	private $sync_queue;
25
	private $full_sync_queue;
26
	private $codec;
27
28
	// singleton functions
29
	private static $instance;
30
31
	public static function get_instance() {
32
		if ( null === self::$instance ) {
33
			self::$instance = new self();
34
		}
35
36
		return self::$instance;
37
	}
38
39
	// this is necessary because you can't use "new" when you declare instance properties >:(
40
	protected function __construct() {
41
		$this->set_defaults();
42
		$this->init();
43
	}
44
45
	private function init() {
46
		foreach ( Jetpack_Sync_Modules::get_modules() as $module ) {
47
			$module->init_before_send();
48
		}
49
	}
50
51
	public function get_next_sync_time( $queue_name ) {
52
		return (double) get_option( self::NEXT_SYNC_TIME_OPTION_NAME . '_' . $queue_name, 0 );
53
	}
54
55
	public function set_next_sync_time( $time, $queue_name ) {
56
		return update_option( self::NEXT_SYNC_TIME_OPTION_NAME . '_' . $queue_name, $time, true );
57
	}
58
59
	public function do_full_sync() {
60
		return $this->do_sync_and_set_delays( $this->full_sync_queue );
61
	}
62
63
	public function do_sync() {
64
		return $this->do_sync_and_set_delays( $this->sync_queue );
65
	}
66
67
	public function do_sync_and_set_delays( $queue ) {
68
		// don't sync if importing
69
		if ( defined( 'WP_IMPORTING' ) && WP_IMPORTING ) {
70
			return false;
71
		}
72
73
		// don't sync if we are throttled
74
		if ( $this->get_next_sync_time( $queue->id ) > microtime( true ) ) {
75
			return false;
76
		}
77
78
		$start_time = microtime( true );
79
80
		$sync_result = $this->do_sync_for_queue( $queue );
81
82
		$exceeded_sync_wait_threshold = ( microtime( true ) - $start_time ) > (double) $this->get_sync_wait_threshold();
83
		
84
		if ( is_wp_error( $sync_result ) ) {
85
			if ( 'unclosed_buffer' === $sync_result->get_error_code() ) {
86
				$this->set_next_sync_time( time() + self::QUEUE_LOCKED_SYNC_DELAY, $queue->id );
87
			} else {
88
				$this->set_next_sync_time( time() + self::WPCOM_ERROR_SYNC_DELAY, $queue->id );
89
			}
90
			$sync_result = false;
91
		} elseif ( $exceeded_sync_wait_threshold ) {
92
			// if we actually sent data and it took a while, wait before sending again
93
			$this->set_next_sync_time( time() + $this->get_sync_wait_time(), $queue->id );
94
		}
95
96
		return $sync_result;
97
	}
98
99
	public function get_items_to_send( $buffer, $encode = true ) {
100
		// track how long we've been processing so we can avoid request timeouts
101
		$start_time = microtime( true );
102
		$upload_size   = 0;
103
		$items_to_send = array();
104
		$items         = $buffer->get_items();
105
		// set up current screen to avoid errors rendering content
106
		require_once( ABSPATH . 'wp-admin/includes/class-wp-screen.php' );
107
		require_once( ABSPATH . 'wp-admin/includes/screen.php' );
108
		set_current_screen( 'sync' );
109
		$skipped_items_ids = array();
110
		// we estimate the total encoded size as we go by encoding each item individually
111
		// this is expensive, but the only way to really know :/
112
		foreach ( $items as $key => $item ) {
113
			// Suspending cache addition help prevent overloading in memory cache of large sites.
114
			wp_suspend_cache_addition( true );
115
			/**
116
			 * Modify the data within an action before it is serialized and sent to the server
117
			 * For example, during full sync this expands Post ID's into full Post objects,
118
			 * so that we don't have to serialize the whole object into the queue.
119
			 *
120
			 * @since 4.2.0
121
			 *
122
			 * @param array The action parameters
123
			 * @param int The ID of the user who triggered the action
124
			 */
125
			$item[1] = apply_filters( 'jetpack_sync_before_send_' . $item[0], $item[1], $item[2] );
126
			wp_suspend_cache_addition( false );
127
			if ( $item[1] === false ) {
128
				$skipped_items_ids[] = $key;
129
				continue;
130
			}
131
			$encoded_item = $encode ? $this->codec->encode( $item ) : $item;
132
			$upload_size += strlen( $encoded_item );
133
			if ( $upload_size > $this->upload_max_bytes && count( $items_to_send ) > 0 ) {
134
				break;
135
			}
136
			$items_to_send[ $key ] = $encoded_item;
137
			if ( microtime(true) - $start_time > $this->max_dequeue_time ) {
138
				break;
139
			}
140
		}
141
142
		return array( $items_to_send, $skipped_items_ids, $items );
143
	}
144
145
	public function do_sync_for_queue( $queue ) {
146
147
		do_action( 'jetpack_sync_before_send_queue_' . $queue->id );
148
		if ( $queue->size() === 0 ) {
149
			return false;
150
		}
151
		// now that we're sure we are about to sync, try to
152
		// ignore user abort so we can avoid getting into a
153
		// bad state
154
		if ( function_exists( 'ignore_user_abort' ) ) {
155
			ignore_user_abort( true );
156
		}
157
		$buffer = $queue->checkout_with_memory_limit( $this->dequeue_max_bytes, $this->upload_max_rows );
158
		if ( ! $buffer ) {
159
			// buffer has no items
160
			return false;
161
		}
162
		if ( is_wp_error( $buffer ) ) {
163
			return $buffer;
164
		}
165
166
		list( $items_to_send, $skipped_items_ids, $items ) = $this->get_items_to_send( $buffer, true );
167
168
		/**
169
		 * Fires when data is ready to send to the server.
170
		 * Return false or WP_Error to abort the sync (e.g. if there's an error)
171
		 * The items will be automatically re-sent later
172
		 *
173
		 * @since 4.2.0
174
		 *
175
		 * @param array $data The action buffer
176
		 * @param string $codec The codec name used to encode the data
177
		 * @param double $time The current time
178
		 * @param string $queue The queue used to send ('sync' or 'full_sync')
179
		 */
180
		$processed_item_ids = apply_filters( 'jetpack_sync_send_data', $items_to_send, $this->codec->name(), microtime( true ), $queue->id );
181
		if ( ! $processed_item_ids || is_wp_error( $processed_item_ids ) ) {
182
			$checked_in_item_ids = $queue->checkin( $buffer );
183
			if ( is_wp_error( $checked_in_item_ids ) ) {
184
				error_log( 'Error checking in buffer: ' . $checked_in_item_ids->get_error_message() );
185
				$queue->force_checkin();
186
			}
187
			if ( is_wp_error( $processed_item_ids ) ) {
188
				return $processed_item_ids;
189
			}
190
			// returning a WP_Error is a sign to the caller that we should wait a while
191
			// before syncing again
192
			return new WP_Error( 'server_error' );
193
		} else {
194
			// detect if the last item ID was an error
195
			$had_wp_error = is_wp_error( end( $processed_item_ids ) );
196
			if ( $had_wp_error ) {
197
				$wp_error = array_pop( $processed_item_ids );
198
			}
199
			// also checkin any items that were skipped
200
			if ( count( $skipped_items_ids ) > 0 ) {
201
				$processed_item_ids = array_merge( $processed_item_ids, $skipped_items_ids );
202
			}
203
			$processed_items = array_intersect_key( $items, array_flip( $processed_item_ids ) );
204
			/**
205
			 * Allows us to keep track of all the actions that have been sent.
206
			 * Allows us to calculate the progress of specific actions.
207
			 *
208
			 * @since 4.2.0
209
			 *
210
			 * @param array $processed_actions The actions that we send successfully.
211
			 */
212
			do_action( 'jetpack_sync_processed_actions', $processed_items );
213
			$queue->close( $buffer, $processed_item_ids );
214
			// returning a WP_Error is a sign to the caller that we should wait a while
215
			// before syncing again
216
			if ( $had_wp_error ) {
217
				return $wp_error;
0 ignored issues
show
Bug introduced by
The variable $wp_error does not seem to be defined for all execution paths leading up to this point.

If you define a variable conditionally, it can happen that it is not defined for all execution paths.

Let’s take a look at an example:

function myFunction($a) {
    switch ($a) {
        case 'foo':
            $x = 1;
            break;

        case 'bar':
            $x = 2;
            break;
    }

    // $x is potentially undefined here.
    echo $x;
}

In the above example, the variable $x is defined if you pass “foo” or “bar” as argument for $a. However, since the switch statement has no default case statement, if you pass any other value, the variable $x would be undefined.

Available Fixes

  1. Check for existence of the variable explicitly:

    function myFunction($a) {
        switch ($a) {
            case 'foo':
                $x = 1;
                break;
    
            case 'bar':
                $x = 2;
                break;
        }
    
        if (isset($x)) { // Make sure it's always set.
            echo $x;
        }
    }
    
  2. Define a default value for the variable:

    function myFunction($a) {
        $x = ''; // Set a default which gets overridden for certain paths.
        switch ($a) {
            case 'foo':
                $x = 1;
                break;
    
            case 'bar':
                $x = 2;
                break;
        }
    
        echo $x;
    }
    
  3. Add a value for the missing path:

    function myFunction($a) {
        switch ($a) {
            case 'foo':
                $x = 1;
                break;
    
            case 'bar':
                $x = 2;
                break;
    
            // We add support for the missing case.
            default:
                $x = '';
                break;
        }
    
        echo $x;
    }
    
Loading history...
218
			}
219
		}
220
		return true;
221
	}
222
223
	function get_sync_queue() {
224
		return $this->sync_queue;
225
	}
226
227
	function get_full_sync_queue() {
228
		return $this->full_sync_queue;
229
	}
230
231
	function get_codec() {
232
		return $this->codec;
233
	}
234
235
	function send_checksum() {
236
		require_once 'class.jetpack-sync-wp-replicastore.php';
237
		$store = new Jetpack_Sync_WP_Replicastore();
238
		do_action( 'jetpack_sync_checksum', $store->checksum_all() );
239
	}
240
241
	function reset_sync_queue() {
242
		$this->sync_queue->reset();
243
	}
244
245
	function set_dequeue_max_bytes( $size ) {
246
		$this->dequeue_max_bytes = $size;
247
	}
248
249
	// in bytes
250
	function set_upload_max_bytes( $max_bytes ) {
251
		$this->upload_max_bytes = $max_bytes;
252
	}
253
254
	// in rows
255
	function set_upload_max_rows( $max_rows ) {
256
		$this->upload_max_rows = $max_rows;
257
	}
258
259
	// in seconds
260
	function set_sync_wait_time( $seconds ) {
261
		$this->sync_wait_time = $seconds;
262
	}
263
264
	function get_sync_wait_time() {
265
		return $this->sync_wait_time;
266
	}
267
268
	// in seconds
269
	function set_sync_wait_threshold( $seconds ) {
270
		$this->sync_wait_threshold = $seconds;
271
	}
272
273
	function get_sync_wait_threshold() {
274
		return $this->sync_wait_threshold;
275
	}
276
277
	// in seconds
278
	function set_max_dequeue_time( $seconds ) {
279
		$this->max_dequeue_time = $seconds;
280
	}
281
282
	function set_defaults() {
283
		$this->sync_queue      = new Jetpack_Sync_Queue( 'sync' );
284
		$this->full_sync_queue = new Jetpack_Sync_Queue( 'full_sync' );
285
		$this->codec           = new Jetpack_Sync_JSON_Deflate_Array_Codec();
286
287
		// saved settings
288
		Jetpack_Sync_Settings::set_importing( null );
289
		$settings = Jetpack_Sync_Settings::get_settings();
290
		$this->set_dequeue_max_bytes( $settings['dequeue_max_bytes'] );
291
		$this->set_upload_max_bytes( $settings['upload_max_bytes'] );
292
		$this->set_upload_max_rows( $settings['upload_max_rows'] );
293
		$this->set_sync_wait_time( $settings['sync_wait_time'] );
294
		$this->set_sync_wait_threshold( $settings['sync_wait_threshold'] );
295
		$this->set_max_dequeue_time( Jetpack_Sync_Defaults::get_max_sync_execution_time() );
296
	}
297
298
	function reset_data() {
299
		$this->reset_sync_queue();
300
301
		foreach ( Jetpack_Sync_Modules::get_modules() as $module ) {
302
			$module->reset_data();
303
		}
304
		
305
		foreach ( array( 'sync', 'full_sync' ) as $queue_name ) {
306
			delete_option( self::NEXT_SYNC_TIME_OPTION_NAME . '_' . $queue_name );
307
		}
308
309
		Jetpack_Sync_Settings::reset_data();
310
	}
311
312
	function uninstall() {
313
		// Lets delete all the other fun stuff like transient and option and the sync queue
314
		$this->reset_data();
315
316
		// delete the full sync status
317
		delete_option( 'jetpack_full_sync_status' );
318
319
		// clear the sync cron.
320
		wp_clear_scheduled_hook( 'jetpack_sync_cron' );
321
	}
322
}
323