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