PluginProbe
Media Cloud Sync / 1.4.1
Media Cloud Sync v1.4.1
1.4.1 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 All 35 releases
media-cloud-sync / includes / base / bg-runner.php

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

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