← All changes | JobsRunner.php +257 -14 1.10.3 → 1.10.29 View file →

		@@ -2,67 +2,261 @@
2	2
3	3	namespace SyncBasalam;
4	4
5	5	use SyncBasalam\Admin\Settings;
	6	+use SyncBasalam\Registrar\AdminRegistrar;
6	7	use SyncBasalam\Services\Api\CircuitBreaker;
	8	+use SyncBasalam\Services\AsyncWorkerHealth;
7	9
8	10	defined('ABSPATH') || exit;
9	11
10	12	class JobsRunner
11	13	{
	14	+ private const ASYNC_ACTION = 'sync_basalam_run_jobs_async';
	15	+ private const ASYNC_DISPATCH_LOCK_TRANSIENT = 'sync_basalam_jobs_runner_async_dispatch_lock';
	16	+ private const ASYNC_IDLE_PROBE_TRANSIENT = 'sync_basalam_jobs_runner_idle_probe_lock';
	17	+ // Keep the dispatch lease longer than a normal async batch. Without this,
	18	+ // every frontend request can boot another full WordPress AJAX worker while
	19	+ // a large product queue is active.
	20	+ private const ASYNC_DISPATCH_LOCK_SECONDS = 25;
	21	+ private const ASYNC_IDLE_PROBE_SECONDS = 5;
	22	+ private const ASYNC_TIME_LIMIT_SECONDS = 20;
12	23	private const GLOBAL_RUNNER_LAST_RUN_OPTION = 'sync_basalam_jobs_runner_last_run';
	24	+ private const STALE_PROCESSING_TIMEOUT_SECONDS = 120;
13	25
14	26	private $jobExecutor;
15	27	private $jobManager;
16	28	private $discountScheduler;
17	29	private $CheckHttpBlockService;
	30	+ private $workerHealth;
	31	+ private bool $isAsyncBatch = false;
18	32
19	33	public function __construct(
20	34	$jobManager,
21	35	$jobExecutor,
22	36	$discountScheduler,
23		- $CheckHttpBlockService
	37	+ $CheckHttpBlockService,
	38	+ ?AsyncWorkerHealth $workerHealth = null
24	39	) {
25		- add_action('init', [$this, 'checkAndRunJobs']);
	40	+ add_action('wp_ajax_' . self::ASYNC_ACTION, [$this, 'handleAsyncRequest']);
	41	+ add_action('wp_ajax_nopriv_' . self::ASYNC_ACTION, [$this, 'handleAsyncRequest']);
	42	+ // Run before later shutdown callbacks can leave an unread mysqli result.
	43	+ add_action('shutdown', [$this, 'maybeDispatchAsyncRequest'], 2);
	44	+ add_action('sync_basalam_job_created', [$this, 'maybeDispatchAsyncRequest'], 10, 0);
	45	+ add_action('admin_notices', [$this, 'renderWorkerHealthNotice']);
	46	+
26	47	$this->jobManager = $jobManager;
27	48	$this->jobExecutor = $jobExecutor;
28	49	$this->discountScheduler = $discountScheduler;
29	50	$this->CheckHttpBlockService = $CheckHttpBlockService;
	51	+ $this->workerHealth = $workerHealth ?? new AsyncWorkerHealth();
30	52	}
31	53
32		- public function checkAndRunJobs(): void
	54	+ public function renderWorkerHealthNotice(): void
33	55	{
	56	+ if (!current_user_can('manage_woocommerce') && !current_user_can('manage_options')) return;
	57	+ if (!AdminRegistrar::isWoosalamRelatedAdminScreen()) return;
	58	+
	59	+ // A live worker may legitimately take over two minutes. Its database
	60	+ // lock, rather than elapsed time alone, determines whether it was lost.
	61	+ if ($this->jobExecutor->acquireGlobalJobsLock(0)) {
	62	+ try {
	63	+ $this->detectInterruptedWorker();
	64	+ } finally {
	65	+ $this->jobExecutor->releaseGlobalJobsLock();
	66	+ }
	67	+ }
	68	+
	69	+ $notice = $this->workerHealth->getNotice();
	70	+ if (!$notice) return;
	71	+ require __DIR__ . '/templates/notifications/AsyncWorkerAlert.php';
	72	+ }
	73	+
	74	+ private function detectInterruptedWorker(): void
	75	+ {
	76	+ $job = $this->jobManager->getStaleProcessingJob(self::STALE_PROCESSING_TIMEOUT_SECONDS);
	77	+ if ($job) $this->workerHealth->recordInterruptedJob($job);
	78	+ }
	79	+
	80	+ public function maybeDispatchAsyncRequest(): void
	81	+ {
	82	+ if ($this->isCurrentAsyncRequest()) return;
	83	+
	84	+ if ($this->isShutdownCallback()) {
	85	+ $this->repairDatabaseConnection();
	86	+ }
	87	+
34	88	if ($this->CheckHttpBlockService->SyncBasalamHttpBlock()) return;
35		- if (!$this->jobExecutor->acquireGlobalJobsLock(0)) return;
	89	+ $isNewJob = function_exists('current_filter') && current_filter() === 'sync_basalam_job_created';
	90	+ if (!$isNewJob && $this->hasActiveAsyncLease(self::ASYNC_IDLE_PROBE_TRANSIENT)) return;
	91	+ if ($this->hasActiveAsyncLease(self::ASYNC_DISPATCH_LOCK_TRANSIENT)) return;
36	92
	93	+ if (!$this->jobManager->hasPendingOrStaleProcessingJobs(self::STALE_PROCESSING_TIMEOUT_SECONDS)) {
	94	+ if (!$isNewJob) {
	95	+ set_transient(
	96	+ self::ASYNC_IDLE_PROBE_TRANSIENT,
	97	+ time() + self::ASYNC_IDLE_PROBE_SECONDS,
	98	+ self::ASYNC_IDLE_PROBE_SECONDS
	99	+ );
	100	+ }
	101	+ return;
	102	+ }
	103	+
	104	+ // An empty-queue probe must not block a job created moments later.
	105	+ // The global database lock still prevents duplicate workers from
	106	+ // processing the same queue when requests race here.
	107	+ if ($this->hasActiveAsyncLease(self::ASYNC_DISPATCH_LOCK_TRANSIENT)) return;
	108	+ set_transient(
	109	+ self::ASYNC_DISPATCH_LOCK_TRANSIENT,
	110	+ time() + self::ASYNC_DISPATCH_LOCK_SECONDS,
	111	+ self::ASYNC_DISPATCH_LOCK_SECONDS
	112	+ );
	113	+
	114	+ $this->dispatchAsyncRequest();
	115	+ }
	116	+
	117	+ private function hasActiveAsyncLease(string $transient): bool
	118	+ {
	119	+ $lease = get_transient($transient);
	120	+ if (!$lease) return false;
	121	+
	122	+ // Keep the deadline in the value too: concurrent transient expiry can
	123	+ // remove the separate timeout option while a new value is being written.
	124	+ if (is_numeric($lease) && (int) $lease > 1) {
	125	+ return (int) $lease > time();
	126	+ }
	127	+
	128	+ // Older versions stored 1. An orphaned database-backed lease must not
	129	+ // stop the queue forever. External object caches enforce their own TTL.
	130	+ if (!wp_using_ext_object_cache() && !wp_installing()) {
	131	+ $expiresAt = (int) get_option('_transient_timeout_' . $transient, 0);
	132	+ return $expiresAt > time();
	133	+ }
	134	+
	135	+ return true;
	136	+ }
	137	+
	138	+ private function isShutdownCallback(): bool
	139	+ {
	140	+ return function_exists('current_filter') && current_filter() === 'shutdown';
	141	+ }
	142	+
	143	+ private function repairDatabaseConnection(): void
	144	+ {
	145	+ global $wpdb;
	146	+
	147	+ if (!isset($wpdb) || !is_object($wpdb)) return;
	148	+
	149	+ if (method_exists($wpdb, 'flush')) {
	150	+ $wpdb->flush();
	151	+ }
	152	+
	153	+ if (method_exists($wpdb, 'check_connection')) {
	154	+ $wpdb->check_connection(false);
	155	+ }
	156	+ }
	157	+
	158	+ public function handleAsyncRequest(): void
	159	+ {
	160	+ if (!check_ajax_referer(self::ASYNC_ACTION, 'nonce', false)) {
	161	+ wp_send_json_error(['message' => 'Invalid async jobs runner nonce.'], 403);
	162	+ }
	163	+
	164	+ if (function_exists('ignore_user_abort')) {
	165	+ ignore_user_abort(true);
	166	+ }
	167	+
	168	+ if (function_exists('session_write_close')) {
	169	+ session_write_close();
	170	+ }
	171	+
	172	+ $processed = $this->runAsyncBatch();
	173	+
	174	+ wp_send_json_success(['processed' => $processed]);
	175	+ }
	176	+
	177	+ public function checkAndRunJobs(): bool
	178	+ {
	179	+ if ($this->CheckHttpBlockService->SyncBasalamHttpBlock()) return false;
	180	+ if (!$this->jobExecutor->acquireGlobalJobsLock(0)) return false;
	181	+
37	182	try {
38		- $this->runEligibleJobs();
	183	+ return $this->runEligibleJobs();
39	184	} finally {
40	185	$this->jobExecutor->releaseGlobalJobsLock();
41	186	}
42	187	}
43	188
44		- private function runEligibleJobs(): void
	189	+ private function runAsyncBatch(): int
45	190	{
46		- $this->jobManager->ConvertStaleProcessingJobs(120);
	191	+ if ($this->CheckHttpBlockService->SyncBasalamHttpBlock()) return 0;
	192	+
	193	+ // Hold the advisory lock for the whole batch, including rate-limit
	194	+ // waits. Previously it was released after every job, so duplicate
	195	+ // async requests could pile up and sleep in parallel until the next
	196	+ // job became eligible, exhausting the site's PHP workers.
	197	+ if (!$this->jobExecutor->acquireGlobalJobsLock(0)) return 0;
	198	+
	199	+ $this->isAsyncBatch = true;
	200	+ $processed = 0;
	201	+ $deadline = microtime(true) + (float) apply_filters(
	202	+ 'sync_basalam_jobs_runner_async_time_limit',
	203	+ self::ASYNC_TIME_LIMIT_SECONDS
	204	+ );
	205	+
	206	+ try {
	207	+ $this->workerHealth->observeAsyncEnvironment();
	208	+ while (microtime(true) < $deadline) {
	209	+ if (!$this->jobManager->hasPendingOrStaleProcessingJobs(self::STALE_PROCESSING_TIMEOUT_SECONDS)) {
	210	+ break;
	211	+ }
	212	+
	213	+ $ranJob = $this->runEligibleJobs();
	214	+
	215	+ if ($ranJob) {
	216	+ $processed++;
	217	+ }
	218	+
	219	+ $delay = $this->secondsUntilNextAllowedRun();
	220	+ if ($delay <= 0.0) {
	221	+ if (!$ranJob) break;
	222	+ continue;
	223	+ }
	224	+
	225	+ if ((microtime(true) + $delay) >= $deadline) {
	226	+ break;
	227	+ }
	228	+
	229	+ usleep((int) ($delay * 1000000));
	230	+ }
	231	+ } finally {
	232	+ $this->isAsyncBatch = false;
	233	+ $this->jobExecutor->releaseGlobalJobsLock();
	234	+ }
	235	+
	236	+ return $processed;
	237	+ }
	238	+
	239	+ private function runEligibleJobs(): bool
	240	+ {
	241	+ $this->detectInterruptedWorker();
	242	+ $this->jobManager->ConvertStaleProcessingJobs(self::STALE_PROCESSING_TIMEOUT_SECONDS);
47	243	$this->discountScheduler->process();
48	244
49	245	$circuitBreaker = new CircuitBreaker();
50	246	if ($circuitBreaker->getState() === CircuitBreaker::STATE_OPEN) {
51		- return;
	247	+ return false;
52	248	}
53	249
54	250	if ($this->jobManager->hasAnyProcessingJob()) {
55		- return;
	251	+ return false;
56	252	}
57	253
58		- $tasksPerMinute = max(1, intval(Settings::getEffectiveTasksPerMinute()));
59		- $thresholdSeconds = 60.0 / $tasksPerMinute;
60	254	$lastRun = floatval(get_option(self::GLOBAL_RUNNER_LAST_RUN_OPTION, 0));
61	255	$now = microtime(true);
62	256
63		- if (($now - $lastRun) < $thresholdSeconds) {
64		- return;
	257	+ if (($now - $lastRun) < $this->getRunThresholdSeconds()) {
	258	+ return false;
65	259	}
66	260
67	261	$sortedJobTypes = $this->jobExecutor->getSortedJobTypes();
68	262
		@@ -85,14 +279,63 @@
85	279	['id' => $job->id]
86	280	);
87	281
88	282	$this->executeJob($job);
89		- break;
	283	+ return true;
90	284	}
	285	+
	286	+ return false;
91	287	}
92	288
93	289	private function executeJob(object $job): void
94	290	{
95	291	$jobType = $job->job_type;
	292	+ $started = microtime(true);
96	293	$this->jobExecutor->execute($jobType, $job);
	294	+ if ($this->isAsyncBatch) {
	295	+ $this->workerHealth->recordJobReturned($job, microtime(true) - $started);
	296	+ }
	297	+ }
	298	+
	299	+ private function dispatchAsyncRequest(): void
	300	+ {
	301	+ $url = add_query_arg('action', self::ASYNC_ACTION, admin_url('admin-ajax.php'));
	302	+
	303	+ wp_remote_post(esc_url_raw($url), [
	304	+ 'timeout' => 0.01,
	305	+ 'blocking' => false,
	306	+ 'body' => [
	307	+ 'action' => self::ASYNC_ACTION,
	308	+ 'nonce' => wp_create_nonce(self::ASYNC_ACTION),
	309	+ ],
	310	+ 'cookies' => $_COOKIE,
	311	+ 'sslverify' => apply_filters('https_local_ssl_verify', false),
	312	+ 'headers' => [
	313	+ 'X-WP-Async-Request' => self::ASYNC_ACTION,
	314	+ ],
	315	+ ]);
	316	+ }
	317	+
	318	+ private function isCurrentAsyncRequest(): bool
	319	+ {
	320	+ if (!wp_doing_ajax()) return false;
	321	+
	322	+ $action = isset($_REQUEST['action']) ? sanitize_key(wp_unslash($_REQUEST['action'])) : '';
	323	+
	324	+ return $action === self::ASYNC_ACTION;
	325	+ }
	326	+
	327	+ private function secondsUntilNextAllowedRun(): float
	328	+ {
	329	+ $lastRun = floatval(get_option(self::GLOBAL_RUNNER_LAST_RUN_OPTION, 0));
	330	+ $elapsed = microtime(true) - $lastRun;
	331	+
	332	+ return max(0.0, $this->getRunThresholdSeconds() - $elapsed);
	333	+ }
	334	+
	335	+ private function getRunThresholdSeconds(): float
	336	+ {
	337	+ $tasksPerMinute = max(1, intval(Settings::getEffectiveTasksPerMinute()));
	338	+
	339	+ return 60.0 / $tasksPerMinute;
97	340	}
98	341	}