From c8232724d1090ae45a8cb02dd10983f7ace5a563 Mon Sep 17 00:00:00 2001 From: Olivier Gorzalka Date: Tue, 13 Oct 2026 17:34:50 +0200 Subject: [PATCH 1/2] feat: laravel schedule driver, and remove the Laravel scheduler bridge The laravel driver queues each event as a job delayed until it is due; the cron option is its record. Removes use_laravel_scheduler, Scheduler, Events, Jobs\JobDispatcher, RegisterSchedulerFiltersUseCase and the wp_events table. Adds pollora:schedule:recover. --- CHANGELOG.md | 3 + config/schedule.php | 34 +- .../Checks/AsyncActionsCheck.php | 4 - src/Providers/PolloraServiceProvider.php | 3 - .../Application/Services/ScheduleStatus.php | 19 + .../RegisterSchedulerFiltersUseCase.php | 66 ---- src/Schedule/Contracts/EventInterface.php | 55 --- .../Contracts/JobDispatcherInterface.php | 22 -- src/Schedule/Contracts/SchedulerInterface.php | 103 ----- src/Schedule/Events/AbstractEvent.php | 139 ------- src/Schedule/Events/RecurringEvent.php | 140 ------- src/Schedule/Events/SingleEvent.php | 26 -- .../Checks/ScheduleDriverCheck.php | 20 +- .../Drivers/LaravelQueueDriver.php | 162 ++++++++ .../Infrastructure/Jobs/RunScheduledEvent.php | 193 +++++++++ .../Services/LateEventRecovery.php | 92 +++++ .../Services/ScheduleFailures.php | 46 +++ src/Schedule/Jobs/JobDispatcher.php | 35 -- src/Schedule/Scheduler.php | 370 ------------------ src/Schedule/SchedulerServiceProvider.php | 66 +--- .../UI/Console/ScheduleRecoverCommand.php | 58 +++ .../Hook/Async/AsyncActionsCheckTest.php | 8 +- tests/Feature/Schedule/LaravelDriverTest.php | 306 +++++++++++++++ .../Feature/Schedule/ScheduleCommandsTest.php | 2 + .../Schedule/SchedulerServiceProviderTest.php | 16 - .../RegisterSchedulerFiltersUseCaseTest.php | 72 ---- 26 files changed, 953 insertions(+), 1107 deletions(-) delete mode 100644 src/Schedule/Application/UseCases/RegisterSchedulerFiltersUseCase.php delete mode 100755 src/Schedule/Contracts/EventInterface.php delete mode 100755 src/Schedule/Contracts/JobDispatcherInterface.php delete mode 100755 src/Schedule/Contracts/SchedulerInterface.php delete mode 100755 src/Schedule/Events/AbstractEvent.php delete mode 100755 src/Schedule/Events/RecurringEvent.php delete mode 100755 src/Schedule/Events/SingleEvent.php create mode 100644 src/Schedule/Infrastructure/Drivers/LaravelQueueDriver.php create mode 100644 src/Schedule/Infrastructure/Jobs/RunScheduledEvent.php create mode 100644 src/Schedule/Infrastructure/Services/LateEventRecovery.php create mode 100644 src/Schedule/Infrastructure/Services/ScheduleFailures.php delete mode 100755 src/Schedule/Jobs/JobDispatcher.php delete mode 100755 src/Schedule/Scheduler.php create mode 100644 src/Schedule/UI/Console/ScheduleRecoverCommand.php create mode 100644 tests/Feature/Schedule/LaravelDriverTest.php delete mode 100644 tests/Unit/Schedule/Application/UseCases/RegisterSchedulerFiltersUseCaseTest.php diff --git a/CHANGELOG.md b/CHANGELOG.md index b49e703a..b3aff6c7 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -18,6 +18,8 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 - `config/schedule.php` (publishable with the `pollora-schedule` tag) chooses the scheduling driver, `wp-cron` (the default) or `action-scheduler`, by `POLLORA_SCHEDULE_DRIVER`, and lists the hooks that stay on WP-Cron whatever the driver. The `cron` option now indexes every driver's events, so `wp_next_scheduled()`, Site Health and WP Crontrol see the ones Action Scheduler keeps, and WP-Cron never runs them. Changing the driver moves nothing by itself - `pollora:schedule:list` shows the scheduled events with their type, frequency and driver (`--type`, `--driver`, `--site`, `--network`, `--json`), read from the `cron` option with no query to a driver; an event on WP-Cron on purpose reads `(protected)` or `(fallback)`, one a former driver keeps reads `(to migrate)`. `pollora:schedule:migrate` moves the pending events from one driver to another (`--from`, `--to`, `--dry-run`), in both directions and safe to run again - `pollora:doctor`, Site Health and an administration notice report a driver that does not exist or is not available (events stay on WP-Cron), and the events a former driver still keeps, with the command that moves them +- The `laravel` scheduling driver (`POLLORA_SCHEDULE_DRIVER=laravel`): every event is a job delayed until it is due, in the application's queue (`database`, Redis or Horizon, SQS), and a worker runs it. There is no table of Pollora's own and no `schedule:run`: the `cron` option is the record of what is scheduled, and cancelling an event takes its entry out of it. A job runs only if the entry still exists with its identifier and time, so a cancelled, moved or duplicated job stops; a recurring event queues its next occurrence once it ran or failed for good, never in the past. `config/schedule.php` gains `laravel.connection`, `queue`, `max_delay` (a job due later than the connection allows, 900 seconds on SQS, wakes and waits again) and `late_after`. On a multisite network the queue is shared, with one worker, and each job runs in the site that scheduled it +- Lost jobs (a flushed Redis, a purged queue) are found again: at most every five minutes, as a request ends and under a lock, and with `php artisan pollora:schedule:recover` (`--network`, `--dry-run`), an event more than `late_after` seconds (600) past its time is queued again with the same identifier, so it never runs twice. `pollora:doctor` and Site Health report late events (no worker), a queue connection that is `sync` or `null`, and a scheduled event that failed for good in the last week ### Changed - `DiscoveryEngine`, `DiscoveryCacheManager`, `InstancePool` and `ReflectionCache` take their dependencies instead of reaching for `app()` and `config()`; reading `structure-discoverer.cache` moved to `LaravelDiscoveryCacheManagerFactory`. `DiscoveryRegistrar::registerFromContainer()` is now `registerInto()`. These are infrastructure classes, outside the public contracts (`Pollora\Discovery\Domain\*`), which do not change @@ -35,6 +37,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 - `Pollora\Attributes\PostType\PostTypeAttribute` and `Pollora\Attributes\Taxonomy\TaxonomyAttribute`, removed in 15.0: no built-in attribute extends them any more; implement `ConfiguresArguments` ### Removed +- **The Laravel scheduling bridge, with no replacement under the same name**: `wordpress.use_laravel_scheduler`, `Pollora\Schedule\Scheduler`, `SchedulerInterface`, `Events\*` (`AbstractEvent`, `SingleEvent`, `RecurringEvent`), `Jobs\JobDispatcher`, `RegisterSchedulerFiltersUseCase`, the `pre_schedule_event`, `pre_option_cron` and related filters it installed, and the `wp_events` table. Setting the option no longer does anything. To keep WP-Cron events on Laravel's queue, set `POLLORA_SCHEDULE_DRIVER=laravel` and run a worker (`php artisan queue:work`) in place of `schedule:run`; events that sat in `wp_events` are not moved (let `#[Schedule]` and your own code schedule them again, then drop the table: `DROP TABLE wp_events`, with the prefix of your database connection). `pollora:doctor` no longer reads the option - `handle()` on `#[Action]`, `#[Filter]`, `#[WpRestRoute]` and `#[WpRestRoute\Method]`, and `Hook` no longer implements `HandlesAttributes`: `HookDiscovery` and `WpRestDiscovery` register hooks and routes themselves. Only the bug below reached `Action::handle()`, and `WpRestRoute::handle()` wrote to a read-only wrapper ### Fixed diff --git a/config/schedule.php b/config/schedule.php index 322b6e65..b59c9175 100644 --- a/config/schedule.php +++ b/config/schedule.php @@ -21,12 +21,44 @@ | are until php artisan pollora:schedule:migrate moves them, and | pollora:doctor reminds you. | - | Supported: "wp-cron", "action-scheduler" + | "laravel" makes each event a job delayed until it is due, in the + | application's queue: it needs a connection that accepts delays (not + | "sync" or "null") and a worker, php artisan queue:work. Nothing else + | runs the events, so without a worker none of them fires. + | + | Supported: "wp-cron", "action-scheduler", "laravel" | */ 'driver' => env('POLLORA_SCHEDULE_DRIVER', 'wp-cron'), + /* + |-------------------------------------------------------------------------- + | The "laravel" driver + |-------------------------------------------------------------------------- + | + | connection: the queue connection of the jobs; null is the application's + | default. On a multisite network every site uses the same one, with a + | single worker. + | + | queue: the queue they wait in; null is the connection's default. + | + | max_delay: the longest delay, in seconds, the connection accepts. A job + | due later waits again when it wakes. null: 900 for SQS, none otherwise. + | + | late_after: an event this many seconds past its time is queued again + | (same identifier, so it never runs twice) when its job was lost. + | Checked every five minutes, or with php artisan pollora:schedule:recover. + | + */ + + 'laravel' => [ + 'connection' => env('POLLORA_SCHEDULE_QUEUE_CONNECTION'), + 'queue' => env('POLLORA_SCHEDULE_QUEUE'), + 'max_delay' => env('POLLORA_SCHEDULE_MAX_DELAY'), + 'late_after' => (int) env('POLLORA_SCHEDULE_LATE_AFTER', 600), + ], + /* |-------------------------------------------------------------------------- | Hooks that stay on WP-Cron diff --git a/src/Hook/Infrastructure/Checks/AsyncActionsCheck.php b/src/Hook/Infrastructure/Checks/AsyncActionsCheck.php index 1ce6440f..11857797 100644 --- a/src/Hook/Infrastructure/Checks/AsyncActionsCheck.php +++ b/src/Hook/Infrastructure/Checks/AsyncActionsCheck.php @@ -179,10 +179,6 @@ private function connectionProblems(): array */ private function cronFix(): string { - if ($this->config->get('wordpress.use_laravel_scheduler')) { - return 'WP-Cron events are Laravel jobs here (wordpress.use_laravel_scheduler): run a queue worker, php artisan queue:work'; - } - return $this->inspector->cronRunsOnPageLoad() ? 'The site gets too few visits to run WP-Cron: add a system cron that requests wp-cron.php every minute' : 'DISABLE_WP_CRON is set (Pollora sets it): add a system cron, e.g. * * * * * curl -s https://example.com/cms/wp-cron.php, or wp cron event run --due-now'; diff --git a/src/Providers/PolloraServiceProvider.php b/src/Providers/PolloraServiceProvider.php index b83b4b7a..efb6d705 100755 --- a/src/Providers/PolloraServiceProvider.php +++ b/src/Providers/PolloraServiceProvider.php @@ -4,7 +4,6 @@ namespace Pollora\Providers; -use Illuminate\Contracts\Bus\Dispatcher; use Illuminate\Support\ServiceProvider; use Log1x\SageDirectives\SageDirectivesServiceProvider; use Pollora\Ability\Infrastructure\Providers\AbilityServiceProvider; @@ -36,7 +35,6 @@ use Pollora\PostType\Infrastructure\Providers\PostTypeServiceProvider; use Pollora\Role\Infrastructure\Providers\RoleServiceProvider; use Pollora\Route\Infrastructure\Providers\RouteServiceProvider; -use Pollora\Schedule\Jobs\JobDispatcher; use Pollora\Schedule\SchedulerServiceProvider; use Pollora\Taxonomy\Infrastructure\Providers\TaxonomyServiceProvider; use Pollora\Theme\Infrastructure\Providers\ThemeServiceProvider; @@ -125,7 +123,6 @@ public function register(): void $this->app->register(RouteServiceProvider::class); $this->app->register(SchedulerServiceProvider::class); - $this->app->singleton(JobDispatcher::class, fn ($app): JobDispatcher => new JobDispatcher($app->make(Dispatcher::class))); // Authentication service provider $this->app->register(AuthServiceProvider::class); diff --git a/src/Schedule/Application/Services/ScheduleStatus.php b/src/Schedule/Application/Services/ScheduleStatus.php index 620fefcb..e6a4bb7b 100644 --- a/src/Schedule/Application/Services/ScheduleStatus.php +++ b/src/Schedule/Application/Services/ScheduleStatus.php @@ -6,6 +6,7 @@ use Illuminate\Contracts\Config\Repository; use Pollora\Schedule\Event; +use Pollora\Schedule\Infrastructure\Drivers\LaravelQueueDriver; use Pollora\Schedule\Schedule; use Pollora\Schedule\Support\CronIndex; @@ -74,6 +75,24 @@ public function toMigrate(): array return $counts; } + /** + * Seconds after which an event of the "laravel" driver counts as late. + */ + public function lateAfter(): int + { + return max(60, (int) $this->config->get('schedule.laravel.late_after', 600)); + } + + /** + * Events of the "laravel" driver that are late: a job lost, or no worker. + */ + public function late(): int + { + $limit = time() - $this->lateAfter(); + + return count(array_filter(Schedule::engine()->index(), static fn (Event $event): bool => $event->driver === LaravelQueueDriver::NAME && $event->timestamp < $limit)); + } + /** * The driver of an event, with what sets it apart from the site's: "wp-cron (protected)", "wp-cron (to migrate)". */ diff --git a/src/Schedule/Application/UseCases/RegisterSchedulerFiltersUseCase.php b/src/Schedule/Application/UseCases/RegisterSchedulerFiltersUseCase.php deleted file mode 100644 index 91784f2a..00000000 --- a/src/Schedule/Application/UseCases/RegisterSchedulerFiltersUseCase.php +++ /dev/null @@ -1,66 +0,0 @@ -registerEventQueryFilters(); - $this->registerEventManagementFilters(); - } - - /** - * Register filters for querying scheduled events. - */ - private function registerEventQueryFilters(): void - { - $filters = [ - 'pre_get_scheduled_event' => 'preGetScheduledEvent', - 'pre_get_ready_cron_jobs' => 'preGetReadyCronJobs', - ]; - - foreach ($filters as $hook => $method) { - $this->filter->add($hook, [$this->scheduler, $method], 10, 5); - } - } - - /** - * Register filters for managing scheduled events. - */ - private function registerEventManagementFilters(): void - { - $filters = [ - 'pre_update_option_cron' => 'preUpdateOptionCron', - 'pre_option_cron' => 'preOptionCron', - 'pre_schedule_event' => 'preScheduleEvent', - 'pre_reschedule_event' => 'preRescheduleEvent', - 'pre_unschedule_event' => 'preUnscheduleEvent', - 'pre_clear_scheduled_hook' => 'preClearScheduledHook', - 'pre_unschedule_hook' => 'preUnscheduleHook', - ]; - - foreach ($filters as $hook => $method) { - $this->filter->add($hook, [$this->scheduler, $method], 10, 5); - } - } -} diff --git a/src/Schedule/Contracts/EventInterface.php b/src/Schedule/Contracts/EventInterface.php deleted file mode 100755 index a3ffb23a..00000000 --- a/src/Schedule/Contracts/EventInterface.php +++ /dev/null @@ -1,55 +0,0 @@ -hook = $event->hook; - $this->timestamp = $event->timestamp; - $this->args = $event->args; - } - } - - /** - * Get the event hook name. - */ - public function getHook(): string - { - return $this->hook; - } - - /** - * Get the event timestamp. - */ - public function getTimestamp(): int - { - return $this->timestamp; - } - - /** - * Get the event arguments. - */ - public function getArgs(): array - { - return $this->args; - } - - /** - * Create and persist a new job instance. - * - * @param object $event WordPress event object - */ - public static function createJob(object $event): self - { - $job = new static($event); - $dispatcher = resolve(JobDispatcher::class); - $jobId = $dispatcher->dispatch($job); - $job->saveToDatabase($jobId); - - return $job; - } - - /** - * Save the event to the database. - * - * @param int $jobId The queue job ID - */ - protected function saveToDatabase($jobId) - { - DB::table('wp_events')->insert([ - 'hook' => $this->hook, - 'args' => json_encode($this->args), - 'schedule' => $this->schedule, - 'interval' => $this->interval, - 'is_recurring' => ! is_null($this->schedule), - 'timestamp' => $this->timestamp, - 'job_id' => $jobId, - 'created_at' => now(), - 'updated_at' => now(), - ]); - } - - /** - * Delete the event from the database. - */ - protected function deleteEvent() - { - DB::table('wp_events')->where('job_id', $this->job->getJobId())->delete(); - } - - /** - * Handle the event execution. - */ - abstract public function handle(): void; -} diff --git a/src/Schedule/Events/RecurringEvent.php b/src/Schedule/Events/RecurringEvent.php deleted file mode 100755 index cc49d9ec..00000000 --- a/src/Schedule/Events/RecurringEvent.php +++ /dev/null @@ -1,140 +0,0 @@ -schedule = $event->schedule; - $this->interval = $event->interval; - } - } - - /** - * Create and persist a new recurring job instance. - * - * @param object $event WordPress event object\n */ - public static function createJob(object $event): self - { - $job = new static($event); - $job->saveToDatabase(); - - return $job; - } - - /** - * Save the recurring job to the database. - * - * @param int|null $jobId Optional job ID - */ - protected function saveToDatabase($jobId = null): void - { - DB::table('wp_events')->insert([ - 'hook' => $this->hook, - 'args' => json_encode($this->args), - 'schedule' => $this->schedule, - 'interval' => $this->interval, - 'created_at' => now(), - 'updated_at' => now(), - ]); - } - - /** - * Schedule all recurring events in the Laravel scheduler. - * - * @param Schedule $schedule Laravel schedule instance - */ - public static function scheduleAllEvents(Schedule $schedule): void - { - $events = DB::table('wp_events')->whereNotNull('schedule')->get(); - - foreach ($events as $event) { - $schedule->call(function () use ($event): void { - do_action_ref_array($event->hook, json_decode((string) $event->args, true)); - })->cron(self::getCronExpression($event->schedule, $event->interval)); - } - } - - /** - * Convert WordPress schedule to cron expression. - * - * @param string $schedule WordPress schedule name - * @param int|null $interval Custom interval in seconds - * @return string Cron expression - */ - public static function getCronExpression(string $schedule, ?int $interval): string - { - $schedules = wp_get_schedules(); - - if (isset($schedules[$schedule])) { - $interval = $schedules[$schedule]['interval']; - } - - return match ($schedule) { - 'minutly' => '* * * * *', - 'hourly' => '0 * * * *', - 'twicedaily' => '0 */12 * * *', - 'daily' => '0 0 * * *', - 'weekly' => '0 0 * * 0', - default => self::calculateCustomInterval($interval), - }; - } - - /** - * Handle the recurring event execution. - */ - public function handle(): void - { - do_action_ref_array($this->hook, $this->args); - } - - /** - * Calculate cron expression for custom interval. - * - * @param int|null $interval Interval in seconds - * @return string Cron expression - */ - private static function calculateCustomInterval(?int $interval): string - { - if ($interval === null || $interval === 0) { - return '0 0 * * *'; - } - - $minutes = $interval / 60; - - if ($minutes < 60) { - return sprintf('*/%s * * * *', $minutes); - } - - if ($minutes == 60) { - return '0 * * * *'; - } - - if ($minutes % 60 === 0) { - $hours = $minutes / 60; - - return sprintf('0 */%s * * *', $hours); - } - - return sprintf('*/%s * * * *', $minutes); - } -} diff --git a/src/Schedule/Events/SingleEvent.php b/src/Schedule/Events/SingleEvent.php deleted file mode 100755 index 5343c16f..00000000 --- a/src/Schedule/Events/SingleEvent.php +++ /dev/null @@ -1,26 +0,0 @@ -hook, $this->args); - $this->deleteEvent(); - } -} diff --git a/src/Schedule/Infrastructure/Checks/ScheduleDriverCheck.php b/src/Schedule/Infrastructure/Checks/ScheduleDriverCheck.php index 125bcf99..20948454 100644 --- a/src/Schedule/Infrastructure/Checks/ScheduleDriverCheck.php +++ b/src/Schedule/Infrastructure/Checks/ScheduleDriverCheck.php @@ -8,6 +8,8 @@ use Pollora\Doctor\Domain\Enums\RunContext; use Pollora\Doctor\Domain\Models\CheckResult; use Pollora\Schedule\Application\Services\ScheduleStatus; +use Pollora\Schedule\Infrastructure\Drivers\LaravelQueueDriver; +use Pollora\Schedule\Infrastructure\Services\ScheduleFailures; /** * Scheduled events are where the configuration says they should be. @@ -47,10 +49,12 @@ public function run(RunContext $context): CheckResult if ($this->status->unknown()) { $problems[] = sprintf('config/schedule.php asks for the driver "%s", which does not exist: every event goes to WP-Cron', $this->status->configured()); - $fix = 'Set POLLORA_SCHEDULE_DRIVER (config/schedule.php driver) to wp-cron or action-scheduler'; + $fix = 'Set POLLORA_SCHEDULE_DRIVER (config/schedule.php driver) to wp-cron, action-scheduler or laravel'; } elseif ($this->status->unavailable()) { $problems[] = sprintf('The "%s" driver is not available, so events stay on WP-Cron', $this->status->active()); - $fix = 'Activate a plugin that bundles Action Scheduler (WooCommerce does), or set the driver back to wp-cron'; + $fix = $this->status->active() === LaravelQueueDriver::NAME + ? 'Point the queue at a connection a worker runs (database, redis, sqs…), never sync or null: QUEUE_CONNECTION or POLLORA_SCHEDULE_QUEUE_CONNECTION' + : 'Activate a plugin that bundles Action Scheduler (WooCommerce does), or set the driver back to wp-cron'; } foreach ($this->status->toMigrate() as $driver => $count) { @@ -58,6 +62,18 @@ public function run(RunContext $context): CheckResult $fix ??= sprintf('php artisan pollora:schedule:migrate --from=%s --to=%s (--dry-run shows what would move)', $driver, $this->status->active()); } + $late = $this->status->late(); + + if ($late > 0) { + $problems[] = sprintf('%d scheduled event(s) are more than %d minutes late: no queue worker is running them', $late, intdiv($this->status->lateAfter(), 60)); + $fix ??= 'Run a queue worker (php artisan queue:work), or send lost jobs again with php artisan pollora:schedule:recover'; + } + + foreach (array_slice(ScheduleFailures::recent(), -3) as $failure) { + $problems[] = sprintf('"%s" failed for good on %s: %s', $failure['hook'], gmdate('Y-m-d H:i', $failure['time']).' UTC', $failure['message']); + $fix ??= 'See php artisan queue:failed; the event is out of the schedule if it was a single one'; + } + if ($problems === []) { return CheckResult::ok(sprintf('Scheduled events are kept by "%s".', $this->status->active())); } diff --git a/src/Schedule/Infrastructure/Drivers/LaravelQueueDriver.php b/src/Schedule/Infrastructure/Drivers/LaravelQueueDriver.php new file mode 100644 index 00000000..d7c631bc --- /dev/null +++ b/src/Schedule/Infrastructure/Drivers/LaravelQueueDriver.php @@ -0,0 +1,162 @@ + 900]; + + public function __construct( + private Dispatcher $dispatcher, + private Repository $config, + ) {} + + public function available(): bool + { + $driver = $this->config->get(sprintf('queue.connections.%s.driver', $this->connection())); + + return is_string($driver) && ! in_array($driver, ['sync', 'null'], true); + } + + public function placeIdentified(Event $event): bool + { + $this->send(RunScheduledEvent::for($event)); + + return true; + } + + public function once(string $hook, int $timestamp, array $args = []): bool + { + return $this->place(new Event($hook, $timestamp, null, $args)); + } + + public function recurring(string $hook, int $timestamp, string $recurrence, array $args = []): bool + { + $interval = (int) ((function_exists('wp_get_schedules') ? wp_get_schedules() : [])[$recurrence]['interval'] ?? 0); + + return $this->place(new Event($hook, $timestamp, $recurrence, $args, $interval > 0 ? $interval : null)); + } + + /** + * What the engine removes from the index is cancelled: a job that finds no entry stops. + */ + public function cancel(string $hook, array $args = []): int + { + return count(array_filter($this->events(), static fn (Event $event): bool => $event->hook === $hook && $event->args === $args)); + } + + public function clear(string $hook): int + { + return count(array_filter($this->events(), static fn (Event $event): bool => $event->hook === $hook)); + } + + public function next(string $hook, array $args = []): ?int + { + foreach ($this->events() as $event) { + if ($event->hook === $hook && $event->args === $args) { + return $event->timestamp; + } + } + + return null; + } + + public function events(): array + { + return CronIndex::events(self::NAME); + } + + /** + * Queue the job of an event, delayed until it is due, or by the longest delay the connection accepts: the job then waits again. + */ + public function send(RunScheduledEvent $job): void + { + $delay = max(0, $job->timestamp - time()); + $longest = $this->longestDelay(); + + if ($longest !== null && $delay > $longest) { + $delay = $longest; + } + + $job->onConnection($this->connection())->onQueue($this->queue()); + + if ($delay > 0) { + $job->delay($delay); + } + + $this->dispatcher->dispatch($job); + } + + /** + * Seconds after which an event that has not run is late, and the recovery sends it again. + */ + public function lateAfter(): int + { + return max(60, (int) $this->config->get('schedule.laravel.late_after', 600)); + } + + /** + * Where the identifier of a placement is not known to the engine: record the event and queue it. + */ + private function place(Event $event): bool + { + $event = $event->withId(bin2hex(random_bytes(8))); + + CronIndex::record($event, self::NAME); + $this->placeIdentified($event); + + return true; + } + + private function connection(): string + { + $connection = $this->config->get('schedule.laravel.connection'); + + return is_string($connection) && $connection !== '' ? $connection : (string) $this->config->get('queue.default', 'sync'); + } + + private function queue(): ?string + { + $queue = $this->config->get('schedule.laravel.queue'); + + return is_string($queue) && $queue !== '' ? $queue : null; + } + + private function longestDelay(): ?int + { + $configured = $this->config->get('schedule.laravel.max_delay'); + + if (is_numeric($configured) && (int) $configured > 0) { + return (int) $configured; + } + + $driver = $this->config->get(sprintf('queue.connections.%s.driver', $this->connection())); + + return is_string($driver) ? (self::MAX_DELAYS[$driver] ?? null) : null; + } +} diff --git a/src/Schedule/Infrastructure/Jobs/RunScheduledEvent.php b/src/Schedule/Infrastructure/Jobs/RunScheduledEvent.php new file mode 100644 index 00000000..7eaa1664 --- /dev/null +++ b/src/Schedule/Infrastructure/Jobs/RunScheduledEvent.php @@ -0,0 +1,193 @@ + $args + */ + public function __construct( + public readonly string $hook, + public readonly array $args, + public readonly int $timestamp, + public readonly string $id, + public readonly ?string $recurrence = null, + public readonly ?int $interval = null, + public readonly int $site = 1, + ) {} + + public static function for(Event $event): self + { + return new self( + $event->hook, + $event->args, + $event->timestamp, + (string) $event->id, + $event->recurrence, + $event->interval, + function_exists('get_current_blog_id') ? get_current_blog_id() : 1, + ); + } + + public function handle(LaravelQueueDriver $driver): void + { + $this->inSite(function () use ($driver): void { + if (! $this->entryHolds()) { + return; + } + + // The delay was cut short for a connection with a ceiling: wait again + if ($this->timestamp - time() > 1) { + $driver->send($this->again()); + + return; + } + + // One run at a time: a copy sent again while this one is still working finds the lock taken + $lock = Cache::lock('pollora_schedule_run_'.$this->id, 3600); + + if (! $lock->get()) { + return; + } + + try { + do_action_ref_array($this->hook, $this->args); + $this->conclude($driver); + } finally { + $lock->release(); + } + }); + } + + /** + * Retries are over: a recurring event goes on with its next occurrence, a single one leaves the index, and the doctor is told. + */ + public function failed(Throwable $exception): void + { + $this->inSite(function () use ($exception): void { + if (! $this->entryHolds()) { + return; + } + + ScheduleFailures::record($this->hook, $exception->getMessage()); + $this->conclude(resolve(LaravelQueueDriver::class)); + }); + } + + /** + * Name shown by Horizon and in the failed jobs table. + */ + public function displayName(): string + { + return 'pollora/schedule '.$this->hook; + } + + /** + * Tags for Horizon. + * + * @return list + */ + public function tags(): array + { + return ['pollora/schedule', $this->hook]; + } + + /** + * Whether the index still has this event, at this time, under this identifier. + */ + private function entryHolds(): bool + { + // A worker lives across jobs: what WordPress cached in memory, the cron option included, is from before + if (function_exists('wp_cache_flush_runtime')) { + wp_cache_flush_runtime(); + } + + foreach (CronIndex::events(LaravelQueueDriver::NAME) as $event) { + if ($event->hook === $this->hook && $event->args === $this->args) { + return $event->id === $this->id && $event->timestamp === $this->timestamp; + } + } + + return false; + } + + /** + * What follows a run: the next occurrence of a recurring event (never in the past), or the end of a single one. + */ + private function conclude(LaravelQueueDriver $driver): void + { + if ($this->recurrence === null || $this->interval === null || $this->interval <= 0) { + CronIndex::forget($this->hook, $this->args, LaravelQueueDriver::NAME); + + return; + } + + $next = $this->timestamp + $this->interval; + + if ($next <= time()) { + $next += (int) ceil((time() - $next + 1) / $this->interval) * $this->interval; + } + + $event = new Event($this->hook, $next, $this->recurrence, $this->args, $this->interval, LaravelQueueDriver::NAME, bin2hex(random_bytes(8))); + + CronIndex::record($event, LaravelQueueDriver::NAME); + $driver->send(self::for($event)); + } + + /** + * The same promise as a new job: the one a worker holds carries its queue message. + */ + private function again(): self + { + return new self($this->hook, $this->args, $this->timestamp, $this->id, $this->recurrence, $this->interval, $this->site); + } + + /** + * Run in the site that scheduled the event: on a multisite network the queue and its worker are shared. + */ + private function inSite(callable $run): void + { + $switch = function_exists('is_multisite') && is_multisite() && $this->site !== get_current_blog_id(); + + if ($switch) { + switch_to_blog($this->site); + } + + try { + $run(); + } finally { + if ($switch) { + restore_current_blog(); + } + } + } +} diff --git a/src/Schedule/Infrastructure/Services/LateEventRecovery.php b/src/Schedule/Infrastructure/Services/LateEventRecovery.php new file mode 100644 index 00000000..7e06b311 --- /dev/null +++ b/src/Schedule/Infrastructure/Services/LateEventRecovery.php @@ -0,0 +1,92 @@ +get()) { + return 0; + } + + try { + $sent = 0; + + foreach ($this->late() as $event) { + if ($event->id === null) { + $event = $event->withId(bin2hex(random_bytes(8))); + CronIndex::record($event, LaravelQueueDriver::NAME); + } + + // Sent again, then left alone for as long as it takes to count as late again + if (! $force && ! Cache::add('pollora_schedule_resent_'.$event->id, 1, $this->driver->lateAfter())) { + continue; + } + + $this->driver->send(RunScheduledEvent::for($event)); + $sent++; + } + + return $sent; + } finally { + $lock->release(); + } + } + + /** + * The events of the index more than late_after seconds past their time; one without an identifier gets one when it is sent again. + * + * @return list + */ + public function late(): array + { + $late = []; + + foreach (CronIndex::events(LaravelQueueDriver::NAME) as $event) { + if ($event->timestamp >= time() - $this->driver->lateAfter()) { + break; + } + + $late[] = $event; + } + + return $late; + } +} diff --git a/src/Schedule/Infrastructure/Services/ScheduleFailures.php b/src/Schedule/Infrastructure/Services/ScheduleFailures.php new file mode 100644 index 00000000..e5ca27c4 --- /dev/null +++ b/src/Schedule/Infrastructure/Services/ScheduleFailures.php @@ -0,0 +1,46 @@ + $hook, 'time' => time(), 'message' => mb_substr($message, 0, 300)]; + + update_option(self::OPTION, array_slice($failures, -self::KEEP), false); + } + + /** + * Failures of the last week, oldest first. + * + * @return list + */ + public static function recent(): array + { + $stored = get_option(self::OPTION, []); + $recent = []; + + foreach (is_array($stored) ? $stored : [] as $failure) { + if (is_array($failure) && is_string($failure['hook'] ?? null) && is_int($failure['time'] ?? null) && $failure['time'] > time() - self::WEEK) { + $recent[] = ['hook' => $failure['hook'], 'time' => $failure['time'], 'message' => (string) ($failure['message'] ?? '')]; + } + } + + return $recent; + } +} diff --git a/src/Schedule/Jobs/JobDispatcher.php b/src/Schedule/Jobs/JobDispatcher.php deleted file mode 100755 index f3fc7fd1..00000000 --- a/src/Schedule/Jobs/JobDispatcher.php +++ /dev/null @@ -1,35 +0,0 @@ -dispatcher->dispatch($job); - } -} diff --git a/src/Schedule/Scheduler.php b/src/Schedule/Scheduler.php deleted file mode 100755 index c4c59006..00000000 --- a/src/Schedule/Scheduler.php +++ /dev/null @@ -1,370 +0,0 @@ -processCronDifferences($old_value, $value); - - return $old_value; - } - - /** - * Retrieve the cron option. - * - * @param mixed $value Current option value - * @return array The cron jobs array - */ - public function preOptionCron($value): array - { - if ($value !== false) { - return $value; - } - - return $this->generateCronArray(); - } - - /** - * Schedule a new event. - * - * @param mixed $pre Pre-filtered value - * @param object $event Event to schedule - * @param bool $wp_error Whether to return WP_Error on failure - * @return AbstractEvent|WP_Error|null The scheduled job or error - */ - public function preScheduleEvent($pre, object $event, bool $wp_error): AbstractEvent|WP_Error|null - { - if ($pre !== null) { - return $pre; - } - - try { - $job = $event->schedule ? new RecurringEvent($event) : new SingleEvent($event); - - return $job->createJob($event); - } catch (\Throwable $throwable) { - return $wp_error ? new WP_Error('schedule_error', $throwable->getMessage()) : null; - } - } - - /** - * Reschedule an event. - * - * @param mixed $pre Pre-filtered value - * @param object $event Event to reschedule - * @param bool $wp_error Whether to return WP_Error on failure - * @return AbstractEvent|WP_Error|null The rescheduled job or error - */ - public function preRescheduleEvent($pre, object $event, bool $wp_error): AbstractEvent|WP_Error|null - { - if ($pre !== null) { - return $pre; - } - - try { - $job = new RecurringEvent($event); - - return $job->createJob($event); - } catch (\Throwable $throwable) { - return $wp_error ? new WP_Error('reschedule_error', $throwable->getMessage()) : null; - } - } - - /** - * Unschedule an event. - * - * @param mixed $pre Pre-filtered value - * @param int $timestamp Event timestamp - * @param string $hook Event hook - * @param array $args Event arguments - * @param bool $wp_error Whether to return WP_Error on failure - * @return bool|WP_Error Whether the event was unscheduled - */ - public function preUnscheduleEvent($pre, int $timestamp, string $hook, array $args, bool $wp_error): bool|WP_Error - { - if ($pre !== null) { - return $pre; - } - - try { - $deleted = DB::table('wp_events') - ->where('hook', $hook) - ->where('timestamp', $timestamp) - ->where('args', json_encode($args)) - ->delete(); - - if ($deleted) { - $this->deleteAssociatedJob($hook, $timestamp, $args); - } - - return $deleted > 0; - } catch (\Throwable $throwable) { - return $wp_error ? new WP_Error('unschedule_error', $throwable->getMessage()) : false; - } - } - - /** - * Clear all scheduled hooks. - * - * @param mixed $pre Pre-filtered value - * @param string $hook Hook to clear - * @param array|null $args Arguments to match - * @param bool $wp_error Whether to return WP_Error on failure - * @return int|WP_Error The number of events cleared - */ - public function preClearScheduledHook($pre, string $hook, ?array $args, bool $wp_error): int|WP_Error - { - if ($pre !== null) { - return $pre; - } - - try { - $query = DB::table('wp_events')->where('hook', $hook); - - if ($args !== null) { - $query->where('args', json_encode($args)); - } - - $events = $query->get(); - $count = $events->count(); - - foreach ($events as $event) { - $this->deleteAssociatedJob($event->hook, $event->timestamp, json_decode((string) $event->args, true)); - } - - $query->delete(); - - return $count; - } catch (\Throwable $throwable) { - return $wp_error ? new WP_Error('clear_hook_error', $throwable->getMessage()) : 0; - } - } - - protected function deleteAssociatedJob(string $hook, int $timestamp, array $args) - { - $jobId = DB::table('wp_events') - ->where('hook', $hook) - ->where('timestamp', $timestamp) - ->where('args', json_encode($args)) - ->value('job_id'); - - if ($jobId) { - DB::table('jobs')->where('id', $jobId)->delete(); - } - } - - /** - * Unschedule all events attached to a specific hook. - * - * @param mixed $pre Pre-filtered value - * @param string $hook Hook to unschedule - * @param bool $wp_error Whether to return WP_Error on failure - * @return int|WP_Error The number of events unscheduled - */ - public function preUnscheduleHook($pre, string $hook, bool $wp_error): int|WP_Error - { - return $this->preClearScheduledHook($pre, $hook, null, $wp_error); - } - - /** - * Retrieve a scheduled event. - * - * @param mixed $pre Pre-filtered value - * @param string $hook Event hook - * @param array $args Event arguments - * @param int|null $timestamp Unix timestamp (UTC) of the event. Null to retrieve next scheduled event. - * @return object|false The event object or false if not found - */ - public function preGetScheduledEvent($pre, string $hook, array $args, ?int $timestamp): object|false - { - if ($pre !== null) { - return $pre; - } - - $query = DB::table('wp_events') - ->where('hook', $hook); - - if ($args !== []) { - $query->where('args', json_encode($args)); - } - - if ($timestamp !== null) { - $query->where('timestamp', $timestamp); - } - - $event = $query->first(); - - if (! $event) { - return false; - } - - return $this->convertDbEventToWpEvent($event); - } - - protected function convertDbEventToWpEvent(object $event): object - { - return (object) [ - 'hook' => $event->hook, - 'timestamp' => $event->schedule ? $this->getNextRunTime($event->schedule, $event->interval) : $event->timestamp, - 'schedule' => $event->schedule, - 'args' => json_decode($event->args, true), - 'interval' => $event->interval, - ]; - } - - /** - * Generate a unique job ID. - * - * @param int|null $timestamp Event timestamp - * @param string $hook Event hook - * @param array $args Event arguments - * @return string The generated job ID - */ - protected function generateJobId(?int $timestamp, string $hook, array $args): string - { - return md5(serialize([$timestamp, $hook, $args])); - } - - /** - * Convert a job to a WordPress event object. - * - * @param object $job The job to convert - * @return object The converted event object - */ - protected function convertJobToEvent(object $job): object - { - $event = new \stdClass; - $event->hook = $job->wp_hook; - $event->timestamp = $job->available_at; - $event->args = json_decode($job->wp_args, true); - $event->schedule = $job->wp_schedule ?? false; - $event->interval = $job->wp_interval ?? null; - - return $event; - } - - /** - * Process the differences between old and new cron arrays. - * - * @param array $oldCrons The old cron array - * @param array $newCrons The new cron array - */ - protected function processCronDifferences(array $oldCrons, array $newCrons): void - { - // Implementation to handle differences (add, update, delete jobs) - } - - /** - * Generate a cron array from the current queue jobs. - * - * @return array The generated cron array - */ - protected function generateCronArray(): array - { - $jobs = $this->getAllJobs(); - - return $this->convertJobsToWordPressCronArray($jobs); - } - - /** - * Get all jobs from the queue. - */ - protected function getAllJobs(): array - { - return DB::table('wp_events') - ->orderBy('timestamp', 'asc') - ->get() - ->map(fn (object $job): object => $this->convertDbEventToWpEvent($job)) - ->all(); - } - - /** - * Retrieve cron jobs ready to be run. - * - * @param mixed $pre Pre-filtered value - * @return array The array of ready cron jobs - */ - public function preGetReadyCronJobs($pre): array - { - if ($pre !== null) { - return $pre; - } - - return []; - } - - /** - * Convert an array of jobs to a WordPress cron array format. - * - * @param array $jobs The jobs to convert - * @return array The converted cron array - */ - protected function convertJobsToWordPressCronArray(array $jobs): array - { - $crons = []; - - foreach ($jobs as $job) { - if (! $job->hook) { - continue; - } - - $timestamp = $job->timestamp; - $hook = $job->hook; - $key = md5(serialize($job->args)); - - $crons[$timestamp][$hook][$key] = [ - 'schedule' => $job->schedule, - 'args' => $job->args, - ]; - - if ($job->interval) { - $crons[$timestamp][$hook][$key]['interval'] = $job->interval; - } - } - - ksort($crons, SORT_NUMERIC); - $crons['version'] = 2; - - return $crons; - } - - /** - * Get the next run time for a scheduled event. - * - * @param string $schedule WordPress schedule name - * @param int|null $interval Custom interval in seconds - * @return int Next run timestamp - */ - protected function getNextRunTime(string $schedule, ?int $interval): int - { - $cron = RecurringEvent::getCronExpression($schedule, $interval); - $cron = new CronExpression($cron); - - return $cron->getNextRunDate()->getTimestamp(); - } -} diff --git a/src/Schedule/SchedulerServiceProvider.php b/src/Schedule/SchedulerServiceProvider.php index 4750e57c..bc1e25fd 100755 --- a/src/Schedule/SchedulerServiceProvider.php +++ b/src/Schedule/SchedulerServiceProvider.php @@ -4,30 +4,24 @@ namespace Pollora\Schedule; -use Illuminate\Console\Scheduling\Schedule; use Illuminate\Contracts\Config\Repository; use Illuminate\Contracts\Foundation\Application; -use Illuminate\Database\QueryException; use Illuminate\Support\ServiceProvider; -use Orchestra\Testbench\TestCase; use Pollora\Doctor\Infrastructure\Providers\DoctorServiceProvider; -use Pollora\Hook\Domain\Contract\Filter; use Pollora\Schedule\Application\Services\ScheduleStatus; -use Pollora\Schedule\Application\UseCases\RegisterSchedulerFiltersUseCase; -use Pollora\Schedule\Contracts\SchedulerInterface; -use Pollora\Schedule\Events\RecurringEvent; use Pollora\Schedule\Infrastructure\Checks\ScheduleDriverCheck; +use Pollora\Schedule\Infrastructure\Drivers\LaravelQueueDriver; +use Pollora\Schedule\Infrastructure\Services\LateEventRecovery; use Pollora\Schedule\Infrastructure\Services\MigrationNotice; use Pollora\Schedule\Infrastructure\Services\ScheduleDiscovery; use Pollora\Schedule\Schedule as ScheduleApi; use Pollora\Schedule\UI\Console\ScheduleListCommand; use Pollora\Schedule\UI\Console\ScheduleMigrateCommand; +use Pollora\Schedule\UI\Console\ScheduleRecoverCommand; /** - * Service provider for WordPress cron scheduler functionality. - * - * Registers and bootstraps the scheduler services, including filters - * and recurring event scheduling. + * Service provider for scheduled events: the driver of the site, the commands + * and the checks around it. */ class SchedulerServiceProvider extends ServiceProvider { @@ -37,16 +31,16 @@ class SchedulerServiceProvider extends ServiceProvider public function register(): void { $this->registerDomainContracts(); - $this->registerUseCases(); $this->mergeConfigFrom(__DIR__.'/../../config/schedule.php', 'schedule'); $this->app->singleton(ScheduleStatus::class, fn (Application $app): ScheduleStatus => new ScheduleStatus($app->make(Repository::class))); ScheduleApi::useContainer($this->app); + ScheduleApi::extend(LaravelQueueDriver::NAME, fn (): LaravelQueueDriver => $this->app->make(LaravelQueueDriver::class)); $this->chooseDriver($this->app->make(Repository::class)); if ($this->app->runningInConsole()) { - $this->commands([ScheduleListCommand::class, ScheduleMigrateCommand::class]); + $this->commands([ScheduleListCommand::class, ScheduleMigrateCommand::class, ScheduleRecoverCommand::class]); } } @@ -63,14 +57,10 @@ public function boot(): void $this->app->tag([ScheduleDriverCheck::class], DoctorServiceProvider::CHECKS_TAG); $this->app->make(MigrationNotice::class)->register(); - if (config('wordpress.use_laravel_scheduler', false)) { - $this->app->make(RegisterSchedulerFiltersUseCase::class)->execute(); - } - $this->app->booted(function (): void { // The driver was chosen when the application started; WordPress can take its hooks by now ScheduleApi::boot(); - $this->scheduleRecurringEvents(); + $this->recoverLateEvents(); }); } @@ -96,44 +86,24 @@ private function chooseDriver(Repository $config): void */ private function registerDomainContracts(): void { - $this->app->singleton(SchedulerInterface::class, Scheduler::class); $this->app->singleton(ScheduleDiscovery::class, fn (): ScheduleDiscovery => new ScheduleDiscovery); + $this->app->singleton(LaravelQueueDriver::class); + $this->app->singleton(LateEventRecovery::class); } /** - * Register application use cases. - */ - private function registerUseCases(): void - { - $this->app->bind(RegisterSchedulerFiltersUseCase::class, fn (Application $app): RegisterSchedulerFiltersUseCase => new RegisterSchedulerFiltersUseCase( - $app->make(Filter::class), - $app->make(SchedulerInterface::class) - )); - - } - - /** - * Schedule all recurring events. + * Look for events of the "laravel" driver whose job was lost, when the request ends, at most every five minutes. */ - protected function scheduleRecurringEvents(): void + private function recoverLateEvents(): void { - if ($this->isOrchestraTest() || defined('WP_CLI')) { + if (! function_exists('add_action') || ScheduleApi::engine()->driverName() !== LaravelQueueDriver::NAME) { return; } - try { - $schedule = $this->app->make(Schedule::class); - RecurringEvent::scheduleAllEvents($schedule); - } catch (QueryException) { - // WordPress tables may not exist yet during initial installation - } - } - - /** - * Check if we're running in an Orchestra test environment. - */ - private function isOrchestraTest(): bool - { - return defined('LARAVEL_START') && class_exists(TestCase::class); + add_action('shutdown', function (): void { + if ($this->app->make(LaravelQueueDriver::class)->available()) { + $this->app->make(LateEventRecovery::class)->recover(); + } + }); } } diff --git a/src/Schedule/UI/Console/ScheduleRecoverCommand.php b/src/Schedule/UI/Console/ScheduleRecoverCommand.php new file mode 100644 index 00000000..ef195176 --- /dev/null +++ b/src/Schedule/UI/Console/ScheduleRecoverCommand.php @@ -0,0 +1,58 @@ +option('network') && function_exists('is_multisite') && is_multisite() + ? array_map(static fn ($site): int => (int) $site->blog_id, get_sites(['fields' => 'objects', 'number' => 0])) + : [null]; + + $total = 0; + + foreach ($sites as $site) { + if ($site !== null) { + switch_to_blog($site); + } + + try { + $total += $this->option('dry-run') ? $this->list($recovery, $site) : $recovery->recover(force: true); + } finally { + if ($site !== null) { + restore_current_blog(); + } + } + } + + $this->info(sprintf($this->option('dry-run') ? '%d late event(s).' : '%d event(s) sent again.', $total)); + + return self::SUCCESS; + } + + private function list(LateEventRecovery $recovery, ?int $site): int + { + $late = $recovery->late(); + + foreach ($late as $event) { + $this->line(sprintf('%s%s %s', $site === null ? '' : sprintf('[site %d] ', $site), gmdate('Y-m-d H:i:s', $event->timestamp), $event->hook)); + } + + return count($late); + } +} diff --git a/tests/Feature/Hook/Async/AsyncActionsCheckTest.php b/tests/Feature/Hook/Async/AsyncActionsCheckTest.php index ed96ee04..78d6e27a 100644 --- a/tests/Feature/Hook/Async/AsyncActionsCheckTest.php +++ b/tests/Feature/Hook/Async/AsyncActionsCheckTest.php @@ -170,8 +170,7 @@ public function dispatch(AsyncPayload $payload, int $delay = 0): void {} 'null' => [['jobs' => ['driver' => 'null']], 'discards jobs'], ]); -it('warns when WP-Cron events are overdue, with the fix for what runs WP-Cron', function (bool $onPageLoad, bool $laravelScheduler, string $fix): void { - config(['wordpress.use_laravel_scheduler' => $laravelScheduler]); +it('warns when WP-Cron events are overdue, with the fix for what runs WP-Cron', function (bool $onPageLoad, string $fix): void { ($this->register)('save_post'); $this->inspector->cronOnPageLoad = $onPageLoad; $this->inspector->overdue = ['count' => 3, 'oldest' => 1_800_000_000]; @@ -182,9 +181,8 @@ public function dispatch(AsyncPayload $payload, int $delay = 0): void {} ->and($result->details[0])->toBe('3 WP-Cron event(s) overdue, the oldest since 2027-01-15 08:00 UTC: WP-Cron and Action Scheduler run nothing until wp-cron.php is requested') ->and($result->fix)->toContain($fix); })->with([ - 'disabled' => [false, false, 'DISABLE_WP_CRON is set'], - 'on page load' => [true, false, 'too few visits'], - 'Laravel scheduler' => [false, true, 'run a queue worker'], + 'disabled' => [false, 'DISABLE_WP_CRON is set'], + 'on page load' => [true, 'too few visits'], ]); it('leaves overdue WP-Cron events alone when no driver in use relies on WP-Cron', function (): void { diff --git a/tests/Feature/Schedule/LaravelDriverTest.php b/tests/Feature/Schedule/LaravelDriverTest.php new file mode 100644 index 00000000..7ba7137e --- /dev/null +++ b/tests/Feature/Schedule/LaravelDriverTest.php @@ -0,0 +1,306 @@ +cron = []; + $this->fired = []; + $this->wpOptions = []; + + Functions\when('_get_cron_array')->alias(fn (): array => $this->cron); + Functions\when('_set_cron_array')->alias(function (array $cron): bool { + $this->cron = $cron; + + return true; + }); + Functions\when('wp_get_schedules')->justReturn(['hourly' => ['interval' => 3600], 'daily' => ['interval' => 86400]]); + Functions\when('add_filter')->justReturn(true); + Functions\when('add_action')->justReturn(true); + Functions\when('remove_filter')->justReturn(true); + Functions\when('do_action_ref_array')->alias(function (string $hook, array $args): void { + $this->fired[] = [$hook, $args]; + }); + Functions\when('wp_cache_flush_runtime')->justReturn(true); + Functions\when('get_current_blog_id')->justReturn(1); + Functions\when('is_multisite')->justReturn(false); + Functions\when('get_option')->alias(fn (string $name, mixed $default = false): mixed => $this->wpOptions[$name] ?? $default); + Functions\when('update_option')->alias(function (string $name, mixed $value): bool { + $this->wpOptions[$name] = $value; + + return true; + }); + Functions\when('did_action')->justReturn(1); + + config([ + 'queue.default' => 'database', + 'queue.connections.database' => ['driver' => 'database'], + 'queue.connections.sqs' => ['driver' => 'sqs'], + 'queue.connections.sync' => ['driver' => 'sync'], + 'schedule.laravel' => ['connection' => null, 'queue' => null, 'max_delay' => null, 'late_after' => 600], + ]); + + Bus::fake(); + Functions\when('wp_clear_scheduled_hook')->justReturn(0); + Functions\when('wp_unschedule_hook')->justReturn(0); + + foreach (['engine', 'fake', 'driverBeforeFake', 'wanted'] as $property) { + (new ReflectionClass(Schedule::class))->setStaticPropertyValue($property, null); + } + + app()->forgetInstance(LaravelQueueDriver::class); + app()->forgetInstance(LateEventRecovery::class); + config(['schedule.driver' => 'laravel', 'schedule.protected_hooks' => []]); + app()->register(SchedulerServiceProvider::class, force: true); + + $this->driver = resolve(LaravelQueueDriver::class); + $this->engine = Schedule::engine(); +}); + +/** + * @return list + */ +function dispatchedJobs(): array +{ + return Bus::dispatched(RunScheduledEvent::class)->all(); +} + +describe('the driver', function (): void { + it('is available on a queue a worker runs, not on sync or null', function (): void { + expect($this->driver->available())->toBeTrue(); + + config(['schedule.laravel.connection' => 'sync']); + expect($this->driver->available())->toBeFalse(); + + config(['schedule.laravel.connection' => 'missing']); + expect($this->driver->available())->toBeFalse(); + }); + + it('queues a delayed job and writes the entry, with its identifier, in the index', function (): void { + $time = time() + 7200; + $this->engine->place(new Event('digest', $time, null, ['u' => 1])); + + $job = dispatchedJobs()[0]; + + expect(dispatchedJobs())->toHaveCount(1) + ->and($job->hook)->toBe('digest') + ->and($job->timestamp)->toBe($time) + ->and($job->delay)->toBeBetween(7199, 7200) + ->and($job->connection)->toBe('database') + ->and(CronIndex::ownerOf('digest', ['u' => 1]))->toBe('laravel') + ->and(CronIndex::idOf('digest', ['u' => 1]))->toBe($job->id) + ->and($this->engine->next('digest', ['u' => 1]))->toBe($time); + }); + + it('queues a job for an event already due without a delay', function (): void { + $this->engine->place(new Event('digest', time() - 5)); + + expect(dispatchedJobs()[0]->delay)->toBeNull(); + }); + + it('waits by the longest delay a connection accepts, and by the one configured', function (): void { + config(['schedule.laravel.connection' => 'sqs']); + $this->engine->place(new Event('digest', time() + 7200)); + + config(['schedule.laravel.connection' => 'database', 'schedule.laravel.max_delay' => 60]); + $this->engine->place(new Event('other', time() + 7200)); + + expect(dispatchedJobs()[0]->delay)->toBe(900) + ->and(dispatchedJobs()[1]->delay)->toBe(60); + }); + + it('cancels by taking the entry out of the index, without reading the queue', function (): void { + $this->engine->place(new Event('digest', time() + 3600, null, ['u' => 1])); + + expect($this->engine->cancel('digest', ['u' => 1]))->toBe(1) + ->and(CronIndex::ownerOf('digest', ['u' => 1]))->toBeNull() + ->and($this->engine->next('digest', ['u' => 1]))->toBeNull(); + }); + + it('stores a single event of the WordPress functions the same way', function (): void { + $this->driver->once('digest', time() + 600, []); + $this->driver->recurring('reports', time() + 600, 'daily', []); + + expect(CronIndex::events('laravel'))->toHaveCount(2) + ->and(CronIndex::events('laravel')[0]->id)->not->toBeNull() + ->and(dispatchedJobs()[1]->interval)->toBe(86400); + }); +}); + +describe('the job', function (): void { + it('fires the hook and takes a single event out of the index', function (): void { + $this->engine->place(new Event('digest', time() - 1, null, ['u' => 1])); + + dispatchedJobs()[0]->handle($this->driver); + + expect($this->fired)->toBe([['digest', ['u' => 1]]]) + ->and(CronIndex::events('laravel'))->toBe([]); + }); + + it('queues the next occurrence of a recurring event, never in the past', function (): void { + $this->engine->place(new Event('reports', time() - 10000, 'hourly', [], 3600)); + $job = dispatchedJobs()[0]; + + $job->handle($this->driver); + + $next = CronIndex::events('laravel')[0]; + + expect($this->fired)->toHaveCount(1) + ->and($next->timestamp)->toBeGreaterThan(time()) + ->and($next->timestamp)->toBeLessThanOrEqual(time() + 3600) + ->and(($next->timestamp - $job->timestamp) % 3600)->toBe(0) + ->and($next->id)->not->toBe($job->id) + ->and(dispatchedJobs())->toHaveCount(2) + ->and(dispatchedJobs()[1]->id)->toBe($next->id); + }); + + it('stops when its event was cancelled, moved or replaced', function (): void { + $this->engine->place(new Event('digest', time() - 1)); + $this->engine->place(new Event('moved', time() - 1)); + $this->engine->place(new Event('replaced', time() - 1)); + [$cancelled, $moved, $replaced] = dispatchedJobs(); + + $this->engine->cancel('digest'); + CronIndex::record((new Event('moved', time() + 5000))->withId($moved->id), 'laravel'); + CronIndex::record((new Event('replaced', time() - 1))->withId('another'), 'laravel'); + + $cancelled->handle($this->driver); + $moved->handle($this->driver); + $replaced->handle($this->driver); + + expect($this->fired)->toBe([]); + }); + + it('waits again when it wakes before its time, and runs only once when sent twice', function (): void { + $this->engine->place(new Event('digest', time() + 3600)); + $early = dispatchedJobs()[0]; + + $early->handle($this->driver); + + expect($this->fired)->toBe([]) + ->and(dispatchedJobs())->toHaveCount(2) + ->and(dispatchedJobs()[1]->id)->toBe($early->id) + ->and(dispatchedJobs()[1]->timestamp)->toBe($early->timestamp); + + $this->cron = []; + $this->engine->place(new Event('again', time() - 1)); + $first = dispatchedJobs()[2]; + $copy = clone $first; + + $first->handle($this->driver); + $copy->handle($this->driver); + + expect($this->fired)->toBe([['again', []]]); + }); + + it('records a final failure, and goes on with a recurring event', function (): void { + $this->engine->place(new Event('reports', time() - 10, 'hourly', [], 3600)); + $this->engine->place(new Event('digest', time() - 10)); + [$recurring, $single] = dispatchedJobs(); + + $recurring->failed(new RuntimeException('boom')); + $single->failed(new RuntimeException('bang')); + + expect(array_column(ScheduleFailures::recent(), 'hook'))->toBe(['reports', 'digest']) + ->and(ScheduleFailures::recent()[1]['message'])->toBe('bang') + ->and(array_column(CronIndex::events('laravel'), 'hook'))->toBe(['reports']) + ->and(dispatchedJobs())->toHaveCount(3); + }); + + it('runs in the site that scheduled the event on a network', function (): void { + $switched = []; + Functions\when('is_multisite')->justReturn(true); + Functions\when('switch_to_blog')->alias(function (int $site) use (&$switched): bool { + $switched[] = $site; + + return true; + }); + Functions\when('restore_current_blog')->alias(function () use (&$switched): bool { + $switched[] = 'restore'; + + return true; + }); + + $this->engine->place(new Event('digest', time() - 1)); + $job = dispatchedJobs()[0]; + $job = new RunScheduledEvent($job->hook, $job->args, $job->timestamp, $job->id, null, null, 3); + + $job->handle($this->driver); + + expect($switched)->toBe([3, 'restore']); + }); +}); + +describe('the recovery', function (): void { + it('sends again the events more than late_after seconds past their time, with the same identifier, once', function (): void { + $this->engine->place(new Event('lost', time() - 1200)); + $this->engine->place(new Event('fresh', time() - 30)); + $this->engine->place(new Event('future', time() + 3000)); + + $queued = count(dispatchedJobs()); + + $recovery = resolve(LateEventRecovery::class); + $sent = $recovery->recover(); + + expect($sent)->toBe(1) + ->and(dispatchedJobs())->toHaveCount($queued + 1) + ->and(dispatchedJobs()[$queued]->hook)->toBe('lost') + ->and(dispatchedJobs()[$queued]->id)->toBe(CronIndex::idOf('lost')) + ->and($recovery->recover())->toBe(0) + ->and($recovery->recover(force: true))->toBe(1); + }); + + it('gives an event of the index that has no identifier one', function (): void { + $this->cron = [time() - 5000 => ['lost' => [md5(serialize([])) => ['schedule' => false, 'args' => [], 'driver' => 'laravel']]]]; + + expect(resolve(LateEventRecovery::class)->recover(force: true))->toBe(1) + ->and(CronIndex::idOf('lost'))->not->toBeNull(); + }); + + it('runs from the command, and lists with --dry-run', function (): void { + $this->engine->place(new Event('lost', time() - 1200)); + + expect(Artisan::call('pollora:schedule:recover', ['--dry-run' => true]))->toBe(0) + ->and(Artisan::output())->toContain('lost', '1 late event(s).') + ->and(dispatchedJobs())->toHaveCount(1) + ->and(Artisan::call('pollora:schedule:recover'))->toBe(0) + ->and(Artisan::output())->toContain('1 event(s) sent again.') + ->and(dispatchedJobs())->toHaveCount(2); + }); +}); + +describe('the doctor check', function (): void { + it('warns of late events, and of a failure that left a single event out of the schedule', function (): void { + $this->engine->place(new Event('lost', time() - 1200)); + ScheduleFailures::record('digest', 'bang'); + + $result = resolve(ScheduleDriverCheck::class)->run(RunContext::Console); + + expect($result->status)->toBe(CheckStatus::Warning) + ->and(implode(' ', $result->details))->toContain('1 scheduled event(s) are more than 10 minutes late', '"digest" failed for good') + ->and($result->fix)->toContain('queue:work'); + }); + + it('says what to change when the queue is sync', function (): void { + config(['schedule.laravel.connection' => 'sync']); + + $result = resolve(ScheduleDriverCheck::class)->run(RunContext::Console); + + expect($result->status)->toBe(CheckStatus::Warning) + ->and($result->fix)->toContain('never sync or null'); + }); +}); diff --git a/tests/Feature/Schedule/ScheduleCommandsTest.php b/tests/Feature/Schedule/ScheduleCommandsTest.php index d07cb73e..85cc56b6 100644 --- a/tests/Feature/Schedule/ScheduleCommandsTest.php +++ b/tests/Feature/Schedule/ScheduleCommandsTest.php @@ -107,6 +107,8 @@ public function events(): array Functions\when('wp_schedule_event')->justReturn(true); Functions\when('wp_schedule_single_event')->justReturn(true); Functions\when('did_action')->justReturn(1); + Functions\when('get_current_blog_id')->justReturn(1); + Functions\when('is_multisite')->justReturn(false); Functions\when('wp_date')->alias(fn (string $format, int $timestamp): string => gmdate($format, $timestamp)); Schedule::extend('memory', fn (): ScheduleDriver => $this->memory); diff --git a/tests/Feature/Schedule/SchedulerServiceProviderTest.php b/tests/Feature/Schedule/SchedulerServiceProviderTest.php index 377ad4f5..90084e35 100644 --- a/tests/Feature/Schedule/SchedulerServiceProviderTest.php +++ b/tests/Feature/Schedule/SchedulerServiceProviderTest.php @@ -4,11 +4,8 @@ use Pollora\Discovery\Domain\Contracts\DiscoveryEngineInterface; use Pollora\Hook\Domain\Contract\Filter; -use Pollora\Schedule\Application\UseCases\RegisterSchedulerFiltersUseCase; -use Pollora\Schedule\Contracts\SchedulerInterface; use Pollora\Schedule\Infrastructure\Services\ScheduleDiscovery; use Pollora\Schedule\Schedule; -use Pollora\Schedule\Scheduler; use Pollora\Schedule\SchedulerServiceProvider; beforeEach(function (): void { @@ -23,13 +20,6 @@ }); describe('SchedulerServiceProvider', function (): void { - it('binds SchedulerInterface as singleton', function (): void { - $scheduler = $this->app->make(SchedulerInterface::class); - - expect($scheduler)->toBeInstanceOf(Scheduler::class); - expect($this->app->make(SchedulerInterface::class))->toBe($scheduler); - }); - it('binds ScheduleDiscovery as singleton', function (): void { $discovery = $this->app->make(ScheduleDiscovery::class); @@ -37,12 +27,6 @@ expect($this->app->make(ScheduleDiscovery::class))->toBe($discovery); }); - it('binds RegisterSchedulerFiltersUseCase', function (): void { - $useCase = $this->app->make(RegisterSchedulerFiltersUseCase::class); - - expect($useCase)->toBeInstanceOf(RegisterSchedulerFiltersUseCase::class); - }); - it('gives the schedule engine the application, so the classes passed to call() get their dependencies', function (): void { $container = (new ReflectionProperty(Schedule::engine(), 'container'))->getValue(Schedule::engine()); diff --git a/tests/Unit/Schedule/Application/UseCases/RegisterSchedulerFiltersUseCaseTest.php b/tests/Unit/Schedule/Application/UseCases/RegisterSchedulerFiltersUseCaseTest.php deleted file mode 100644 index 27641a5a..00000000 --- a/tests/Unit/Schedule/Application/UseCases/RegisterSchedulerFiltersUseCaseTest.php +++ /dev/null @@ -1,72 +0,0 @@ -filter = m::mock(Filter::class); - $this->scheduler = m::mock(SchedulerInterface::class); - $this->useCase = new RegisterSchedulerFiltersUseCase($this->filter, $this->scheduler); -}); - -describe('RegisterSchedulerFiltersUseCase', function (): void { - - it('registers all nine WordPress cron filters', function (): void { - $this->filter->shouldReceive('add')->times(9); - - $this->useCase->execute(); - }); - - it('registers event query filters with correct hooks', function (): void { - $scheduler = $this->scheduler; - - $this->filter->shouldReceive('add') - ->with('pre_get_scheduled_event', [$scheduler, 'preGetScheduledEvent'], 10, 5) - ->once(); - $this->filter->shouldReceive('add') - ->with('pre_get_ready_cron_jobs', [$scheduler, 'preGetReadyCronJobs'], 10, 5) - ->once(); - - // Allow the management filters - $this->filter->shouldReceive('add')->times(7); - - $this->useCase->execute(); - }); - - it('registers event management filters with correct hooks', function (): void { - $scheduler = $this->scheduler; - - $expectedManagementFilters = [ - 'pre_update_option_cron' => 'preUpdateOptionCron', - 'pre_option_cron' => 'preOptionCron', - 'pre_schedule_event' => 'preScheduleEvent', - 'pre_reschedule_event' => 'preRescheduleEvent', - 'pre_unschedule_event' => 'preUnscheduleEvent', - 'pre_clear_scheduled_hook' => 'preClearScheduledHook', - 'pre_unschedule_hook' => 'preUnscheduleHook', - ]; - - foreach ($expectedManagementFilters as $hook => $method) { - $this->filter->shouldReceive('add') - ->with($hook, [$scheduler, $method], 10, 5) - ->once(); - } - - // Allow the query filters - $this->filter->shouldReceive('add')->times(2); - - $this->useCase->execute(); - }); - - it('registers all filters with priority 10 and 5 accepted args', function (): void { - $this->filter->shouldReceive('add') - ->with(m::type('string'), m::type('array'), 10, 5) - ->times(9); - - $this->useCase->execute(); - }); -}); From 65df8ce8137b25b4c374467e4c83153d4be0618a Mon Sep 17 00:00:00 2001 From: Olivier Gorzalka Date: Tue, 13 Oct 2026 17:57:15 +0200 Subject: [PATCH 2/2] fix: build the laravel driver before the bus exists, extend before the engine is created The driver is chosen while providers register, so it resolves the bus when a job is queued. Requires pollora/schedule ^0.1.3 (Identifies). --- composer.json | 2 +- .../Infrastructure/Drivers/LaravelQueueDriver.php | 8 ++++++-- src/Schedule/SchedulerServiceProvider.php | 3 ++- tests/Feature/Schedule/LaravelDriverTest.php | 7 +++++++ 4 files changed, 16 insertions(+), 4 deletions(-) diff --git a/composer.json b/composer.json index a926bd66..9530450e 100644 --- a/composer.json +++ b/composer.json @@ -36,7 +36,7 @@ "pollora/entity": "^1.2", "pollora/hook": "^2.0", "pollora/option": "^1.1", - "pollora/schedule": "^0.1.1 || ^1.0", + "pollora/schedule": "^0.1.3 || ^1.0", "pollora/wp-cli": "^0.1 || ^1.0", "pollora/wp-rest": "^0.1 || ^1.0", "watson/rememberable": "^7.1", diff --git a/src/Schedule/Infrastructure/Drivers/LaravelQueueDriver.php b/src/Schedule/Infrastructure/Drivers/LaravelQueueDriver.php index d7c631bc..c7bca8a3 100644 --- a/src/Schedule/Infrastructure/Drivers/LaravelQueueDriver.php +++ b/src/Schedule/Infrastructure/Drivers/LaravelQueueDriver.php @@ -6,6 +6,7 @@ use Illuminate\Contracts\Bus\Dispatcher; use Illuminate\Contracts\Config\Repository; +use Illuminate\Contracts\Container\Container; use Pollora\Schedule\Contracts\Identifies; use Pollora\Schedule\Contracts\ScheduleDriver; use Pollora\Schedule\Event; @@ -31,8 +32,11 @@ */ private const array MAX_DELAYS = ['sqs' => 900]; + /** + * The bus is resolved when a job is queued: the driver is built while the application is still registering its providers. + */ public function __construct( - private Dispatcher $dispatcher, + private Container $container, private Repository $config, ) {} @@ -109,7 +113,7 @@ public function send(RunScheduledEvent $job): void $job->delay($delay); } - $this->dispatcher->dispatch($job); + $this->container->make(Dispatcher::class)->dispatch($job); } /** diff --git a/src/Schedule/SchedulerServiceProvider.php b/src/Schedule/SchedulerServiceProvider.php index bc1e25fd..87998f59 100755 --- a/src/Schedule/SchedulerServiceProvider.php +++ b/src/Schedule/SchedulerServiceProvider.php @@ -35,8 +35,9 @@ public function register(): void $this->mergeConfigFrom(__DIR__.'/../../config/schedule.php', 'schedule'); $this->app->singleton(ScheduleStatus::class, fn (Application $app): ScheduleStatus => new ScheduleStatus($app->make(Repository::class))); - ScheduleApi::useContainer($this->app); + // Before anything creates the engine: POLLORA_SCHEDULE_DRIVER may name it ScheduleApi::extend(LaravelQueueDriver::NAME, fn (): LaravelQueueDriver => $this->app->make(LaravelQueueDriver::class)); + ScheduleApi::useContainer($this->app); $this->chooseDriver($this->app->make(Repository::class)); if ($this->app->runningInConsole()) { diff --git a/tests/Feature/Schedule/LaravelDriverTest.php b/tests/Feature/Schedule/LaravelDriverTest.php index 7ba7137e..298e29b5 100644 --- a/tests/Feature/Schedule/LaravelDriverTest.php +++ b/tests/Feature/Schedule/LaravelDriverTest.php @@ -3,6 +3,7 @@ declare(strict_types=1); use Brain\Monkey\Functions; +use Illuminate\Container\Container; use Illuminate\Support\Facades\Artisan; use Illuminate\Support\Facades\Bus; use Pollora\Doctor\Domain\Enums\CheckStatus; @@ -90,6 +91,12 @@ function dispatchedJobs(): array expect($this->driver->available())->toBeFalse(); }); + it('is built before the application has a bus: the driver is chosen while the providers register', function (): void { + $driver = new LaravelQueueDriver(new Container, resolve('config')); + + expect($driver->available())->toBeTrue(); + }); + it('queues a delayed job and writes the entry, with its identifier, in the index', function (): void { $time = time() + 7200; $this->engine->place(new Event('digest', $time, null, ['u' => 1]));