← All changes | JobsRunner.php +110 -12 1.10.21 → 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,12 +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';
	16	+ private const ASYNC_IDLE_PROBE_TRANSIENT = 'sync_basalam_jobs_runner_idle_probe_lock';
14	17	// Keep the dispatch lease longer than a normal async batch. Without this,
15	18	// every frontend request can boot another full WordPress AJAX worker while
16	19	// a large product queue is active.
17	20	private const ASYNC_DISPATCH_LOCK_SECONDS = 25;
	21	+ private const ASYNC_IDLE_PROBE_SECONDS = 5;
18	22	private const ASYNC_TIME_LIMIT_SECONDS = 20;
19	23	private const GLOBAL_RUNNER_LAST_RUN_OPTION = 'sync_basalam_jobs_runner_last_run';
20	24	private const STALE_PROCESSING_TIMEOUT_SECONDS = 120;
21	25
		@@ -22,50 +26,136 @@
22	26	private $jobExecutor;
23	27	private $jobManager;
24	28	private $discountScheduler;
25	29	private $CheckHttpBlockService;
	30	+ private $workerHealth;
	31	+ private bool $isAsyncBatch = false;
26	32
27	33	public function __construct(
28	34	$jobManager,
29	35	$jobExecutor,
30	36	$discountScheduler,
31		- $CheckHttpBlockService
	37	+ $CheckHttpBlockService,
	38	+ ?AsyncWorkerHealth $workerHealth = null
32	39	) {
33	40	add_action('wp_ajax_' . self::ASYNC_ACTION, [$this, 'handleAsyncRequest']);
34	41	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);
	42	+ // Run before later shutdown callbacks can leave an unread mysqli result.
	43	+ add_action('shutdown', [$this, 'maybeDispatchAsyncRequest'], 2);
38	44	add_action('sync_basalam_job_created', [$this, 'maybeDispatchAsyncRequest'], 10, 0);
	45	+ add_action('admin_notices', [$this, 'renderWorkerHealthNotice']);
39	46
40	47	$this->jobManager = $jobManager;
41	48	$this->jobExecutor = $jobExecutor;
42	49	$this->discountScheduler = $discountScheduler;
43	50	$this->CheckHttpBlockService = $CheckHttpBlockService;
	51	+ $this->workerHealth = $workerHealth ?? new AsyncWorkerHealth();
44	52	}
45	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	+
46	80	public function maybeDispatchAsyncRequest(): void
47	81	{
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		- // 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.
	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	+ }
	101	+ return;
	102	+ }
	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;
55	108	set_transient(
56	109	self::ASYNC_DISPATCH_LOCK_TRANSIENT,
57		- 1,
	110	+ time() + self::ASYNC_DISPATCH_LOCK_SECONDS,
58	111	self::ASYNC_DISPATCH_LOCK_SECONDS
59	112	);
60	113
61		- if (!$this->jobManager->hasPendingOrStaleProcessingJobs(self::STALE_PROCESSING_TIMEOUT_SECONDS)) {
62		- return;
	114	+ $this->dispatchAsyncRequest();
	115	+ }
	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();
63	126	}
64	127
65		- $this->dispatchAsyncRequest();
	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;
66	136	}
67	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	+
68	158	public function handleAsyncRequest(): void
69	159	{
70	160	if (!check_ajax_referer(self::ASYNC_ACTION, 'nonce', false)) {
71	161	wp_send_json_error(['message' => 'Invalid async jobs runner nonce.'], 403);
		@@ -105,8 +195,9 @@
105	195	// async requests could pile up and sleep in parallel until the next
106	196	// job became eligible, exhausting the site's PHP workers.
107	197	if (!$this->jobExecutor->acquireGlobalJobsLock(0)) return 0;
108	198
	199	+ $this->isAsyncBatch = true;
109	200	$processed = 0;
110	201	$deadline = microtime(true) + (float) apply_filters(
111	202	'sync_basalam_jobs_runner_async_time_limit',
112	203	self::ASYNC_TIME_LIMIT_SECONDS
		@@ -112,8 +203,9 @@
112	203	self::ASYNC_TIME_LIMIT_SECONDS
113	204	);
114	205
115	206	try {
	207	+ $this->workerHealth->observeAsyncEnvironment();
116	208	while (microtime(true) < $deadline) {
117	209	if (!$this->jobManager->hasPendingOrStaleProcessingJobs(self::STALE_PROCESSING_TIMEOUT_SECONDS)) {
118	210	break;
119	211	}
		@@ -136,8 +228,9 @@
136	228
137	229	usleep((int) ($delay * 1000000));
138	230	}
139	231	} finally {
	232	+ $this->isAsyncBatch = false;
140	233	$this->jobExecutor->releaseGlobalJobsLock();
141	234	}
142	235
143	236	return $processed;
		@@ -144,8 +237,9 @@
144	237	}
145	238
146	239	private function runEligibleJobs(): bool
147	240	{
	241	+ $this->detectInterruptedWorker();
148	242	$this->jobManager->ConvertStaleProcessingJobs(self::STALE_PROCESSING_TIMEOUT_SECONDS);
149	243	$this->discountScheduler->process();
150	244
151	245	$circuitBreaker = new CircuitBreaker();
		@@ -194,9 +288,13 @@
194	288
195	289	private function executeJob(object $job): void
196	290	{
197	291	$jobType = $job->job_type;
	292	+ $started = microtime(true);
198	293	$this->jobExecutor->execute($jobType, $job);
	294	+ if ($this->isAsyncBatch) {
	295	+ $this->workerHealth->recordJobReturned($job, microtime(true) - $started);
	296	+ }
199	297	}
200	298
201	299	private function dispatchAsyncRequest(): void
202	300	{