← All changes | JobsRunner.php +80 -29 1.10.14 → 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,11 +33,10 @@
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		- // Keep database work out of shutdown, where another callback may have
33		- // left the shared mysqli connection with an unread result set.
34		- add_action('init', [$this, 'maybeDispatchAsyncRequest'], 10, 0);
	37	+ // Run before later shutdown callbacks can leave an unread mysqli result.
	38	+ add_action('shutdown', [$this, 'maybeDispatchAsyncRequest'], 2);
35	39	add_action('sync_basalam_job_created', [$this, 'maybeDispatchAsyncRequest'], 10, 0);
36	40
37	41	$this->jobManager = $jobManager;
38	42	$this->jobExecutor = $jobExecutor;
		@@ -41,28 +45,63 @@
41	45	}
42	46
43	47	public function maybeDispatchAsyncRequest(): void
44	48	{
45		- // Retain a guard for third-party callers that may still invoke this
46		- // method during shutdown even though no shutdown hook is registered.
47		- if (function_exists('did_action') && did_action('shutdown')) return;
48	49	if ($this->isCurrentAsyncRequest()) return;
	50	+
	51	+ if ($this->isShutdownCallback()) {
	52	+ $this->finishFastCgiResponse();
	53	+ $this->repairDatabaseConnection();
	54	+ }
	55	+
49	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;
50	59	if (get_transient(self::ASYNC_DISPATCH_LOCK_TRANSIENT)) return;
51	60
52	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	+ }
53	65	return;
54	66	}
55	67
56		- set_transient(
57		- self::ASYNC_DISPATCH_LOCK_TRANSIENT,
58		- 1,
59		- self::ASYNC_DISPATCH_LOCK_SECONDS
60		- );
	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);
61	73
62	74	$this->dispatchAsyncRequest();
63	75	}
64	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	+
65	104	public function handleAsyncRequest(): void
66	105	{
67	106	if (!check_ajax_referer(self::ASYNC_ACTION, 'nonce', false)) {
68	107	wp_send_json_error(['message' => 'Invalid async jobs runner nonce.'], 403);
		@@ -94,8 +133,16 @@
94	133	}
95	134
96	135	private function runAsyncBatch(): int
97	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	+
98	145	$processed = 0;
99	146	$deadline = microtime(true) + (float) apply_filters(
100	147	'sync_basalam_jobs_runner_async_time_limit',
101	148	self::ASYNC_TIME_LIMIT_SECONDS
		@@ -100,30 +147,34 @@
100	147	'sync_basalam_jobs_runner_async_time_limit',
101	148	self::ASYNC_TIME_LIMIT_SECONDS
102	149	);
103	150
104		- while (microtime(true) < $deadline) {
105		- if (!$this->jobManager->hasPendingOrStaleProcessingJobs(self::STALE_PROCESSING_TIMEOUT_SECONDS)) {
106		- break;
107		- }
	151	+ try {
	152	+ while (microtime(true) < $deadline) {
	153	+ if (!$this->jobManager->hasPendingOrStaleProcessingJobs(self::STALE_PROCESSING_TIMEOUT_SECONDS)) {
	154	+ break;
	155	+ }
108	156
109		- $ranJob = $this->checkAndRunJobs();
	157	+ $ranJob = $this->runEligibleJobs();
110	158
111		- if ($ranJob) {
112		- $processed++;
113		- }
	159	+ if ($ranJob) {
	160	+ $processed++;
	161	+ }
114	162
115		- $delay = $this->secondsUntilNextAllowedRun();
116		- if ($delay <= 0.0) {
117		- if (!$ranJob) break;
118		- continue;
119		- }
	163	+ $delay = $this->secondsUntilNextAllowedRun();
	164	+ if ($delay <= 0.0) {
	165	+ if (!$ranJob) break;
	166	+ continue;
	167	+ }
120	168
121		- if ((microtime(true) + $delay) >= $deadline) {
122		- break;
	169	+ if ((microtime(true) + $delay) >= $deadline) {
	170	+ break;
	171	+ }
	172	+
	173	+ usleep((int) ($delay * 1000000));
123	174	}
124		-
125		- usleep((int) ($delay * 1000000));
	175	+ } finally {
	176	+ $this->jobExecutor->releaseGlobalJobsLock();
126	177	}
127	178
128	179	return $processed;
129	180	}