PluginProbe
Media Cloud Sync / trunk
Media Cloud Sync vtrunk
1.4.0 1.3.12 1.3.11 1.3.10 trunk 1.0.0 1.0.1 1.0.2 1.0.3 1.1.0 1.1.1 1.2.0 1.2.10 1.2.11 1.2.12 1.2.13 1.2.2 1.2.3 1.2.4 1.2.5 1.2.6 1.2.7 1.2.8 1.2.9 1.3.0 All 34 releases
media-cloud-sync / includes / base / bg-runner.php

bg-runner.php in Media Cloud Sync trunk, at includes/base/bg-runner.php

643 lines 24.2 KB
No matching file
Up and down to move Enter to open Esc to close
Raw Download Zip
1 <?php
2 namespace Dudlewebs\WPMCS;
3
4 defined('ABSPATH') || exit;
5
6 class BGRunner {
7 private static $instance = null;
8 private $meta_key = 'dw_bg_runner_meta';
9 private $transient_key = 'dw_bg_runner_state';
10 private $control_transient_key = 'dw_bg_runner_control';
11 // Unlike $meta_key's state rows, never cleared on idle — see get_type_settings().
12 private $settings_meta_key = 'dw_bg_runner_settings';
13 private $action_hook = 'dw_bg_runner_cron';
14 private $callback_map = []; // type => callback
15 private $tick_supported = []; // type => bool, whether ajax-driven ticking is wired up for this type
16 private $state_cache = []; // type => cached state array for that type
17 private static $lock_duration = 5 * 60; // lock duration (seconds)
18 // How long a tick-supported type can go without a successful pass before cron treats
19 // it as abandoned (e.g. laptop lid closed mid-sync) and steps in as a safety net.
20 private static $ajax_abandoned_threshold = 120;
21
22 private function __construct() {
23 $this->meta_key = WPMCS_TOKEN . '_bg_runner_meta';
24 $this->transient_key = WPMCS_TOKEN . '_bg_runner_state';
25 $this->control_transient_key = WPMCS_TOKEN . '_bg_runner_control';
26 $this->settings_meta_key = Schema::getConstant('BG_RUNNER_SETTINGS_KEY');
27 $this->action_hook = WPMCS_TOKEN . '_bg_runner_cron';
28
29 add_filter('cron_schedules', [$this, 'add_cron_schedules']);
30 add_action($this->action_hook, [$this, 'run_all']);
31
32 // Ensure cron always exists
33 if (!wp_next_scheduled($this->action_hook)) {
34 wp_schedule_event(time(), 'every_minute', $this->action_hook);
35 }
36
37 // Force remove lock
38 add_action('init', [$this, 'force_remove_lock']);
39 }
40
41 public static function instance() {
42 if (!self::$instance) self::$instance = new self();
43 return self::$instance;
44 }
45
46 /**
47 * Register a job type's per-iteration callback.
48 *
49 * @param string $type Job type key.
50 * @param callable $callback Called with ($offset, $total_iterations) once per iteration.
51 * @param bool $supports_tick Whether this type also supports being driven synchronously
52 * via run_all($type, ...) from a REST request (ajax/mixed sync
53 * mode), in addition to the WP-Cron tick.
54 */
55 public function set_callback(string $type, callable $callback, bool $supports_tick = false) {
56 // Empty type would collide with any other empty-type registration on the same rows.
57 if ($type === '') {
58 return;
59 }
60
61 $this->callback_map[$type] = $callback;
62 $this->tick_supported[$type] = $supports_tick;
63 }
64
65 public function start(string $type, int $iterations) {
66 $s = $this->get_type_state($type);
67 $s['status'] = 'running';
68 $s['iterations_total'] = $iterations;
69 $s['iterations_done'] = 0;
70 $s['failed_count'] = 0;
71 $s['last_run'] = time();
72 $this->save_type_state($type, $s);
73
74 // reset control flags
75 $this->save_type_control($type, $this->default_control_state());
76 }
77
78 public function pause(string $type) {
79 $c = $this->get_type_control($type);
80 $c['pause_requested'] = true;
81 $c['time'] = time();
82 $this->save_type_control($type, $c);
83 }
84
85 public function stop(string $type) {
86 $c = $this->get_type_control($type);
87 $c['stop_requested'] = true;
88 $c['time'] = time();
89 $this->save_type_control($type, $c);
90 }
91
92 public function resume(string $type) {
93 $c = $this->get_type_control($type);
94 $c['pause_requested'] = false;
95 $c['stop_requested'] = false;
96 $c['time'] = time();
97 $this->save_type_control($type, $c);
98
99 $s = $this->get_type_state($type);
100 if (in_array($s['status'] ?? 'stopped', ['paused','stopped'], true)) {
101 $s['status'] = 'running';
102 $s['last_run'] = time();
103 $this->save_type_state($type, $s);
104 }
105 }
106
107 public function status(string $type) {
108 $s = $this->get_type_state($type);
109 $c = $this->get_type_control($type);
110
111 // if running but pause/stop requested and lock expired, update state
112 if(
113 (( $s['status'] ?? 'stopped') === 'running' ) &&
114 ( isset($c['time']) && ( $c['time'] > 0 ) && ( ( time() - $c['time'] ) > self::$lock_duration ) ) &&
115 ($c['pause_requested'] === true || $c['stop_requested'] === true)
116 ) {
117 if($c['stop_requested'] === true) {
118 $s = $this->default_type_state();
119 } else {
120 $s['status'] = 'paused';
121 }
122 $this->save_type_state($type, $s);
123
124 // Consume the request this self-heal just acted on.
125 $c['pause_requested'] = false;
126 $c['stop_requested'] = false;
127 $c['time'] = time();
128 $this->save_type_control($type, $c);
129 }
130
131 return $this->format_status($type, $s);
132 }
133
134 public function all_statuses() {
135 $statuses = [];
136 foreach (array_keys($this->callback_map) as $type) {
137 $statuses[$type] = $this->status($type);
138 }
139 return $statuses;
140 }
141
142 private function format_status(string $type, array $s) {
143 $c = $this->get_type_control($type);
144
145 $total = $s['iterations_total'] ?? 0;
146 $done = $s['iterations_done'] ?? 0;
147 $failed_count = $s['failed_count'] ?? 0;
148 $remaining = max(0, $total - $done);
149 $percentage = $total > 0 ? round(($done / $total) * 100, 2) : 0;
150
151 $status = [
152 'total' => $total,
153 'processed' => $done,
154 'failed' => $failed_count,
155 'remaining' => $remaining,
156 'percentage' => $percentage,
157 'status' => $s['status'],
158 'last_run' => $s['last_run'] ?? 0,
159 'pause_requested' => $c['pause_requested'] ?? false,
160 'stop_requested' => $c['stop_requested'] ?? false,
161 'sync_method' => $this->get_sync_method($type),
162 ];
163
164 // Report completed if marked so in state
165 if (!empty($s['completed'])) {
166 $status['percentage'] = 100;
167 $status['remaining'] = 0;
168 $status['status'] = 'completed';
169 }
170
171 // Once idle, delete state/control rows instead of resetting in place — a missing
172 // row and a freshly-defaulted one read back identically.
173 if(
174 !empty($s['completed']) ||
175 (
176 $status['status'] === 'stopped' &&
177 $s['iterations_done'] < $s['iterations_total']
178 )
179 ) {
180 $this->delete_type_state($type);
181 $this->delete_type_control($type);
182 }
183
184 return $status;
185 }
186
187 /**
188 * Process registered job types.
189 *
190 * Called by WP-Cron with no args (processes every running type, up to 50
191 * iterations each). Also called synchronously from a REST request with a
192 * specific $type and a small $max_per_run to drive ajax/mixed sync mode.
193 *
194 * @param string|null $only_type If set, only this type is processed (all others are
195 * left for the next cron tick to own).
196 * @param int $max_per_run Max iterations to process per type in this call.
197 */
198 public function run_all(?string $only_type = null, int $max_per_run = 50) {
199 $max_per_run = max(1, $max_per_run);
200
201 $max_exec = (int) ini_get('max_execution_time');
202 $max_exec = $max_exec !== 0 ? $max_exec : 55;
203 $time_safe = $max_exec * 0.80;
204 $run_start_time = microtime(true);
205
206 // Registered types are the source of truth now that state/control/lock live per type.
207 foreach (array_keys($this->callback_map) as $type) {
208 if ($only_type !== null && $type !== $only_type) continue;
209
210 // shared budget for this whole tick, not per type
211 if ((microtime(true) - $run_start_time) > $time_safe) {
212 break;
213 }
214
215 // refresh state for each type to pick up changes made during processing of other types
216 $s = $this->get_type_state($type, true);
217
218 if (($s['status'] ?? 'stopped') !== 'running') {
219 $c = $this->get_type_control($type);
220
221 if ($c['stop_requested'] ?? false) {
222 $s['status'] = 'stopped';
223 $this->save_type_state($type, $s);
224 $c['stop_requested'] = false;
225 $this->save_type_control($type, $c);
226 }
227 continue;
228 }
229
230 // Ajax mode: the browser drives this type via its own run_all($type, ...) calls,
231 // so cron's untargeted pass yields to it while last_run is still recent. Only
232 // applies to the cron-driven invocation; a type that never opted into ticking is
233 // never skipped here. If last_run goes stale (browser gone, e.g. laptop lid
234 // closed without firing pagehide), cron steps back in as a safety net.
235 if ($only_type === null && !empty($this->tick_supported[$type]) && $this->get_effective_sync_method($type) === 'ajax') {
236 $last_run = $s['last_run'] ?? 0;
237 if ((time() - $last_run) < self::$ajax_abandoned_threshold) {
238 continue;
239 }
240 }
241
242 if ($this->is_locked($s)) continue;
243
244 // Real cross-process lock — is_locked() above is a cheap, non-atomic first
245 // filter; acquire_process_lock() is what actually prevents two processes from
246 // working the same type at once.
247 if (!$this->acquire_process_lock($type)) continue;
248
249 try {
250 // lock, save and process
251 $s = $this->lock($s);
252 $this->save_type_state($type, $s);
253
254 $this->process_iterations($type, $this->callback_map[$type], $run_start_time, $time_safe, $max_per_run);
255 } finally {
256 // Fetch state again to ensure we have the latest and then unlock
257 $latest_s = $this->get_type_state($type, true);
258 $latest_s = $this->unlock($latest_s);
259 $this->save_type_state($type, $latest_s);
260 $this->release_process_lock($type);
261 }
262 }
263 }
264
265 /**
266 * Whether Pro is installed, licensed, and set to ajax/mixed sync mode for this type.
267 * Falls back to 'cron' whenever Pro isn't installed and currently licensed.
268 *
269 * @return string 'ajax'|'cron'|'mixed'
270 */
271 private function get_effective_sync_method(string $type) {
272 return Utils::is_pro_licensed() ? $this->get_sync_method($type) : 'cron';
273 }
274
275 /**
276 * Force the next get_transient() for this key to be a real database read instead of a
277 * possibly-stale value from WordPress's per-request object cache, which otherwise keeps
278 * returning the same cached copy for the life of a request no matter how many times
279 * it's read again.
280 */
281 private function bust_transient_cache(string $key) {
282 wp_cache_delete($key, 'transient');
283 wp_cache_delete('_transient_' . $key, 'options');
284 }
285
286 /**
287 * Cross-process mutual exclusion for a type, via add_option()'s atomic INSERT (backed
288 * by the UNIQUE index on wp_options.option_name) rather than a MySQL-specific locking
289 * function. One row per type so unrelated sync types never block each other. Any
290 * process can reclaim a stale lock by age — see renew_process_lock() for why a
291 * heartbeat is needed rather than just the acquisition timestamp.
292 *
293 * @return bool True if the lock was acquired (fresh or reclaimed from a stale holder).
294 */
295 private function acquire_process_lock(string $type) {
296 $lock_option = $this->meta_key . '_proc_lock_' . $type;
297
298 if (add_option($lock_option, time(), '', false)) {
299 return true;
300 }
301
302 $acquired_at = (int) get_option($lock_option, 0);
303 if ($acquired_at > 0 && (time() - $acquired_at) > self::$lock_duration) {
304 delete_option($lock_option);
305 return add_option($lock_option, time(), '', false);
306 }
307
308 return false;
309 }
310
311 private function release_process_lock(string $type) {
312 delete_option($this->meta_key . '_proc_lock_' . $type);
313 }
314
315 /**
316 * Refresh the lock's timestamp so a batch that's simply slow doesn't look identical to
317 * a dead one to acquire_process_lock()'s staleness check.
318 */
319 private function renew_process_lock(string $type) {
320 update_option($this->meta_key . '_proc_lock_' . $type, time(), false);
321 }
322
323 private function process_iterations(string $type, callable $callback, float $run_start_time, float $time_safe, int $max_per_run = 50) {
324 $s = $this->get_type_state_counts($type, true);
325
326 $max_mem = ini_get('memory_limit') ? $this->return_bytes(ini_get('memory_limit')) : 128 * 1024 * 1024;
327 $memory_safe = $max_mem * 0.80;
328 $iterations_count = 0;
329
330 while ($s['iterations_done'] < $s['iterations_total'] && $iterations_count < $max_per_run) {
331 // limit iterations per run to avoid long blocking
332 $iterations_count++;
333
334 // force refresh so pause/stop requests are seen immediately
335 $latest_c = $this->get_type_control($type);
336
337 // stop if paused or stopped
338 if (!empty($latest_c['pause_requested']) || !empty($latest_c['stop_requested'])) {
339 break;
340 }
341
342 // check memory and time using more precise calls (time budget shared across all types in this tick)
343 if ((memory_get_usage(false) > $memory_safe) || ((microtime(true) - $run_start_time) > $time_safe)) {
344 break;
345 }
346
347 try {
348 $already_done = $s['iterations_done'];
349 $result = call_user_func($callback, $already_done, $s['iterations_total']);
350 if ($result !== true) {
351 $s['failed_count']++;
352 }
353 } catch (\Exception $e) {
354 $s['failed_count']++;
355 }
356
357 $s['iterations_done']++;
358
359 // Write progress every item, not just at the end of the batch, so a slow item
360 // can't silently lose the batch's progress if it gets killed mid-way.
361 $this->update_type_state_counts($type, $s, true, false);
362 $this->renew_process_lock($type);
363
364 if($iterations_count % 10 === 0) {
365 gc_collect_cycles();
366 }
367 }
368
369 // Get latest state again
370 $latest_s = $this->get_type_state($type, true);
371 $latest_c = $this->get_type_control($type);
372 $is_completed = ($latest_s['iterations_done'] ?? 0) >= ($latest_s['iterations_total'] ?? 0);
373
374 // Update state if completed, paused or stopped
375 if ($is_completed || ($latest_c['stop_requested'] ?? false) || ($latest_c['pause_requested'] ?? false)) {
376 $latest_s['status'] = $is_completed || ($latest_c['stop_requested'] ?? false) ? 'stopped' : 'paused';
377 $latest_s['completed'] = $is_completed;
378 $this->save_type_state($type, $latest_s);
379
380 $latest_c['pause_requested'] = false;
381 $latest_c['stop_requested'] = false;
382 $latest_c['time'] = time();
383 $this->save_type_control($type, $latest_c);
384 }
385
386 gc_collect_cycles();
387 }
388
389 private function lock(array $s) {
390 $s['lock_until'] = time() + self::$lock_duration;
391 return $s;
392 }
393
394 private function unlock(array $s) {
395 $s['lock_until'] = 0;
396 $s['last_run'] = time();
397 return $s;
398 }
399
400 private function is_locked(array $s) {
401 return ($s['lock_until'] ?? 0) && time() < $s['lock_until'];
402 }
403
404 private function return_bytes($val) {
405 $val = trim($val);
406 // "-1"/blank mean no limit; treating it as a numeric byte count made the
407 // memory-budget check in process_iterations() permanently true.
408 if ($val === '' || $val === '-1') {
409 return PHP_INT_MAX;
410 }
411 $last = strtolower($val[strlen($val)-1] ?? '');
412 $num = (int) $val;
413 switch ($last) {
414 case 'g': $num *= 1024 * 1024 * 1024; break;
415 case 'm': $num *= 1024 * 1024; break;
416 case 'k': $num *= 1024; break;
417 }
418 return $num;
419 }
420
421 public function add_cron_schedules($schedules) {
422 $schedules['every_minute'] = [
423 'interval' => 60,
424 'display' => 'Every Minute'
425 ];
426 return $schedules;
427 }
428
429
430 /**
431 * Get only the count-related fields of the state for a given type.
432 *
433 * @param string $type The job type.
434 * @param bool $force Reload state from transient/option instead of cache.
435 *
436 * @return array {
437 * @type int $iterations_total
438 * @type int $iterations_done
439 * @type int $failed_count
440 * }
441 */
442 private function get_type_state_counts(string $type, bool $force = false) {
443 $s = $this->get_type_state($type, $force);
444
445 return [
446 'iterations_total' => (int) ($s['iterations_total'] ?? 0),
447 'iterations_done' => (int) ($s['iterations_done'] ?? 0),
448 'failed_count' => (int) ($s['failed_count'] ?? 0),
449 ];
450 }
451
452
453 /**
454 * Update only the count-related fields for a given type.
455 *
456 * @param string $type The job type.
457 * @param array $counts {
458 * @type int $iterations_total
459 * @type int $iterations_done
460 * @type int $failed_count
461 * }
462 * @param bool $update_transient Whether to update transient.
463 * @param bool $update_options Whether to update options.
464 */
465 private function update_type_state_counts(string $type, array $counts, bool $update_transient = true, bool $update_options = true) {
466 $s = $this->get_type_state($type, true);
467
468 // update only the count fields
469 if (isset($counts['iterations_total'])) {
470 $s['iterations_total'] = (int) $counts['iterations_total'];
471 }
472 if (isset($counts['iterations_done'])) {
473 $s['iterations_done'] = (int) $counts['iterations_done'];
474 }
475 if (isset($counts['failed_count'])) {
476 $s['failed_count'] = (int) $counts['failed_count'];
477 }
478
479 $this->save_type_state($type, $s, $update_transient, $update_options);
480 }
481
482
483 /**
484 * Get the state for one job type. Each type has its own transient/option row rather
485 * than one row holding every type's state, so concurrent processing of two different
486 * types never races on the same shared row.
487 *
488 * @param string $type The job type.
489 * @param bool $force Reload the state from the transient or option.
490 *
491 * @return array The state for this type.
492 */
493 private function get_type_state(string $type, bool $force = false) {
494 // if not forcing and cache already exists, return cached
495 if (!$force && isset($this->state_cache[$type])) {
496 return $this->state_cache[$type];
497 }
498
499 $transient_key = $this->transient_key . '_' . $type;
500
501 if ($force) {
502 $this->bust_transient_cache($transient_key);
503 }
504
505 // try transient first
506 $s = get_transient($transient_key);
507
508 if ($s !== false) {
509 if ($force) {
510 // update cache with the fresh transient
511 $this->state_cache[$type] = $s;
512 }
513 return $s;
514 }
515
516 // fallback to option if transient missing (e.g. evicted from a persistent object cache)
517 $option_key = $this->meta_key . '_' . $type;
518 $s = get_option($option_key, null);
519 if ($s === null) {
520 // Genuinely idle/never started — return the default without persisting a row,
521 // since run_all() now touches every registered type on every cron tick.
522 $s = $this->default_type_state();
523 $this->state_cache[$type] = $s;
524 return $s;
525 }
526 $this->state_cache[$type] = $s;
527 set_transient($transient_key, $s, WEEK_IN_SECONDS);
528
529 return $s;
530 }
531
532 private function save_type_state(string $type, array $s, bool $update_transient = true, bool $update_options = true) {
533 // Only write if changed to reduce option churn
534 if (!isset($this->state_cache[$type]) || $this->state_cache[$type] !== $s) {
535 $this->state_cache[$type] = $s;
536 if ($update_transient) set_transient($this->transient_key . '_' . $type, $s, WEEK_IN_SECONDS);
537 if ($update_options) update_option($this->meta_key . '_' . $type, $s, false);
538 }
539 }
540
541 private function delete_type_state(string $type) {
542 delete_transient($this->transient_key . '_' . $type);
543 delete_option($this->meta_key . '_' . $type);
544 unset($this->state_cache[$type]);
545 }
546
547 private function default_type_state() {
548 return [
549 'lock_until' => 0,
550 'status' => 'stopped',
551 'iterations_total' => 0,
552 'iterations_done' => 0,
553 'failed_count' => 0,
554 'completed' => false,
555 'last_run' => 0,
556 ];
557 }
558
559 /**
560 * Control (pause/stop) always busts the cache before reading — every caller needs it
561 * live, unlike state there's no non-forced fast path here.
562 *
563 * @return array
564 */
565 private function get_type_control(string $type) {
566 $key = $this->control_transient_key . '_' . $type;
567 $this->bust_transient_cache($key);
568
569 $c = get_transient($key);
570 return $c !== false ? $c : $this->default_control_state();
571 }
572
573 private function save_type_control(string $type, array $c) {
574 set_transient($this->control_transient_key . '_' . $type, $c, WEEK_IN_SECONDS);
575 }
576
577 private function delete_type_control(string $type) {
578 delete_transient($this->control_transient_key . '_' . $type);
579 }
580
581 private function default_control_state() {
582 return [
583 'pause_requested' => false,
584 'stop_requested' => false,
585 'time' => time(),
586 ];
587 }
588
589 // One option row for every type, keyed by $type — not one option per type.
590 private function get_type_settings(string $type) {
591 return Utils::get_option($type, [], $this->settings_meta_key);
592 }
593
594 private function save_type_settings(string $type, array $settings) {
595 Utils::update_option($type, $settings, $this->settings_meta_key);
596 }
597
598 // Raw stored preference, not license-gated — see get_effective_sync_method() for that.
599 public function get_sync_method(string $type) {
600 $settings = $this->get_type_settings($type);
601 return $settings['sync_method'] ?? 'cron';
602 }
603
604 public function set_sync_method(string $type, string $method) {
605 if (!in_array($method, ['ajax', 'cron', 'mixed'], true)) {
606 return false;
607 }
608 $settings = $this->get_type_settings($type);
609 $settings['sync_method'] = $method;
610 $this->save_type_settings($type, $settings);
611 return true;
612 }
613
614 /**
615 * Forces removal of the bg runner lock. This is a debug utility and should not be used in production.
616 * The lock is removed when the query string parameter 'force_reset_sync' is set to '1'.
617 * The purpose of this function is to allow for easy reset of the bg runner lock in debug environments.
618 * It is not intended for use in production and can potentially cause issues with the bg runner's operation.
619 */
620 public function force_remove_lock() {
621 if (isset($_GET['force_reset_sync']) && $_GET['force_reset_sync'] == '1') {
622 if (! current_user_can('manage_options')) {
623 return;
624 }
625
626 // Legacy pre-per-type keys, in case any still linger.
627 delete_transient($this->transient_key);
628 delete_option($this->meta_key);
629 delete_transient($this->control_transient_key);
630
631 foreach (array_keys($this->callback_map) as $type) {
632 $this->delete_type_state($type);
633 $this->delete_type_control($type);
634 delete_option($this->meta_key . '_proc_lock_' . $type);
635 }
636
637 if (! defined('DOING_AJAX')) {
638 wp_die('Locks removed successfully.');
639 }
640 }
641 }
642 }
643