Async actions: tries(), backoff(), asUser(), onQueue()

CHANGELOG.md

@@ -8,6 +8,10 @@ All notable changes to `pollora/hook` are documented in this file.
 
 - `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.
 - `when()`: queue only when the callback, given the hook arguments, returns `true`. It runs before the capture, and an exception it throws goes through, as it would synchronously.
+- `tries()`: a handler that throws is queued again, through the driver that queued it, until it has used its attempts; each failed attempt is reported, the last one is fired as `pollora/async/failed`. `backoff()` sets the seconds before each retry (default 10, 60, then 300; the last value repeats). A handler that no longer exists is not retried.
+- `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.
+- Payloads record the driver that queued them, the queue, `asUser` and the backoff. These fields are optional, so payloads queued by 1.2.0 still run.
 
 ## [1.2.0] - 2026-10-07
 

src/Async/AsyncDispatcher.php

@@ -38,6 +38,8 @@ public function dispatch(QueuedHandler $handler, array $arguments): void
         }
 
         try {
+            [$driverName, $driver] = $this->driverFor($handler);
+
             $payload = new AsyncPayload(
                 id: AsyncPayload::newId(),
                 hook: $handler->hook,
@@ -46,10 +48,15 @@ public function dispatch(QueuedHandler $handler, array $arguments): void
                 arguments: $this->normalizer->normalize($arguments),
                 origin: self::origin(),
                 captured: $this->normalizer->normalize($this->capture($handler, $arguments), 'captured value'),
+                tries: $handler->options->attempts(),
                 keepMissing: $handler->options->keepsMissing(),
+                driver: $driverName,
+                queue: $handler->options->queue(),
+                asUser: $handler->options->runsAsUser(),
+                backoff: $handler->options->backoffDelays(),
             );
 
-            $this->driverFor($handler)->dispatch($payload, $handler->options->delayInSeconds());
+            $driver->dispatch($payload, $handler->options->delayInSeconds());
         } catch (\Throwable $throwable) {
             if (Async::isDebug()) {
                 throw $throwable;
@@ -107,28 +114,30 @@ private function capture(QueuedHandler $handler, array $arguments): array
     }
 
     /**
-     * The driver requested by the registration, or the default one.
+     * The driver requested by the registration, or the default one, with its name.
+     *
+     * @return array{0: string, 1: AsyncDriver}
      *
      * @throws DriverUnavailable
      */
-    private function driverFor(QueuedHandler $handler): AsyncDriver
+    private function driverFor(QueuedHandler $handler): array
     {
         $requested = $handler->options->driver();
 
-        if ($requested === null) {
-            return Async::driver();
-        }
+        if ($requested !== null) {
+            try {
+                return [$requested, Async::driver($requested)];
+            } catch (DriverUnavailable $driverUnavailable) {
+                if (Async::isDebug()) {
+                    throw $driverUnavailable;
+                }
 
-        try {
-            return Async::driver($requested);
-        } catch (DriverUnavailable $driverUnavailable) {
-            if (Async::isDebug()) {
-                throw $driverUnavailable;
+                Async::report($driverUnavailable, ['hook' => $handler->hook, 'handler' => $handler->descriptor]);
             }
+        }
 
-            Async::report($driverUnavailable, ['hook' => $handler->hook, 'handler' => $handler->descriptor]);
+        $default = Async::defaultDriver();
 
-            return Async::driver();
-        }
+        return [$default, Async::driver($default)];
     }
 }

src/Async/AsyncPayload.php

@@ -29,6 +29,10 @@
      * @param  int  $attempt  Attempt number, from 1
      * @param  int  $tries  Number of attempts allowed
      * @param  bool  $keepMissing  Run with null in place of a referenced object that no longer exists
+     * @param  string|null  $driver  Driver that queued the payload, used again to retry it
+     * @param  string|null  $queue  Queue name: the group with Action Scheduler, ignored by WP-Cron
+     * @param  bool  $asUser  Run as the user who fired the hook
+     * @param  list<int>  $backoff  Seconds before each retry; the last value repeats
      */
     public function __construct(
         public string $id,
@@ -41,6 +45,10 @@ public function __construct(
         public int $attempt = 1,
         public int $tries = 1,
         public bool $keepMissing = false,
+        public ?string $driver = null,
+        public ?string $queue = null,
+        public bool $asUser = false,
+        public array $backoff = [],
     ) {}
 
     /**
@@ -94,6 +102,11 @@ public static function fromJson(string $json): self
             attempt: self::field($data, 'attempt', 'is_int'),
             tries: self::field($data, 'tries', 'is_int'),
             keepMissing: self::field($data, 'keepMissing', 'is_bool'),
+            // Fields added after 1.2.0: optional, so payloads queued by 1.2.0 still run
+            driver: self::optionalField($data, 'driver', 'is_string', null),
+            queue: self::optionalField($data, 'queue', 'is_string', null),
+            asUser: self::optionalField($data, 'asUser', 'is_bool', false),
+            backoff: self::backoffField($data),
         );
     }
 
@@ -116,6 +129,10 @@ public function toJson(): string
             'attempt' => $this->attempt,
             'tries' => $this->tries,
             'keepMissing' => $this->keepMissing,
+            'driver' => $this->driver,
+            'queue' => $this->queue,
+            'asUser' => $this->asUser,
+            'backoff' => $this->backoff,
         ], JSON_THROW_ON_ERROR | JSON_UNESCAPED_SLASHES | JSON_UNESCAPED_UNICODE | JSON_PRESERVE_ZERO_FRACTION);
     }
 
@@ -127,6 +144,18 @@ public function withAttempt(int $attempt): self
         return new self(...[...get_object_vars($this), 'attempt' => $attempt]);
     }
 
+    /**
+     * Seconds to wait before retrying after the given failed attempt.
+     */
+    public function retryDelay(int $failedAttempt): int
+    {
+        if ($this->backoff === []) {
+            return 0;
+        }
+
+        return $this->backoff[min(max($failedAttempt, 1), count($this->backoff)) - 1];
+    }
+
     /**
      * The context handed to the handler, with the captured values rebuilt.
      *
@@ -145,6 +174,34 @@ public function context(array $captured = []): AsyncContext
         );
     }
 
+    /**
+     * @param  array<string, mixed>  $data
+     * @param  callable(mixed): bool  $check
+     */
+    private static function optionalField(array $data, string $field, callable $check, mixed $default): mixed
+    {
+        if (! array_key_exists($field, $data) || $data[$field] === null) {
+            return $default;
+        }
+
+        return self::field($data, $field, $check);
+    }
+
+    /**
+     * @param  array<string, mixed>  $data
+     * @return list<int>
+     */
+    private static function backoffField(array $data): array
+    {
+        $backoff = self::optionalField($data, 'backoff', 'is_array', []);
+
+        if (! array_is_list($backoff) || array_filter($backoff, fn (mixed $delay): bool => ! is_int($delay) || $delay < 0) !== []) {
+            throw InvalidPayload::invalidField('backoff');
+        }
+
+        return $backoff;
+    }
+
     /**
      * @param  array<string, mixed>  $data
      * @param  callable(mixed): bool  $check

src/Async/AsyncRunner.php

@@ -10,9 +10,10 @@
  * Runs a queued handler: resolves it, rebuilds its arguments, restores the
  * site and locale of the original request, calls it.
  *
- * A failure is reported and announced through the 'pollora/async/failed'
- * action; it never escapes, so one failing handler does not stop the others
- * a WP-Cron run holds.
+ * A handler that throws is queued again, through the driver that queued it,
+ * until it has used its attempts; the last failure is reported and announced
+ * through the 'pollora/async/failed' action. Nothing escapes, so one failing
+ * handler does not stop the others a WP-Cron run holds.
  *
  * @internal
  */
@@ -47,7 +48,7 @@ public function run(AsyncPayload $payload): void
         try {
             $this->call($callable, $arguments, $payload->context($captured));
         } catch (\Throwable $throwable) {
-            $this->fail($payload, $throwable);
+            $this->retryOrFail($payload, $throwable);
         } finally {
             $leaveContext();
 
@@ -80,32 +81,67 @@ public function isRunning(string $key): bool
     }
 
     /**
-     * Switch to the site and locale of the original request.
+     * Switch to the site, the user (with asUser) and the locale of the original request.
      *
      * @return \Closure(): void Switches back
      */
     private function enterContext(AsyncPayload $payload): \Closure
     {
         $blogId = $payload->origin['blogId'];
+        $userId = $payload->origin['userId'];
         $locale = $payload->origin['locale'];
 
         $switchedBlog = function_exists('is_multisite') && is_multisite() && $blogId !== get_current_blog_id()
             && switch_to_blog($blogId);
 
+        $previousUser = null;
+        if ($payload->asUser && $userId > 0 && function_exists('wp_set_current_user')) {
+            $previousUser = get_current_user_id();
+            wp_set_current_user($userId);
+        }
+
         $switchedLocale = function_exists('switch_to_locale') && determine_locale() !== $locale
             && switch_to_locale($locale);
 
-        return static function () use ($switchedBlog, $switchedLocale): void {
+        return static function () use ($switchedBlog, $previousUser, $switchedLocale): void {
             if ($switchedLocale) {
                 restore_previous_locale();
             }
 
+            if ($previousUser !== null) {
+                wp_set_current_user($previousUser);
+            }
+
             if ($switchedBlog) {
                 restore_current_blog();
             }
         };
     }
 
+    /**
+     * Queue the handler again when it has attempts left, report the failure otherwise.
+     */
+    private function retryOrFail(AsyncPayload $payload, \Throwable $throwable): void
+    {
+        if ($payload->attempt < $payload->tries) {
+            $delay = $payload->retryDelay($payload->attempt);
+
+            try {
+                Async::driver($payload->driver ?? Async::defaultDriver())->dispatch($payload->withAttempt($payload->attempt + 1), $delay);
+                Async::report(
+                    sprintf('Attempt %d of %d failed, retried in %d s: %s', $payload->attempt, $payload->tries, $delay, $throwable->getMessage()),
+                    ['hook' => $payload->hook, 'handler' => $payload->handler, 'payload' => $payload->id, 'exception' => $throwable],
+                );
+
+                return;
+            } catch (\Throwable $retryFailure) {
+                Async::report($retryFailure, ['hook' => $payload->hook, 'handler' => $payload->handler, 'payload' => $payload->id]);
+            }
+        }
+
+        $this->fail($payload, $throwable);
+    }
+
     private function fail(AsyncPayload $payload, \Throwable $throwable): void
     {
         Async::report($throwable, ['hook' => $payload->hook, 'handler' => $payload->handler, 'payload' => $payload->id]);

src/Async/PendingAsync.php

@@ -21,6 +21,15 @@ final class PendingAsync
 
     private bool $keepMissing = false;
 
+    private int $tries = 1;
+
+    /** @var list<int> */
+    private array $backoff = [10, 60, 300];
+
+    private bool $asUser = false;
+
+    private ?string $queue = null;
+
     /** @var (\Closure(mixed...): array<string, mixed>)|null */
     private ?\Closure $capture = null;
 
@@ -101,6 +110,66 @@ public function keepMissing(bool $keepMissing = true): self
         return $this;
     }
 
+    /**
+     * Number of attempts when the handler throws: it is queued again after each
+     * failure but the last, which is reported and fired as 'pollora/async/failed'.
+     *
+     * @throws \InvalidArgumentException When $tries is lower than 1
+     */
+    public function tries(int $tries): self
+    {
+        if ($tries < 1) {
+            throw new \InvalidArgumentException(sprintf('tries() expects at least 1 attempt, %d given.', $tries));
+        }
+
+        $this->tries = $tries;
+
+        return $this;
+    }
+
+    /**
+     * Seconds to wait before each retry; the last value repeats. Default: 10, 60, then 300.
+     *
+     * @param  int|array<int|string, mixed>  $seconds  Validated: integers of 0 or more
+     *
+     * @throws \InvalidArgumentException When a delay is negative or the list is empty
+     */
+    public function backoff(int|array $seconds): self
+    {
+        $seconds = array_values((array) $seconds);
+
+        if ($seconds === [] || array_filter($seconds, fn (mixed $delay): bool => ! is_int($delay) || $delay < 0) !== []) {
+            throw new \InvalidArgumentException('backoff() expects one or more delays of 0 seconds or more.');
+        }
+
+        $this->backoff = $seconds;
+
+        return $this;
+    }
+
+    /**
+     * Run the handler as the user who fired the hook, then remove that user.
+     *
+     * Without it the handler runs without a current user, so a capability
+     * check fails there, which is the safe default.
+     */
+    public function asUser(bool $asUser = true): self
+    {
+        $this->asUser = $asUser;
+
+        return $this;
+    }
+
+    /**
+     * Queue name: the group with Action Scheduler, the queue with a Laravel queue. WP-Cron ignores it.
+     */
+    public function onQueue(string $queue): self
+    {
+        $this->queue = $queue;
+
+        return $this;
+    }
+
     /**
      * Record values at trigger time, read back at execution through AsyncContext::get().
      *
@@ -181,6 +250,40 @@ public function keepsMissing(): bool
         return $this->keepMissing;
     }
 
+    /**
+     * @internal
+     */
+    public function attempts(): int
+    {
+        return $this->tries;
+    }
+
+    /**
+     * @internal
+     *
+     * @return list<int>
+     */
+    public function backoffDelays(): array
+    {
+        return $this->backoff;
+    }
+
+    /**
+     * @internal
+     */
+    public function runsAsUser(): bool
+    {
+        return $this->asUser;
+    }
+
+    /**
+     * @internal
+     */
+    public function queue(): ?string
+    {
+        return $this->queue;
+    }
+
     /**
      * @internal
      *