← All changes | JobsRunner.php +82 -26 1.10.10 → 1.10.22 View file →

		@@ -10,9 +10,14 @@
10	10	class JobsRunner
11	11	{
12	12	private const ASYNC_ACTION = 'sync_basalam_run_jobs_async';
13	13	private const ASYNC_DISPATCH_LOCK_TRANSIENT = 'sync_basalam_jobs_runner_async_dispatch_lock';
14		- private const ASYNC_DISPATCH_LOCK_SECONDS = 1;
	14	+ private const ASYNC_IDLE_PROBE_TRANSIENT = 'sync_basalam_jobs_runner_idle_probe_lock';
	15	+ // Keep the dispatch lease longer than a normal async batch. Without this,
	16	+ // every frontend request can boot another full WordPress AJAX worker while
	17	+ // a large product queue is active.
	18	+ private const ASYNC_DISPATCH_LOCK_SECONDS = 25;
	19	+ private const ASYNC_IDLE_PROBE_SECONDS = 5;
15	20	private const ASYNC_TIME_LIMIT_SECONDS = 20;
16	21	private const GLOBAL_RUNNER_LAST_RUN_OPTION = 'sync_basalam_jobs_runner_last_run';
17	22	private const STALE_PROCESSING_TIMEOUT_SECONDS = 120;
18	23
		@@ -28,10 +33,11 @@
28	33	$CheckHttpBlockService
29	34	) {
30	35	add_action('wp_ajax_' . self::ASYNC_ACTION, [$this, 'handleAsyncRequest']);
31	36	add_action('wp_ajax_nopriv_' . self::ASYNC_ACTION, [$this, 'handleAsyncRequest']);
32		- add_action('shutdown', [$this, 'maybeDispatchAsyncRequest'], PHP_INT_MAX);
33		- add_action('sync_basalam_job_created', [$this, 'maybeDispatchAsyncRequest']);
	37	+ // Run before later shutdown callbacks can leave an unread mysqli result.
	38	+ add_action('shutdown', [$this, 'maybeDispatchAsyncRequest'], 2);
	39	+ add_action('sync_basalam_job_created', [$this, 'maybeDispatchAsyncRequest'], 10, 0);
34	40
35	41	$this->jobManager = $jobManager;
36	42	$this->jobExecutor = $jobExecutor;
37	43	$this->discountScheduler = $discountScheduler;
		@@ -40,24 +46,62 @@
40	46
41	47	public function maybeDispatchAsyncRequest(): void
42	48	{
43	49	if ($this->isCurrentAsyncRequest()) return;
	50	+
	51	+ if ($this->isShutdownCallback()) {
	52	+ $this->finishFastCgiResponse();
	53	+ $this->repairDatabaseConnection();
	54	+ }
	55	+
44	56	if ($this->CheckHttpBlockService->SyncBasalamHttpBlock()) return;
	57	+ $isNewJob = function_exists('current_filter') && current_filter() === 'sync_basalam_job_created';
	58	+ if (!$isNewJob && get_transient(self::ASYNC_IDLE_PROBE_TRANSIENT)) return;
45	59	if (get_transient(self::ASYNC_DISPATCH_LOCK_TRANSIENT)) return;
46	60
47		- set_transient(
48		- self::ASYNC_DISPATCH_LOCK_TRANSIENT,
49		- 1,
50		- self::ASYNC_DISPATCH_LOCK_SECONDS
51		- );
52		-
53	61	if (!$this->jobManager->hasPendingOrStaleProcessingJobs(self::STALE_PROCESSING_TIMEOUT_SECONDS)) {
	62	+ if (!$isNewJob) {
	63	+ set_transient(self::ASYNC_IDLE_PROBE_TRANSIENT, 1, self::ASYNC_IDLE_PROBE_SECONDS);
	64	+ }
54	65	return;
55	66	}
56	67
	68	+ // An empty-queue probe must not block a job created moments later.
	69	+ // The global database lock still prevents duplicate workers from
	70	+ // processing the same queue when requests race here.
	71	+ if (get_transient(self::ASYNC_DISPATCH_LOCK_TRANSIENT)) return;
	72	+ set_transient(self::ASYNC_DISPATCH_LOCK_TRANSIENT, 1, self::ASYNC_DISPATCH_LOCK_SECONDS);
	73	+
57	74	$this->dispatchAsyncRequest();
58	75	}
59	76
	77	+ private function isShutdownCallback(): bool
	78	+ {
	79	+ return function_exists('current_filter') && current_filter() === 'shutdown';
	80	+ }
	81	+
	82	+ private function finishFastCgiResponse(): void
	83	+ {
	84	+ if (function_exists('fastcgi_finish_request')) {
	85	+ fastcgi_finish_request();
	86	+ }
	87	+ }
	88	+
	89	+ private function repairDatabaseConnection(): void
	90	+ {
	91	+ global $wpdb;
	92	+
	93	+ if (!isset($wpdb) || !is_object($wpdb)) return;
	94	+
	95	+ if (method_exists($wpdb, 'flush')) {
	96	+ $wpdb->flush();
	97	+ }
	98	+
	99	+ if (method_exists($wpdb, 'check_connection')) {
	100	+ $wpdb->check_connection(false);
	101	+ }
	102	+ }
	103	+
60	104	public function handleAsyncRequest(): void
61	105	{
62	106	if (!check_ajax_referer(self::ASYNC_ACTION, 'nonce', false)) {
63	107	wp_send_json_error(['message' => 'Invalid async jobs runner nonce.'], 403);
		@@ -89,8 +133,16 @@
89	133	}
90	134
91	135	private function runAsyncBatch(): int
92	136	{
	137	+ if ($this->CheckHttpBlockService->SyncBasalamHttpBlock()) return 0;
	138	+
	139	+ // Hold the advisory lock for the whole batch, including rate-limit
	140	+ // waits. Previously it was released after every job, so duplicate
	141	+ // async requests could pile up and sleep in parallel until the next
	142	+ // job became eligible, exhausting the site's PHP workers.
	143	+ if (!$this->jobExecutor->acquireGlobalJobsLock(0)) return 0;
	144	+
93	145	$processed = 0;
94	146	$deadline = microtime(true) + (float) apply_filters(
95	147	'sync_basalam_jobs_runner_async_time_limit',
96	148	self::ASYNC_TIME_LIMIT_SECONDS
		@@ -95,30 +147,34 @@
95	147	'sync_basalam_jobs_runner_async_time_limit',
96	148	self::ASYNC_TIME_LIMIT_SECONDS
97	149	);
98	150
99		- while (microtime(true) < $deadline) {
100		- if (!$this->jobManager->hasPendingOrStaleProcessingJobs(self::STALE_PROCESSING_TIMEOUT_SECONDS)) {
101		- break;
102		- }
	151	+ try {
	152	+ while (microtime(true) < $deadline) {
	153	+ if (!$this->jobManager->hasPendingOrStaleProcessingJobs(self::STALE_PROCESSING_TIMEOUT_SECONDS)) {
	154	+ break;
	155	+ }
103	156
104		- $ranJob = $this->checkAndRunJobs();
	157	+ $ranJob = $this->runEligibleJobs();
105	158
106		- if ($ranJob) {
107		- $processed++;
108		- }
	159	+ if ($ranJob) {
	160	+ $processed++;
	161	+ }
109	162
110		- $delay = $this->secondsUntilNextAllowedRun();
111		- if ($delay <= 0.0) {
112		- if (!$ranJob) break;
113		- continue;
114		- }
	163	+ $delay = $this->secondsUntilNextAllowedRun();
	164	+ if ($delay <= 0.0) {
	165	+ if (!$ranJob) break;
	166	+ continue;
	167	+ }
115	168
116		- if ((microtime(true) + $delay) >= $deadline) {
117		- break;
	169	+ if ((microtime(true) + $delay) >= $deadline) {
	170	+ break;
	171	+ }
	172	+
	173	+ usleep((int) ($delay * 1000000));
118	174	}
119		-
120		- usleep((int) ($delay * 1000000));
	175	+ } finally {
	176	+ $this->jobExecutor->releaseGlobalJobsLock();
121	177	}
122	178
123	179	return $processed;
124	180	}