← All changes | JobsRunner.php +191 -35 1.8.5 → 1.10.20 View file →

		@@ -8,8 +8,18 @@
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;
	19	+ private const GLOBAL_RUNNER_LAST_RUN_OPTION = 'sync_basalam_jobs_runner_last_run';
	20	+ private const STALE_PROCESSING_TIMEOUT_SECONDS = 120;
	21	+
12	22	private $jobExecutor;
13	23	private $jobManager;
14	24	private $discountScheduler;
15	25	private $CheckHttpBlockService;
		@@ -18,11 +28,16 @@
18	28	$jobManager,
19	29	$jobExecutor,
20	30	$discountScheduler,
21	31	$CheckHttpBlockService
22		- )
23		- {
24		- add_action('init', [$this, 'checkAndRunJobs']);
	32	+ ) {
	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	+
25	40	$this->jobManager = $jobManager;
26	41	$this->jobExecutor = $jobExecutor;
27	42	$this->discountScheduler = $discountScheduler;
28	43	$this->CheckHttpBlockService = $CheckHttpBlockService;
		@@ -27,57 +42,155 @@
27	42	$this->discountScheduler = $discountScheduler;
28	43	$this->CheckHttpBlockService = $CheckHttpBlockService;
29	44	}
30	45
31		- public function checkAndRunJobs(): void
	46	+ public function maybeDispatchAsyncRequest(): void
32	47	{
33		- if($this->CheckHttpBlockService->SyncBasalamHttpBlock()) return;
34		- $this->jobManager->ConvertStaleProcessingJobs(120);
	48	+ if ($this->isCurrentAsyncRequest()) return;
	49	+ if ($this->CheckHttpBlockService->SyncBasalamHttpBlock()) return;
	50	+ if (get_transient(self::ASYNC_DISPATCH_LOCK_TRANSIENT)) return;
	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	+
	92	+ try {
	93	+ return $this->runEligibleJobs();
	94	+ } finally {
	95	+ $this->jobExecutor->releaseGlobalJobsLock();
	96	+ }
	97	+ }
	98	+
	99	+ private function runAsyncBatch(): int
	100	+ {
	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);
35	149	$this->discountScheduler->process();
36	150
37	151	$circuitBreaker = new CircuitBreaker();
38	152	if ($circuitBreaker->getState() === CircuitBreaker::STATE_OPEN) {
39		- return;
	153	+ return false;
40	154	}
41	155
42		- $tasksPerMinute = max(1, intval(Settings::getEffectiveTasksPerMinute()));
43		- $thresholdSeconds = 60.0 / $tasksPerMinute;
	156	+ if ($this->jobManager->hasAnyProcessingJob()) {
	157	+ return false;
	158	+ }
44	159
	160	+ $lastRun = floatval(get_option(self::GLOBAL_RUNNER_LAST_RUN_OPTION, 0));
	161	+ $now = microtime(true);
	162	+
	163	+ if (($now - $lastRun) < $this->getRunThresholdSeconds()) {
	164	+ return false;
	165	+ }
	166	+
45	167	$sortedJobTypes = $this->jobExecutor->getSortedJobTypes();
46	168
47	169	foreach ($sortedJobTypes as $jobType => $jobExecutor) {
48		- if (!$this->jobExecutor->acquireLock($jobType, 0)) continue;
49		- try {
50		- $lastRun = floatval(get_option($jobType . '_last_run', 0));
51		- $now = microtime(true);
	170	+ if (!$this->jobExecutor->canRun($jobType)) {
	171	+ continue;
	172	+ }
52	173
53		- if (($now - $lastRun) >= $thresholdSeconds) {
54		- if (!$this->jobExecutor->canRun($jobType)) {
55		- continue;
56		- }
	174	+ $job = $this->jobManager->getNextEligibleJob($jobType);
	175	+ $processingJob = $this->jobManager->getJob(['job_type' => $jobType, 'status' => 'processing']);
57	176
58		- $job = $this->jobManager->getNextEligibleJob($jobType);
59		- $processingJob = $this->jobManager->getJob(['job_type' => $jobType, 'status' => 'processing']);
	177	+ if (!$job || $processingJob) {
	178	+ continue;
	179	+ }
60	180
61		- if ($job && !$processingJob) {
62		- update_option($jobType . '_last_run', microtime(true), false);
	181	+ update_option(self::GLOBAL_RUNNER_LAST_RUN_OPTION, microtime(true), false);
63	182
64		- $this->jobManager->updateJob(
65		- ['status' => 'processing', 'started_at' => time()],
66		- ['id' => $job->id]
67		- );
	183	+ $this->jobManager->updateJob(
	184	+ ['status' => 'processing', 'started_at' => time()],
	185	+ ['id' => $job->id]
	186	+ );
68	187
69		- $this->jobExecutor->releaseLock($jobType);
	188	+ $this->executeJob($job);
	189	+ return true;
	190	+ }
70	191
71		- $this->executeJob($job);
72		-
73		- break;
74		- }
75		- }
76		- } finally {
77		- $this->jobExecutor->releaseLock($jobType);
78		- }
79		- }
	192	+ return false;
80	193	}
81	194
82	195	private function executeJob(object $job): void
83	196	{
		@@ -82,6 +195,49 @@
82	195	private function executeJob(object $job): void
83	196	{
84	197	$jobType = $job->job_type;
85	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;
86	242	}
87	243	}