From 63e2e2e1743306b936a5eb3dd6bc3f8570b6e30c Mon Sep 17 00:00:00 2001 From: Olivier Gorzalka Date: Wed, 7 Oct 2026 15:17:41 +0200 Subject: [PATCH] feat: queue async actions through Action Scheduler and resolve auto --- .github/workflows/tests.yml | 1 + CHANGELOG.md | 8 + src/Async/Async.php | 53 +++++- src/Async/Drivers/ActionSchedulerDriver.php | 80 ++++++++ src/Async/Drivers/WpCronDriver.php | 35 ++-- src/Async/PayloadStore.php | 65 +++++++ src/Async/PendingAsync.php | 4 +- tests/Stubs/wordpress.php | 48 +++++ .../Unit/Async/ActionSchedulerDriverTest.php | 171 ++++++++++++++++++ tests/Unit/Async/AsyncActionTest.php | 2 +- tests/Unit/Async/AsyncTest.php | 27 ++- tests/WordPress/ActionSchedulerTest.php | 128 +++++++++++++ tests/WordPress/Fixtures/async-actions.php | 35 ++++ 13 files changed, 628 insertions(+), 29 deletions(-) create mode 100644 src/Async/Drivers/ActionSchedulerDriver.php create mode 100644 src/Async/PayloadStore.php create mode 100644 tests/Unit/Async/ActionSchedulerDriverTest.php create mode 100644 tests/WordPress/ActionSchedulerTest.php diff --git a/.github/workflows/tests.yml b/.github/workflows/tests.yml index a59d76e..486d06f 100644 --- a/.github/workflows/tests.yml +++ b/.github/workflows/tests.yml @@ -96,6 +96,7 @@ jobs: wp config create --path=.wordpress --dbname=wordpress --dbuser=root --dbpass=root --dbhost=127.0.0.1:3306 wp core install --path=.wordpress --url=http://localhost --title=Hook \ --admin_user=admin --admin_password=admin --admin_email=admin@example.com --skip-email + wp plugin install action-scheduler --activate --path=.wordpress - name: Run tests against WordPress run: composer test:wordpress diff --git a/CHANGELOG.md b/CHANGELOG.md index e82554d..a0e5584 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -4,6 +4,12 @@ All notable changes to `pollora/hook` are documented in this file. ## [Unreleased] +### Changed + +- **The default driver is now `auto`**: Action Scheduler when a plugin bundling it is active and initialised, WP-Cron otherwise, resolved on each dispatch. A site with WooCommerce therefore queues through Action Scheduler after upgrading; payloads already queued through WP-Cron still run. Set `POLLORA_ASYNC_DRIVER` to `wp-cron` to keep the previous behaviour. +- The WP-Cron recovery task also leaves alone a stored payload whose Action Scheduler action is still pending. +- The type of `capture()` and `when()` callbacks is documented as `callable`, so closures with typed parameters pass static analysis. + ### Added - `capture()` on an asynchronous registration: a callback run at trigger time, in the original request, with the hook arguments; the array it returns is read back at execution through `AsyncContext::get()`. Captured values travel like arguments. A handler class's public `capture()` method is used when `capture()` is not called. @@ -12,6 +18,8 @@ All notable changes to `pollora/hook` are documented in this file. - `asUser()`: run as the user who fired the hook, then remove that user. Without it the handler runs without a current user. - `onQueue()`: queue name, used as the Action Scheduler group and the Laravel queue; WP-Cron ignores it. - `unique()`: merge identical triggers (same hook, same handler, same arguments) while the first one has not started running. The lock is an option that is not autoloaded, taken atomically with `add_option()`, released when the first attempt starts, and expiring after `unique(for: …)` seconds (one day by default) if never released; the daily maintenance task deletes expired locks. A lock that cannot be stored is a queuing failure, never a silent merge. +- `action-scheduler` driver, offered when a plugin bundling Action Scheduler is active and its data store is initialised (never a dependency). The payload travels whole as the action argument, the queue name becomes the group (`pollora` by default), and history is visible in Tools › Scheduled Actions. A payload too long for Action Scheduler's arguments (8,000 characters as JSON) is kept in an option that is not autoloaded and the action carries only its identifier. +- `Async::setAutoDrivers()`: the drivers `auto` tries, in order. - Payloads record the driver that queued them, the queue, `asUser`, the backoff and the unique lock. These fields are optional, so payloads queued by 1.2.0 still run. ## [1.2.0] - 2026-10-07 diff --git a/src/Async/Async.php b/src/Async/Async.php index 71c5f97..17b10b1 100644 --- a/src/Async/Async.php +++ b/src/Async/Async.php @@ -6,6 +6,7 @@ use Pollora\Hook\Async\Contracts\AsyncDriver; use Pollora\Hook\Async\Contracts\ObjectReference; +use Pollora\Hook\Async\Drivers\ActionSchedulerDriver; use Pollora\Hook\Async\Drivers\SyncDriver; use Pollora\Hook\Async\Drivers\WpCronDriver; use Pollora\Hook\Async\Exceptions\DriverUnavailable; @@ -24,7 +25,10 @@ final class Async */ public const string HOOK = 'pollora/async/run'; - public const string DEFAULT_DRIVER = 'wp-cron'; + /** + * Default driver: the first available among the auto drivers. + */ + public const string DEFAULT_DRIVER = 'auto'; /** * Constant that sets the default driver, in wp-config.php. @@ -44,6 +48,9 @@ final class Async private static ?string $defaultDriver = null; + /** @var list Drivers 'auto' tries, in order */ + private static array $autoDrivers = ['action-scheduler', 'wp-cron']; + private static ?CallbackResolverInterface $resolver = null; private static ?ArgumentNormalizer $normalizer = null; @@ -96,10 +103,12 @@ public static function driver(?string $name = null): AsyncDriver /** * Name of the default driver, from the first of: setDefaultDriver(), the * POLLORA_ASYNC_DRIVER constant, the 'pollora/hook/async_driver' filter, - * 'wp-cron'. + * 'auto'. * - * 'auto' picks the best mechanism available; until the Action Scheduler - * driver exists, that is WP-Cron. + * 'auto' resolves to the first available of the auto drivers: Action + * Scheduler when a plugin bundling it is active and initialised, WP-Cron + * otherwise. It is resolved on each dispatch, so a hook fired before Action + * Scheduler is initialised still queues, through WP-Cron. */ public static function defaultDriver(): string { @@ -116,7 +125,17 @@ public static function defaultDriver(): string $driver ??= self::DEFAULT_DRIVER; - return $driver === 'auto' ? self::DEFAULT_DRIVER : $driver; + return $driver === 'auto' ? self::resolveAuto() : $driver; + } + + /** + * Set the drivers 'auto' tries, in order. The framework puts its Laravel queue first. + * + * @param array $drivers The last one should always be available: 'wp-cron' + */ + public static function setAutoDrivers(array $drivers): void + { + self::$autoDrivers = array_values($drivers); } /** @@ -165,12 +184,12 @@ public static function listen(): void /** * Run a queued handler, from the message a driver hands back. * - * @param string $message The payload as JSON, or the identifier of a payload the wp-cron driver stored + * @param string $message The payload as JSON, or the identifier of a payload kept by PayloadStore */ public static function receive(string $message): void { - if (WpCronDriver::handles($message)) { - $message = WpCronDriver::claim($message); + if (PayloadStore::handles($message)) { + $message = PayloadStore::claim($message); if ($message === null) { return; @@ -263,6 +282,7 @@ public static function flush(): void self::$factories = []; self::$drivers = []; self::$defaultDriver = null; + self::$autoDrivers = ['action-scheduler', 'wp-cron']; self::$resolver = null; self::$normalizer = null; self::$dispatcher = null; @@ -271,6 +291,22 @@ public static function flush(): void self::$reporter = null; } + /** + * The first available auto driver, WP-Cron when none is. + */ + private static function resolveAuto(): string + { + $factories = self::factories(); + + foreach (self::$autoDrivers as $name) { + if (isset($factories[$name]) && (self::$drivers[$name] ??= $factories[$name]())->available()) { + return $name; + } + } + + return 'wp-cron'; + } + private static function normalizer(): ArgumentNormalizer { return self::$normalizer ??= new ArgumentNormalizer; @@ -282,6 +318,7 @@ private static function normalizer(): ArgumentNormalizer private static function factories(): array { return self::$factories + [ + 'action-scheduler' => static fn (): AsyncDriver => new ActionSchedulerDriver, 'wp-cron' => static fn (): AsyncDriver => new WpCronDriver, 'sync' => static fn (): AsyncDriver => new SyncDriver, ]; diff --git a/src/Async/Drivers/ActionSchedulerDriver.php b/src/Async/Drivers/ActionSchedulerDriver.php new file mode 100644 index 0000000..0d50753 --- /dev/null +++ b/src/Async/Drivers/ActionSchedulerDriver.php @@ -0,0 +1,80 @@ + 0; + } + + /** + * @throws AsyncException When Action Scheduler does not create the action + */ + public function dispatch(AsyncPayload $payload, int $delay = 0): void + { + $group = $payload->queue ?? self::DEFAULT_GROUP; + $arguments = [$payload->toJson()]; + $storedApart = strlen((string) json_encode($arguments)) > self::MAX_ARGUMENTS_LENGTH; + + if ($storedApart) { + PayloadStore::store($payload); + $arguments = [$payload->id]; + } + + try { + $actionId = $delay > 0 + ? as_schedule_single_action(time() + $delay, Async::HOOK, $arguments, $group) + : as_enqueue_async_action(Async::HOOK, $arguments, $group); + } catch (\Throwable $throwable) { + $actionId = 0; + $reason = $throwable->getMessage(); + } + + if ($actionId === 0) { + if ($storedApart) { + PayloadStore::forget($payload->id); + } + + throw new AsyncException(sprintf("Action Scheduler: the execution of '%s' could not be queued%s", $payload->hook, isset($reason) ? ': '.$reason : '.')); + } + + if ($storedApart) { + WpCronDriver::scheduleRecovery(); + } + } +} diff --git a/src/Async/Drivers/WpCronDriver.php b/src/Async/Drivers/WpCronDriver.php index b56aedc..b943cf4 100644 --- a/src/Async/Drivers/WpCronDriver.php +++ b/src/Async/Drivers/WpCronDriver.php @@ -8,6 +8,7 @@ use Pollora\Hook\Async\AsyncPayload; use Pollora\Hook\Async\Contracts\AsyncDriver; use Pollora\Hook\Async\Exceptions\AsyncException; +use Pollora\Hook\Async\PayloadStore; use Pollora\Hook\Async\UniqueLock; /** @@ -27,7 +28,7 @@ */ final class WpCronDriver implements AsyncDriver { - public const string OPTION_PREFIX = 'pollora_async_'; + public const string OPTION_PREFIX = PayloadStore::OPTION_PREFIX; /** * Daily task that schedules again the payloads whose event was lost. @@ -44,8 +45,6 @@ final class WpCronDriver implements AsyncDriver */ public const int RECOVERY_BATCH = 500; - private const string ID_PATTERN = '/^[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}$/'; - public function available(): bool { return function_exists('wp_schedule_single_event') && function_exists('add_option'); @@ -56,16 +55,16 @@ public function available(): bool */ public function dispatch(AsyncPayload $payload, int $delay = 0): void { - $option = self::OPTION_PREFIX.$payload->id; - - if (! add_option($option, $payload->toJson(), '', false)) { - throw new AsyncException(sprintf("WP-Cron: the payload of '%s' could not be stored.", $payload->hook)); + try { + PayloadStore::store($payload); + } catch (AsyncException $asyncException) { + throw new AsyncException('WP-Cron: '.lcfirst($asyncException->getMessage()), 0, $asyncException); } $scheduled = wp_schedule_single_event(time() + max(0, $delay), Async::HOOK, [$payload->id], true); if ($scheduled !== true) { - delete_option($option); + PayloadStore::forget($payload->id); $reason = is_object($scheduled) && method_exists($scheduled, 'get_error_message') ? ': '.$scheduled->get_error_message() : '.'; @@ -116,7 +115,7 @@ public static function recover(): int foreach ((array) $rows as $row) { $id = substr((string) $row['option_name'], strlen(self::OPTION_PREFIX)); - if (! self::handles($id) || wp_next_scheduled(Async::HOOK, [$id]) !== false) { + if (! PayloadStore::handles($id) || self::isPending($id)) { continue; } @@ -153,7 +152,7 @@ public static function recover(): int */ public static function handles(string $message): bool { - return preg_match(self::ID_PATTERN, $message) === 1; + return PayloadStore::handles($message); } /** @@ -166,13 +165,19 @@ public static function handles(string $message): bool */ public static function claim(string $id): ?string { - $option = self::OPTION_PREFIX.$id; - $payload = get_option($option, null); + return PayloadStore::claim($id); + } - if (! is_string($payload) || ! delete_option($option)) { - return null; + /** + * Whether an event still carries the identifier: a WP-Cron event, or an + * Action Scheduler action for a payload too long for its arguments. + */ + private static function isPending(string $id): bool + { + if (wp_next_scheduled(Async::HOOK, [$id]) !== false) { + return true; } - return $payload; + return function_exists('as_has_scheduled_action') && as_has_scheduled_action(Async::HOOK, [$id]); } } diff --git a/src/Async/PayloadStore.php b/src/Async/PayloadStore.php new file mode 100644 index 0000000..47e7566 --- /dev/null +++ b/src/Async/PayloadStore.php @@ -0,0 +1,65 @@ +id, $payload->toJson(), '', false)) { + throw new AsyncException(sprintf("The payload of '%s' could not be stored.", $payload->hook)); + } + } + + /** + * Delete a stored payload that will not be queued after all. + */ + public static function forget(string $id): void + { + delete_option(self::OPTION_PREFIX.$id); + } + + /** + * Whether a driver message is the identifier of a stored payload. + */ + public static function handles(string $message): bool + { + return preg_match(self::ID_PATTERN, $message) === 1; + } + + /** + * Take a stored payload out of the database, to run it. + * + * @return string|null The payload as JSON, or null when it was already taken or never existed + */ + public static function claim(string $id): ?string + { + $option = self::OPTION_PREFIX.$id; + $payload = get_option($option, null); + + if (! is_string($payload) || ! delete_option($option)) { + return null; + } + + return $payload; + } +} diff --git a/src/Async/PendingAsync.php b/src/Async/PendingAsync.php index e530a75..b2a76ef 100644 --- a/src/Async/PendingAsync.php +++ b/src/Async/PendingAsync.php @@ -200,7 +200,7 @@ public function onQueue(string $queue): self * * ->capture(fn (int $postId, WP_Post $post) => ['status' => $post->post_status]) * - * @param callable(mixed...): array $capture + * @param callable $capture Receives the hook arguments, returns array */ public function capture(callable $capture): self { @@ -214,7 +214,7 @@ public function capture(callable $capture): self * * ->when(fn (int $postId) => ! wp_is_post_revision($postId)) * - * @param callable(mixed...): bool $condition + * @param callable $condition Receives the hook arguments, returns bool */ public function when(callable $condition): self { diff --git a/tests/Stubs/wordpress.php b/tests/Stubs/wordpress.php index ffa2b9b..6acaeea 100644 --- a/tests/Stubs/wordpress.php +++ b/tests/Stubs/wordpress.php @@ -375,3 +375,51 @@ function wp_cache_delete(int|string $key, string $group = ''): bool return true; } } + +/* + * Action Scheduler, available only when $GLOBALS['as_initialized'] is true. + * Actions are kept in $GLOBALS['as_actions']; $GLOBALS['as_fail'] makes creating one fail. + */ + +if (! class_exists('ActionScheduler', false)) { + class ActionScheduler + { + public static function is_initialized(?string $functionName = null): bool + { + return $GLOBALS['as_initialized'] ?? false; + } + } +} + +if (! function_exists('as_enqueue_async_action')) { + function as_enqueue_async_action(string $hook, array $args = [], string $group = '', bool $unique = false, int $priority = 10): int + { + return as_schedule_single_action(time(), $hook, $args, $group); + } +} + +if (! function_exists('as_schedule_single_action')) { + function as_schedule_single_action(int $timestamp, string $hook, array $args = [], string $group = '', bool $unique = false, int $priority = 10): int + { + if ($GLOBALS['as_fail'] ?? false) { + throw new InvalidArgumentException('ActionScheduler_Action::$args too long.'); + } + + $GLOBALS['as_actions'][] = ['timestamp' => $timestamp, 'hook' => $hook, 'args' => $args, 'group' => $group]; + + return count($GLOBALS['as_actions']); + } +} + +if (! function_exists('as_has_scheduled_action')) { + function as_has_scheduled_action(string $hook, ?array $args = null, string $group = ''): bool + { + foreach ($GLOBALS['as_actions'] ?? [] as $action) { + if ($action['hook'] === $hook && ($args === null || $action['args'] === $args)) { + return true; + } + } + + return false; + } +} diff --git a/tests/Unit/Async/ActionSchedulerDriverTest.php b/tests/Unit/Async/ActionSchedulerDriverTest.php new file mode 100644 index 0000000..57c4ed5 --- /dev/null +++ b/tests/Unit/Async/ActionSchedulerDriverTest.php @@ -0,0 +1,171 @@ + 5, 'blog' => 1, 'locale' => 'fr_FR', 'multisite' => false]; + $GLOBALS['wpdb'] = wp_stub_wpdb(); + $GLOBALS['as_actions'] = []; + $GLOBALS['as_fail'] = false; + $GLOBALS['as_initialized'] = true; + $GLOBALS['incidents'] = []; + RecordingHandler::$calls = []; + RecordingHandler::$failuresLeft = 0; + + Async::flush(); + Async::setDebug(true); + Async::reportUsing(function (string $message): void { + $GLOBALS['incidents'][] = $message; + }); + + $this->action = new Action; +}); + +afterEach(function (): void { + $GLOBALS['as_initialized'] = false; + Async::flush(); +}); + +/** + * Run the queued Action Scheduler actions, as its queue runner does. + */ +function runScheduledActions(): void +{ + for ($index = 0; $index < count($GLOBALS['as_actions']); $index++) { + $action = $GLOBALS['as_actions'][$index]; + if ($action['timestamp'] <= time()) { + Async::receive(...$action['args']); + } + } +} + +it('is the default driver once Action Scheduler is initialised', function (): void { + $this->action->add('save_post', [RecordingHandler::class, 'handle'])->async(); + + wp_stub_fire('save_post', 42); + + expect($GLOBALS['as_actions'])->toHaveCount(1) + ->and($GLOBALS['wp_cron_events'])->toBe([]) + ->and(AsyncPayload::fromJson($GLOBALS['as_actions'][0]['args'][0])->driver)->toBe('action-scheduler'); +}); + +it('falls back to WP-Cron while Action Scheduler is not initialised yet', function (): void { + $GLOBALS['as_initialized'] = false; + $this->action->add('save_post', [RecordingHandler::class, 'handle'])->async(); + + wp_stub_fire('save_post', 42); + + expect($GLOBALS['as_actions'])->toBe([]) + ->and(array_column($GLOBALS['wp_cron_events'], 'hook'))->toContain(Async::HOOK); +}); + +it('carries the payload whole in the action, in the pollora group by default', function (): void { + $this->action->add('save_post', [RecordingHandler::class, 'handle'])->async(); + + wp_stub_fire('save_post', 42); + + $action = $GLOBALS['as_actions'][0]; + expect($action['hook'])->toBe(Async::HOOK) + ->and($action['group'])->toBe(ActionSchedulerDriver::DEFAULT_GROUP) + ->and(AsyncPayload::fromJson($action['args'][0])->arguments)->toBe([42]) + ->and($GLOBALS['wp_options'])->toBe([]); +}); + +it('uses the queue name as the group', function (): void { + $this->action->add('save_post', [RecordingHandler::class, 'handle'])->async()->onQueue('integrations'); + + wp_stub_fire('save_post', 42); + + expect($GLOBALS['as_actions'][0]['group'])->toBe('integrations'); +}); + +it('schedules a delayed action for later', function (): void { + $this->action->add('save_post', [RecordingHandler::class, 'handle'])->async()->delay(120); + $before = time(); + + wp_stub_fire('save_post', 42); + + expect($GLOBALS['as_actions'][0]['timestamp'])->toBeGreaterThanOrEqual($before + 120); +}); + +it('runs the handler when Action Scheduler runs the action', function (): void { + $this->action->add('save_post', [RecordingHandler::class, 'handle'])->async(); + wp_stub_fire('save_post', 42); + + runScheduledActions(); + + expect(RecordingHandler::$calls[0]['arguments'][0])->toBe(42); +}); + +it('keeps a payload too long for Action Scheduler apart and runs it once', function (): void { + $this->action->add('save_post', [RecordingHandler::class, 'handle'])->async() + ->capture(fn (): array => ['report' => str_repeat('x', 9000)]); + + wp_stub_fire('save_post', 42); + [$id] = $GLOBALS['as_actions'][0]['args']; + + expect(PayloadStore::handles($id))->toBeTrue() + ->and($GLOBALS['wp_options'][PayloadStore::OPTION_PREFIX.$id]['autoload'])->toBeFalse() + ->and(wp_next_scheduled(WpCronDriver::RECOVERY_HOOK))->not->toBeFalse(); + + runScheduledActions(); + runScheduledActions(); + + expect(RecordingHandler::$calls)->toHaveCount(1) + ->and(strlen(RecordingHandler::$calls[0]['arguments'][1]->get('report')))->toBe(9000) + ->and($GLOBALS['wp_options'])->toBe([]); +}); + +it('leaves a payload kept apart alone during recovery while its action is pending', function (): void { + $this->action->add('save_post', [RecordingHandler::class, 'handle'])->async()->delay(7200) + ->capture(fn (): array => ['report' => str_repeat('x', 9000)]); + wp_stub_fire('save_post', 42); + [$id] = $GLOBALS['as_actions'][0]['args']; + $option = PayloadStore::OPTION_PREFIX.$id; + $stored = json_decode($GLOBALS['wp_options'][$option]['value'], true); + $stored['origin']['dispatchedAt'] = gmdate(DATE_ATOM, time() - 2 * WpCronDriver::RECOVERY_GRACE); + $GLOBALS['wp_options'][$option]['value'] = json_encode($stored); + + expect(WpCronDriver::recover())->toBe(0); +}); + +it('retries through Action Scheduler', function (): void { + RecordingHandler::$failuresLeft = 1; + $this->action->add('save_post', [RecordingHandler::class, 'flaky'])->async()->tries(2)->backoff(0); + wp_stub_fire('save_post', 42); + + runScheduledActions(); + + expect($GLOBALS['as_actions'])->toHaveCount(2) + ->and(array_column(RecordingHandler::$calls, 'arguments'))->toBe([[42, 1], [42, 2]]); +}); + +it('fails, and forgets the payload kept apart, when Action Scheduler refuses the action', function (): void { + $GLOBALS['as_fail'] = true; + $this->action->add('save_post', [RecordingHandler::class, 'handle'])->async() + ->capture(fn (): array => ['report' => str_repeat('x', 9000)]); + + expect(fn () => wp_stub_fire('save_post', 42)) + ->toThrow(AsyncException::class, "Action Scheduler: the execution of 'save_post' could not be queued: ActionScheduler_Action::\$args too long."); + expect($GLOBALS['wp_options'])->toBe([]); +}); + +it('is unavailable without Action Scheduler', function (): void { + $GLOBALS['as_initialized'] = false; + + expect((new ActionSchedulerDriver)->available())->toBeFalse(); +}); diff --git a/tests/Unit/Async/AsyncActionTest.php b/tests/Unit/Async/AsyncActionTest.php index cd591af..e42eaa1 100644 --- a/tests/Unit/Async/AsyncActionTest.php +++ b/tests/Unit/Async/AsyncActionTest.php @@ -222,7 +222,7 @@ it('falls back to the default driver and reports, in production, when the requested one is unknown', function (): void { Async::setDebug(false); - Async::extend(Async::DEFAULT_DRIVER, fn (): AsyncDriver => $this->driver); + Async::extend('wp-cron', fn (): AsyncDriver => $this->driver); $this->action->add('save_post', [RecordingHandler::class, 'handle'])->async()->via('rabbitmq'); wp_stub_fire('save_post', 1); diff --git a/tests/Unit/Async/AsyncTest.php b/tests/Unit/Async/AsyncTest.php index c7b3177..1d646eb 100644 --- a/tests/Unit/Async/AsyncTest.php +++ b/tests/Unit/Async/AsyncTest.php @@ -50,7 +50,7 @@ public function dispatch(AsyncPayload $payload, int $delay = 0): void {} it('names the registered drivers when one is unknown', function (): void { Async::extend('custom', fn (): AsyncDriver => asyncTestDriver(true)); - expect(fn () => Async::driver('rabbitmq'))->toThrow(DriverUnavailable::class, 'Registered drivers: custom, wp-cron, sync.'); + expect(fn () => Async::driver('rabbitmq'))->toThrow(DriverUnavailable::class, 'Registered drivers: custom, action-scheduler, wp-cron, sync.'); }); it('refuses a driver unavailable in this request', function (): void { @@ -123,8 +123,29 @@ public function dispatch(AsyncPayload $payload, int $delay = 0): void {} expect(Async::defaultDriver())->toBe('sync'); }); - it('resolves auto to WP-Cron until another mechanism is supported', function (): void { - Async::setDefaultDriver('auto'); + it('resolves auto to WP-Cron when Action Scheduler is not available', function (): void { + $GLOBALS['as_initialized'] = false; + + expect(Async::defaultDriver())->toBe('wp-cron'); + }); + + it('resolves auto to Action Scheduler once it is initialised', function (): void { + $GLOBALS['as_initialized'] = true; + + try { + expect(Async::defaultDriver())->toBe('action-scheduler'); + } finally { + $GLOBALS['as_initialized'] = false; + } + }); + + it('resolves auto through the drivers set for it, in order', function (): void { + Async::extend('queue', fn (): AsyncDriver => asyncTestDriver(true)); + Async::setAutoDrivers(['queue', 'wp-cron']); + + expect(Async::defaultDriver())->toBe('queue'); + + Async::extend('queue', fn (): AsyncDriver => asyncTestDriver(false)); expect(Async::defaultDriver())->toBe('wp-cron'); }); diff --git a/tests/WordPress/ActionSchedulerTest.php b/tests/WordPress/ActionSchedulerTest.php new file mode 100644 index 0000000..628078b --- /dev/null +++ b/tests/WordPress/ActionSchedulerTest.php @@ -0,0 +1,128 @@ +markTestSkipped('The Action Scheduler plugin is not active: wp plugin install action-scheduler --activate'); + } + + global $wpdb; + + as_unschedule_all_actions(Async::HOOK); + $wpdb->query($wpdb->prepare("DELETE FROM {$wpdb->options} WHERE option_name LIKE %s", $wpdb->esc_like(PayloadStore::OPTION_PREFIX).'%')); + delete_option(PolloraHookFixture::RUNS); + delete_option(PolloraHookFixture::FAILURES); + update_option(PolloraHookFixture::DRIVER, 'action-scheduler', false); + wp_unschedule_hook(Async::HOOK); + wp_cache_flush(); +}); + +afterEach(function (): void { + delete_option(PolloraHookFixture::DRIVER); +}); + +/** + * Run the Action Scheduler queue in a separate WordPress process. + */ +function runActionScheduler(): void +{ + $script = tempnam(sys_get_temp_dir(), 'pollora-hook').'.php'; + file_put_contents($script, sprintf("run('WP CLI');\n", var_export(ABSPATH.'wp-load.php', true))); + + try { + $output = (string) shell_exec(escapeshellarg(PHP_BINARY).' '.escapeshellarg($script).' 2>&1'); + // Incidents are reported to the PHP error log, so only PHP errors fail here + expect($output)->not->toMatch('/(Fatal|Warning|Deprecated|Notice):/'); + } finally { + unlink($script); + wp_cache_flush(); + } +} + +/** + * @return list}> + */ +function pendingActions(): array +{ + $actions = []; + + foreach ((array) as_get_scheduled_actions(['hook' => Async::HOOK, 'status' => 'pending', 'per_page' => -1]) as $action) { + $actions[] = ['group' => $action->get_group(), 'args' => $action->get_args()]; + } + + return $actions; +} + +function fixtureRunsOf(string $key): array +{ + return array_values(array_filter(get_option(PolloraHookFixture::RUNS, []), fn (array $run): bool => $run['key'] === $key)); +} + +it('is chosen by auto, carries the payload whole and runs through the Action Scheduler queue', function (): void { + update_option(PolloraHookFixture::DRIVER, 'auto', false); + $postId = wp_insert_post(['post_title' => 'Queued', 'post_status' => 'publish']); + + do_action('pollora_fixture_event', $postId, get_post($postId)); + + $actions = pendingActions(); + expect($actions)->toHaveCount(1) + ->and($actions[0]['group'])->toBe('pollora') + ->and(AsyncPayload::fromJson($actions[0]['args'][0])->driver)->toBe('action-scheduler') + ->and(_get_cron_array())->not->toHaveKey('pollora/async/run'); + + runActionScheduler(); + + expect(fixtureRunsOf('event'))->toHaveCount(1) + ->and(fixtureRunsOf('event')[0]['title'])->toBe('Queued') + ->and(pendingActions())->toBe([]); + + wp_delete_post($postId, true); +}); + +it('uses the queue name as the group', function (): void { + $postId = wp_insert_post(['post_title' => 'Grouped', 'post_status' => 'publish']); + + do_action('pollora_fixture_queued', $postId, get_post($postId)); + + expect(pendingActions()[0]['group'])->toBe('integrations'); + + wp_delete_post($postId, true); +}); + +it('keeps a payload too long for Action Scheduler apart, then runs it once', function (): void { + do_action('pollora_fixture_large', 5); + + [$id] = pendingActions()[0]['args']; + expect(PayloadStore::handles($id))->toBeTrue() + ->and(get_option(PayloadStore::OPTION_PREFIX.$id))->toBeString(); + + runActionScheduler(); + runActionScheduler(); + + expect(fixtureRunsOf('large'))->toBe([['key' => 'large', 'id' => 5, 'length' => 9000]]) + ->and(get_option(PayloadStore::OPTION_PREFIX.$id))->toBeFalse(); +}); + +it('retries a failing handler through Action Scheduler', function (): void { + update_option(PolloraHookFixture::FAILURES, 1, false); + + do_action('pollora_fixture_flaky', 1); + runActionScheduler(); + runActionScheduler(); + + expect(array_column(fixtureRunsOf('flaky'), 'attempt'))->toBe([1, 2]) + ->and(fixtureRunsOf('failed'))->toBe([]) + ->and(pendingActions())->toBe([]); +}); diff --git a/tests/WordPress/Fixtures/async-actions.php b/tests/WordPress/Fixtures/async-actions.php index 87cc5b7..f7bbf42 100644 --- a/tests/WordPress/Fixtures/async-actions.php +++ b/tests/WordPress/Fixtures/async-actions.php @@ -16,6 +16,12 @@ final class PolloraHookFixture { public const string RUNS = 'pollora_hook_fixture_runs'; + /** Default driver of the suite, 'wp-cron' unless a test sets it */ + public const string DRIVER = 'pollora_hook_fixture_driver'; + + /** Failures flaky() still has to throw */ + public const string FAILURES = 'pollora_hook_fixture_failures'; + /** * @param array $data */ @@ -38,6 +44,25 @@ public function event(int $postId, WP_Post $post, AsyncContext $context): void ]); } + public function large(int $id, AsyncContext $context): void + { + self::record('large', ['id' => $id, 'length' => strlen((string) $context->get('report'))]); + } + + public function flaky(int $id, AsyncContext $context): void + { + self::record('flaky', ['attempt' => $context->attempt]); + + wp_cache_delete(self::FAILURES, 'options'); + $failures = (int) get_option(self::FAILURES, 0); + + if ($failures > 0) { + update_option(self::FAILURES, $failures - 1, false); + + throw new RuntimeException('Temporary failure'); + } + } + public function fails(int $id): void { throw new RuntimeException('Fixture failure'); @@ -54,6 +79,16 @@ public function loop(int $id): void Action::add('pollora_fixture_delayed', [PolloraHookFixture::class, 'event'])->async()->delay(3600); Action::add('pollora_fixture_fails', [PolloraHookFixture::class, 'fails'])->async(); Action::add('pollora_fixture_loop', [PolloraHookFixture::class, 'loop'])->async(); +Action::add('pollora_fixture_queued', [PolloraHookFixture::class, 'event'])->async()->onQueue('integrations'); +Action::add('pollora_fixture_large', [PolloraHookFixture::class, 'large'])->async() + ->capture(fn (int $id): array => ['report' => str_repeat('x', 9000)]); +Action::add('pollora_fixture_flaky', [PolloraHookFixture::class, 'flaky'])->async()->tries(2)->backoff(0); + +// WP-Cron unless a test chooses another driver; the POLLORA_ASYNC_DRIVER constant still comes first +add_filter('pollora/hook/async_driver', fn (): string => (string) get_option(PolloraHookFixture::DRIVER, 'wp-cron')); + +// Action Scheduler would otherwise start its queue through a loopback request at shutdown +add_filter('action_scheduler_allow_async_request_runner', '__return_false'); add_action('pollora/async/failed', function (AsyncPayload $payload, Throwable $throwable): void { PolloraHookFixture::record('failed', ['hook' => $payload->hook, 'message' => $throwable->getMessage()]);