← All changes | JobsRunner.php +140 -27 1.10.14 → 1.10.29 View file →

		@@ -2,9 +2,11 @@
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
		@@ -10,9 +12,14 @@
10	12	class JobsRunner
11	13	{
12	14	private const ASYNC_ACTION = 'sync_basalam_run_jobs_async';
13	15	private const ASYNC_DISPATCH_LOCK_TRANSIENT = 'sync_basalam_jobs_runner_async_dispatch_lock';
14		- private const ASYNC_DISPATCH_LOCK_SECONDS = 1;
	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;
15	22	private const ASYNC_TIME_LIMIT_SECONDS = 20;
16	23	private const GLOBAL_RUNNER_LAST_RUN_OPTION = 'sync_basalam_jobs_runner_last_run';
17	24	private const STALE_PROCESSING_TIMEOUT_SECONDS = 120;
18	25
		@@ -19,44 +26,89 @@
19	26	private $jobExecutor;
20	27	private $jobManager;
21	28	private $discountScheduler;
22	29	private $CheckHttpBlockService;
	30	+ private $workerHealth;
	31	+ private bool $isAsyncBatch = false;
23	32
24	33	public function __construct(
25	34	$jobManager,
26	35	$jobExecutor,
27	36	$discountScheduler,
28		- $CheckHttpBlockService
	37	+ $CheckHttpBlockService,
	38	+ ?AsyncWorkerHealth $workerHealth = null
29	39	) {
30	40	add_action('wp_ajax_' . self::ASYNC_ACTION, [$this, 'handleAsyncRequest']);
31	41	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);
	42	+ // Run before later shutdown callbacks can leave an unread mysqli result.
	43	+ add_action('shutdown', [$this, 'maybeDispatchAsyncRequest'], 2);
35	44	add_action('sync_basalam_job_created', [$this, 'maybeDispatchAsyncRequest'], 10, 0);
	45	+ add_action('admin_notices', [$this, 'renderWorkerHealthNotice']);
36	46
37	47	$this->jobManager = $jobManager;
38	48	$this->jobExecutor = $jobExecutor;
39	49	$this->discountScheduler = $discountScheduler;
40	50	$this->CheckHttpBlockService = $CheckHttpBlockService;
	51	+ $this->workerHealth = $workerHealth ?? new AsyncWorkerHealth();
41	52	}
42	53
	54	+ public function renderWorkerHealthNotice(): void
	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	+
43	80	public function maybeDispatchAsyncRequest(): void
44	81	{
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	82	if ($this->isCurrentAsyncRequest()) return;
	83	+
	84	+ if ($this->isShutdownCallback()) {
	85	+ $this->repairDatabaseConnection();
	86	+ }
	87	+
49	88	if ($this->CheckHttpBlockService->SyncBasalamHttpBlock()) return;
50		- if (get_transient(self::ASYNC_DISPATCH_LOCK_TRANSIENT)) 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;
51	92
52	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	+ }
53	101	return;
54	102	}
55	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;
56	108	set_transient(
57	109	self::ASYNC_DISPATCH_LOCK_TRANSIENT,
58		- 1,
	110	+ time() + self::ASYNC_DISPATCH_LOCK_SECONDS,
59	111	self::ASYNC_DISPATCH_LOCK_SECONDS
60	112	);
61	113
62	114	$this->dispatchAsyncRequest();
		@@ -61,8 +113,49 @@
61	113
62	114	$this->dispatchAsyncRequest();
63	115	}
64	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	+
65	158	public function handleAsyncRequest(): void
66	159	{
67	160	if (!check_ajax_referer(self::ASYNC_ACTION, 'nonce', false)) {
68	161	wp_send_json_error(['message' => 'Invalid async jobs runner nonce.'], 403);
		@@ -94,8 +187,17 @@
94	187	}
95	188
96	189	private function runAsyncBatch(): int
97	190	{
	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;
98	200	$processed = 0;
99	201	$deadline = microtime(true) + (float) apply_filters(
100	202	'sync_basalam_jobs_runner_async_time_limit',
101	203	self::ASYNC_TIME_LIMIT_SECONDS
		@@ -100,30 +202,36 @@
100	202	'sync_basalam_jobs_runner_async_time_limit',
101	203	self::ASYNC_TIME_LIMIT_SECONDS
102	204	);
103	205
104		- while (microtime(true) < $deadline) {
105		- if (!$this->jobManager->hasPendingOrStaleProcessingJobs(self::STALE_PROCESSING_TIMEOUT_SECONDS)) {
106		- break;
107		- }
	206	+ try {
	207	+ $this->workerHealth->observeAsyncEnvironment();
	208	+ while (microtime(true) < $deadline) {
	209	+ if (!$this->jobManager->hasPendingOrStaleProcessingJobs(self::STALE_PROCESSING_TIMEOUT_SECONDS)) {
	210	+ break;
	211	+ }
108	212
109		- $ranJob = $this->checkAndRunJobs();
	213	+ $ranJob = $this->runEligibleJobs();
110	214
111		- if ($ranJob) {
112		- $processed++;
113		- }
	215	+ if ($ranJob) {
	216	+ $processed++;
	217	+ }
114	218
115		- $delay = $this->secondsUntilNextAllowedRun();
116		- if ($delay <= 0.0) {
117		- if (!$ranJob) break;
118		- continue;
119		- }
	219	+ $delay = $this->secondsUntilNextAllowedRun();
	220	+ if ($delay <= 0.0) {
	221	+ if (!$ranJob) break;
	222	+ continue;
	223	+ }
120	224
121		- if ((microtime(true) + $delay) >= $deadline) {
122		- break;
	225	+ if ((microtime(true) + $delay) >= $deadline) {
	226	+ break;
	227	+ }
	228	+
	229	+ usleep((int) ($delay * 1000000));
123	230	}
124		-
125		- usleep((int) ($delay * 1000000));
	231	+ } finally {
	232	+ $this->isAsyncBatch = false;
	233	+ $this->jobExecutor->releaseGlobalJobsLock();
126	234	}
127	235
128	236	return $processed;
129	237	}
		@@ -129,8 +237,9 @@
129	237	}
130	238
131	239	private function runEligibleJobs(): bool
132	240	{
	241	+ $this->detectInterruptedWorker();
133	242	$this->jobManager->ConvertStaleProcessingJobs(self::STALE_PROCESSING_TIMEOUT_SECONDS);
134	243	$this->discountScheduler->process();
135	244
136	245	$circuitBreaker = new CircuitBreaker();
		@@ -179,9 +288,13 @@
179	288
180	289	private function executeJob(object $job): void
181	290	{
182	291	$jobType = $job->job_type;
	292	+ $started = microtime(true);
183	293	$this->jobExecutor->execute($jobType, $job);
	294	+ if ($this->isAsyncBatch) {
	295	+ $this->workerHealth->recordJobReturned($job, microtime(true) - $started);
	296	+ }
184	297	}
185	298
186	299	private function dispatchAsyncRequest(): void
187	300	{