← All changes | JobsRunner.php +158 -13 1.10.9 → 1.10.19 View file →

		@@ -8,9 +8,17 @@
8	8	defined('ABSPATH') || exit;
9	9
10	10	class JobsRunner
11	11	{
	12	+ private const ASYNC_ACTION = 'sync_basalam_run_jobs_async';
	13	+ private const ASYNC_DISPATCH_LOCK_TRANSIENT = 'sync_basalam_jobs_runner_async_dispatch_lock';
	14	+ // Keep the dispatch lease longer than a normal async batch. Without this,
	15	+ // every frontend request can boot another full WordPress AJAX worker while
	16	+ // a large product queue is active.
	17	+ private const ASYNC_DISPATCH_LOCK_SECONDS = 25;
	18	+ private const ASYNC_TIME_LIMIT_SECONDS = 20;
12	19	private const GLOBAL_RUNNER_LAST_RUN_OPTION = 'sync_basalam_jobs_runner_last_run';
	20	+ private const STALE_PROCESSING_TIMEOUT_SECONDS = 120;
13	21
14	22	private $jobExecutor;
15	23	private $jobManager;
16	24	private $discountScheduler;
		@@ -21,9 +29,15 @@
21	29	$jobExecutor,
22	30	$discountScheduler,
23	31	$CheckHttpBlockService
24	32	) {
25		- add_action('init', [$this, 'checkAndRunJobs']);
	33	+ add_action('wp_ajax_' . self::ASYNC_ACTION, [$this, 'handleAsyncRequest']);
	34	+ add_action('wp_ajax_nopriv_' . self::ASYNC_ACTION, [$this, 'handleAsyncRequest']);
	35	+ // Probe and dispatch after the response path so normal storefront
	36	+ // requests never pay for the queue query or loopback HTTP request.
	37	+ add_action('shutdown', [$this, 'maybeDispatchAsyncRequest'], PHP_INT_MAX);
	38	+ add_action('sync_basalam_job_created', [$this, 'maybeDispatchAsyncRequest'], 10, 0);
	39	+
26	40	$this->jobManager = $jobManager;
27	41	$this->jobExecutor = $jobExecutor;
28	42	$this->discountScheduler = $discountScheduler;
29	43	$this->CheckHttpBlockService = $CheckHttpBlockService;
		@@ -28,41 +42,127 @@
28	42	$this->discountScheduler = $discountScheduler;
29	43	$this->CheckHttpBlockService = $CheckHttpBlockService;
30	44	}
31	45
32		- public function checkAndRunJobs(): void
	46	+ public function maybeDispatchAsyncRequest(): void
33	47	{
	48	+ if ($this->isCurrentAsyncRequest()) return;
34	49	if ($this->CheckHttpBlockService->SyncBasalamHttpBlock()) return;
35		- if (!$this->jobExecutor->acquireGlobalJobsLock(0)) return;
	50	+ if (get_transient(self::ASYNC_DISPATCH_LOCK_TRANSIENT)) return;
36	51
	52	+ // Reserve the dispatch lease before probing the queue. This keeps
	53	+ // concurrent shutdown callbacks from all running the queue query and
	54	+ // dispatching duplicate async workers.
	55	+ set_transient(
	56	+ self::ASYNC_DISPATCH_LOCK_TRANSIENT,
	57	+ 1,
	58	+ self::ASYNC_DISPATCH_LOCK_SECONDS
	59	+ );
	60	+
	61	+ if (!$this->jobManager->hasPendingOrStaleProcessingJobs(self::STALE_PROCESSING_TIMEOUT_SECONDS)) {
	62	+ return;
	63	+ }
	64	+
	65	+ $this->dispatchAsyncRequest();
	66	+ }
	67	+
	68	+ public function handleAsyncRequest(): void
	69	+ {
	70	+ if (!check_ajax_referer(self::ASYNC_ACTION, 'nonce', false)) {
	71	+ wp_send_json_error(['message' => 'Invalid async jobs runner nonce.'], 403);
	72	+ }
	73	+
	74	+ if (function_exists('ignore_user_abort')) {
	75	+ ignore_user_abort(true);
	76	+ }
	77	+
	78	+ if (function_exists('session_write_close')) {
	79	+ session_write_close();
	80	+ }
	81	+
	82	+ $processed = $this->runAsyncBatch();
	83	+
	84	+ wp_send_json_success(['processed' => $processed]);
	85	+ }
	86	+
	87	+ public function checkAndRunJobs(): bool
	88	+ {
	89	+ if ($this->CheckHttpBlockService->SyncBasalamHttpBlock()) return false;
	90	+ if (!$this->jobExecutor->acquireGlobalJobsLock(0)) return false;
	91	+
37	92	try {
38		- $this->runEligibleJobs();
	93	+ return $this->runEligibleJobs();
39	94	} finally {
40	95	$this->jobExecutor->releaseGlobalJobsLock();
41	96	}
42	97	}
43	98
44		- private function runEligibleJobs(): void
	99	+ private function runAsyncBatch(): int
