← All changes | JobsRunner.php +75 -13 1.10.24 → 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
		@@ -24,14 +26,17 @@
24	26	private $jobExecutor;
25	27	private $jobManager;
26	28	private $discountScheduler;
27	29	private $CheckHttpBlockService;
	30	+ private $workerHealth;
	31	+ private bool $isAsyncBatch = false;
28	32
29	33	public function __construct(
30	34	$jobManager,
31	35	$jobExecutor,
32	36	$discountScheduler,
33		- $CheckHttpBlockService
	37	+ $CheckHttpBlockService,
	38	+ ?AsyncWorkerHealth $workerHealth = null
34	39	) {
35	40	add_action('wp_ajax_' . self::ASYNC_ACTION, [$this, 'handleAsyncRequest']);
36	41	add_action('wp_ajax_nopriv_' . self::ASYNC_ACTION, [$this, 'handleAsyncRequest']);
37	42	// Run before later shutdown callbacks can leave an unread mysqli result.
		@@ -36,32 +41,63 @@
36	41	add_action('wp_ajax_nopriv_' . self::ASYNC_ACTION, [$this, 'handleAsyncRequest']);
37	42	// Run before later shutdown callbacks can leave an unread mysqli result.
38	43	add_action('shutdown', [$this, 'maybeDispatchAsyncRequest'], 2);
39	44	add_action('sync_basalam_job_created', [$this, 'maybeDispatchAsyncRequest'], 10, 0);
	45	+ add_action('admin_notices', [$this, 'renderWorkerHealthNotice']);
40	46
41	47	$this->jobManager = $jobManager;
42	48	$this->jobExecutor = $jobExecutor;
43	49	$this->discountScheduler = $discountScheduler;
44	50	$this->CheckHttpBlockService = $CheckHttpBlockService;
	51	+ $this->workerHealth = $workerHealth ?? new AsyncWorkerHealth();
45	52	}
46	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	+
47	80	public function maybeDispatchAsyncRequest(): void
48	81	{
49	82	if ($this->isCurrentAsyncRequest()) return;
50	83
51	84	if ($this->isShutdownCallback()) {
52		- $this->finishFastCgiResponse();
53	85	$this->repairDatabaseConnection();
54	86	}
55	87
56	88	if ($this->CheckHttpBlockService->SyncBasalamHttpBlock()) return;
57	89	$isNewJob = function_exists('current_filter') && current_filter() === 'sync_basalam_job_created';
58		- if (!$isNewJob && get_transient(self::ASYNC_IDLE_PROBE_TRANSIENT)) return;
59		- if (get_transient(self::ASYNC_DISPATCH_LOCK_TRANSIENT)) return;
	90	+ if (!$isNewJob && $this->hasActiveAsyncLease(self::ASYNC_IDLE_PROBE_TRANSIENT)) return;
	91	+ if ($this->hasActiveAsyncLease(self::ASYNC_DISPATCH_LOCK_TRANSIENT)) return;
60	92
61	93	if (!$this->jobManager->hasPendingOrStaleProcessingJobs(self::STALE_PROCESSING_TIMEOUT_SECONDS)) {
62	94	if (!$isNewJob) {
63		- set_transient(self::ASYNC_IDLE_PROBE_TRANSIENT, 1, self::ASYNC_IDLE_PROBE_SECONDS);
	95	+ set_transient(
	96	+ self::ASYNC_IDLE_PROBE_TRANSIENT,
	97	+ time() + self::ASYNC_IDLE_PROBE_SECONDS,
	98	+ self::ASYNC_IDLE_PROBE_SECONDS
	99	+ );
64	100	}
65	101	return;
66	102	}
67	103
		@@ -67,24 +103,42 @@
67	103
68	104	// An empty-queue probe must not block a job created moments later.
69	105	// The global database lock still prevents duplicate workers from
70	106	// 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);
	107	+ if ($this->hasActiveAsyncLease(self::ASYNC_DISPATCH_LOCK_TRANSIENT)) return;
	108	+ set_transient(
	109	+ self::ASYNC_DISPATCH_LOCK_TRANSIENT,
	110	+ time() + self::ASYNC_DISPATCH_LOCK_SECONDS,
	111	+ self::ASYNC_DISPATCH_LOCK_SECONDS
	112	+ );
73	113
74	114	$this->dispatchAsyncRequest();
75	115	}
76	116
77		- private function isShutdownCallback(): bool
	117	+ private function hasActiveAsyncLease(string $transient): bool
78	118	{
79		- return function_exists('current_filter') && current_filter() === 'shutdown';
	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;
80	136	}
81	137
82		- private function finishFastCgiResponse(): void
	138	+ private function isShutdownCallback(): bool
83	139	{
84		- if (function_exists('fastcgi_finish_request')) {
85		- fastcgi_finish_request();
86		- }
	140	+ return function_exists('current_filter') && current_filter() === 'shutdown';
87	141	}
88	142
89	143	private function repairDatabaseConnection(): void
90	144	{
		@@ -141,8 +195,9 @@
141	195	// async requests could pile up and sleep in parallel until the next
142	196	// job became eligible, exhausting the site's PHP workers.
143	197	if (!$this->jobExecutor->acquireGlobalJobsLock(0)) return 0;
144	198
	199	+ $this->isAsyncBatch = true;
145	200	$processed = 0;
146	201	$deadline = microtime(true) + (float) apply_filters(
147	202	'sync_basalam_jobs_runner_async_time_limit',
148	203	self::ASYNC_TIME_LIMIT_SECONDS
		@@ -148,8 +203,9 @@
148	203	self::ASYNC_TIME_LIMIT_SECONDS
149	204	);
150	205
151	206	try {
	207	+ $this->workerHealth->observeAsyncEnvironment();
152	208	while (microtime(true) < $deadline) {
153	209	if (!$this->jobManager->hasPendingOrStaleProcessingJobs(self::STALE_PROCESSING_TIMEOUT_SECONDS)) {
154	210	break;
155	211	}
		@@ -172,8 +228,9 @@
172	228
173	229	usleep((int) ($delay * 1000000));
174	230	}
175	231	} finally {
	232	+ $this->isAsyncBatch = false;
176	233	$this->jobExecutor->releaseGlobalJobsLock();
177	234	}
178	235
179	236	return $processed;
		@@ -180,8 +237,9 @@
180	237	}
181	238
182	239	private function runEligibleJobs(): bool
183	240	{
	241	+ $this->detectInterruptedWorker();
184	242	$this->jobManager->ConvertStaleProcessingJobs(self::STALE_PROCESSING_TIMEOUT_SECONDS);
185	243	$this->discountScheduler->process();
186	244
187	245	$circuitBreaker = new CircuitBreaker();
		@@ -230,9 +288,13 @@
230	288
231	289	private function executeJob(object $job): void
232	290	{
233	291	$jobType = $job->job_type;
	292	+ $started = microtime(true);
234	293	$this->jobExecutor->execute($jobType, $job);
	294	+ if ($this->isAsyncBatch) {
	295	+ $this->workerHealth->recordJobReturned($job, microtime(true) - $started);
	296	+ }
235	297	}
236	298
237	299	private function dispatchAsyncRequest(): void
238	300	{