45	100	{
46		- $this->jobManager->ConvertStaleProcessingJobs(120);
	101	+ if ($this->CheckHttpBlockService->SyncBasalamHttpBlock()) return 0;
	102	+
	103	+ // Hold the advisory lock for the whole batch, including rate-limit
	104	+ // waits. Previously it was released after every job, so duplicate
	105	+ // async requests could pile up and sleep in parallel until the next
	106	+ // job became eligible, exhausting the site's PHP workers.
	107	+ if (!$this->jobExecutor->acquireGlobalJobsLock(0)) return 0;
	108	+
	109	+ $processed = 0;
	110	+ $deadline = microtime(true) + (float) apply_filters(
	111	+ 'sync_basalam_jobs_runner_async_time_limit',
	112	+ self::ASYNC_TIME_LIMIT_SECONDS
	113	+ );
	114	+
	115	+ try {
	116	+ while (microtime(true) < $deadline) {
	117	+ if (!$this->jobManager->hasPendingOrStaleProcessingJobs(self::STALE_PROCESSING_TIMEOUT_SECONDS)) {
	118	+ break;
	119	+ }
	120	+
	121	+ $ranJob = $this->runEligibleJobs();
	122	+
	123	+ if ($ranJob) {
	124	+ $processed++;
	125	+ }
	126	+
	127	+ $delay = $this->secondsUntilNextAllowedRun();
	128	+ if ($delay <= 0.0) {
	129	+ if (!$ranJob) break;
	130	+ continue;
	131	+ }
	132	+
	133	+ if ((microtime(true) + $delay) >= $deadline) {
	134	+ break;
	135	+ }
	136	+
	137	+ usleep((int) ($delay * 1000000));
	138	+ }
	139	+ } finally {
	140	+ $this->jobExecutor->releaseGlobalJobsLock();
	141	+ }
	142	+
	143	+ return $processed;
	144	+ }
	145	+
	146	+ private function runEligibleJobs(): bool
	147	+ {
	148	+ $this->jobManager->ConvertStaleProcessingJobs(self::STALE_PROCESSING_TIMEOUT_SECONDS);
47	149	$this->discountScheduler->process();
48	150
49	151	$circuitBreaker = new CircuitBreaker();
50	152	if ($circuitBreaker->getState() === CircuitBreaker::STATE_OPEN) {
51		- return;
	153	+ return false;
52	154	}
53	155
54	156	if ($this->jobManager->hasAnyProcessingJob()) {
55		- return;
	157	+ return false;
56	158	}
57	159
58		- $tasksPerMinute = max(1, intval(Settings::getEffectiveTasksPerMinute()));
59		- $thresholdSeconds = 60.0 / $tasksPerMinute;
60	160	$lastRun = floatval(get_option(self::GLOBAL_RUNNER_LAST_RUN_OPTION, 0));
61	161	$now = microtime(true);
62	162
63		- if (($now - $lastRun) < $thresholdSeconds) {
64		- return;
	163	+ if (($now - $lastRun) < $this->getRunThresholdSeconds()) {
	164	+ return false;
65	165	}
66	166
67	167	$sortedJobTypes = $this->jobExecutor->getSortedJobTypes();
68	168
		@@ -85,10 +185,12 @@
85	185	['id' => $job->id]
86	186	);
87	187
88	188	$this->executeJob($job);
89		- break;
	189	+ return true;
90	190	}
	191	+
	192	+ return false;
91	193	}
92	194
93	195	private function executeJob(object $job): void
94	196	{
		@@ -93,6 +195,49 @@
93	195	private function executeJob(object $job): void
94	196	{
95	197	$jobType = $job->job_type;
96	198	$this->jobExecutor->execute($jobType, $job);
	199	+ }
	200	+
	201	+ private function dispatchAsyncRequest(): void
	202	+ {
	203	+ $url = add_query_arg('action', self::ASYNC_ACTION, admin_url('admin-ajax.php'));
	204	+
	205	+ wp_remote_post(esc_url_raw($url), [
	206	+ 'timeout' => 0.01,
	207	+ 'blocking' => false,
	208	+ 'body' => [
	209	+ 'action' => self::ASYNC_ACTION,
	210	+ 'nonce' => wp_create_nonce(self::ASYNC_ACTION),
	211	+ ],
	212	+ 'cookies' => $_COOKIE,
	213	+ 'sslverify' => apply_filters('https_local_ssl_verify', false),
	214	+ 'headers' => [
	215	+ 'X-WP-Async-Request' => self::ASYNC_ACTION,
	216	+ ],
	217	+ ]);
	218	+ }
	219	+
	220	+ private function isCurrentAsyncRequest(): bool
	221	+ {
	222	+ if (!wp_doing_ajax()) return false;
	223	+
	224	+ $action = isset($_REQUEST['action']) ? sanitize_key(wp_unslash($_REQUEST['action'])) : '';
	225	+
	226	+ return $action === self::ASYNC_ACTION;
	227	+ }
	228	+
	229	+ private function secondsUntilNextAllowedRun(): float
	230	+ {
	231	+ $lastRun = floatval(get_option(self::GLOBAL_RUNNER_LAST_RUN_OPTION, 0));
	232	+ $elapsed = microtime(true) - $lastRun;
	233	+
	234	+ return max(0.0, $this->getRunThresholdSeconds() - $elapsed);
	235	+ }
	236	+
	237	+ private function getRunThresholdSeconds(): float
	238	+ {
	239	+ $tasksPerMinute = max(1, intval(Settings::getEffectiveTasksPerMinute()));
	240	+
	241	+ return 60.0 / $tasksPerMinute;
97	242	}
98	243	}