Async actions: unique()

CHANGELOG.md

@@ -11,7 +11,8 @@ All notable changes to `pollora/hook` are documented in this file.
 - `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.
+- `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.
+- 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
 

src/Async/AsyncDispatcher.php

@@ -37,15 +37,29 @@ public function dispatch(QueuedHandler $handler, array $arguments): void
             return;
         }
 
+        $uniqueKey = null;
+
         try {
             [$driverName, $driver] = $this->driverFor($handler);
+            $normalizedArguments = $this->normalizer->normalize($arguments);
+
+            $uniqueFor = $handler->options->uniqueFor();
+            if ($uniqueFor !== null) {
+                $key = UniqueLock::key($handler->hook, $handler->descriptor, $normalizedArguments);
+
+                if (! UniqueLock::acquire($key, $uniqueFor)) {
+                    return;
+                }
+
+                $uniqueKey = $key;
+            }
 
             $payload = new AsyncPayload(
                 id: AsyncPayload::newId(),
                 hook: $handler->hook,
                 handler: $handler->descriptor,
                 priority: $handler->priority,
-                arguments: $this->normalizer->normalize($arguments),
+                arguments: $normalizedArguments,
                 origin: self::origin(),
                 captured: $this->normalizer->normalize($this->capture($handler, $arguments), 'captured value'),
                 tries: $handler->options->attempts(),
@@ -54,10 +68,15 @@ public function dispatch(QueuedHandler $handler, array $arguments): void
                 queue: $handler->options->queue(),
                 asUser: $handler->options->runsAsUser(),
                 backoff: $handler->options->backoffDelays(),
+                uniqueKey: $uniqueKey,
             );
 
             $driver->dispatch($payload, $handler->options->delayInSeconds());
         } catch (\Throwable $throwable) {
+            if ($uniqueKey !== null) {
+                UniqueLock::release($uniqueKey);
+            }
+
             if (Async::isDebug()) {
                 throw $throwable;
             }

src/Async/AsyncPayload.php

@@ -33,6 +33,7 @@
      * @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
+     * @param  string|null  $uniqueKey  Lock to release when the first attempt starts, see UniqueLock
      */
     public function __construct(
         public string $id,
@@ -49,6 +50,7 @@ public function __construct(
         public ?string $queue = null,
         public bool $asUser = false,
         public array $backoff = [],
+        public ?string $uniqueKey = null,
     ) {}
 
     /**
@@ -107,6 +109,7 @@ public static function fromJson(string $json): self
             queue: self::optionalField($data, 'queue', 'is_string', null),
             asUser: self::optionalField($data, 'asUser', 'is_bool', false),
             backoff: self::backoffField($data),
+            uniqueKey: self::optionalField($data, 'uniqueKey', 'is_string', null),
         );
     }
 
@@ -133,6 +136,7 @@ public function toJson(): string
             'queue' => $this->queue,
             'asUser' => $this->asUser,
             'backoff' => $this->backoff,
+            'uniqueKey' => $this->uniqueKey,
         ], JSON_THROW_ON_ERROR | JSON_UNESCAPED_SLASHES | JSON_UNESCAPED_UNICODE | JSON_PRESERVE_ZERO_FRACTION);
     }
 

src/Async/AsyncRunner.php

@@ -29,6 +29,11 @@ public function __construct(
 
     public function run(AsyncPayload $payload): void
     {
+        // Identical triggers queue again from the moment the first attempt starts
+        if ($payload->uniqueKey !== null && $payload->attempt === 1) {
+            UniqueLock::release($payload->uniqueKey);
+        }
+
         try {
             $callable = $this->descriptor->resolve($payload->handler);
             $arguments = $this->normalizer->denormalize($payload->arguments, $payload->keepMissing);

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\UniqueLock;
 
 /**
  * Queues through WP-Cron.
@@ -19,7 +20,8 @@
  * an execution as a duplicate of another scheduled within ten minutes.
  *
  * A payload whose event was lost (the 'cron' option reset, an event deleted by
- * a plugin) is found by a daily recovery task and scheduled again. Only the
+ * a plugin) is found by a daily recovery task and scheduled again; the same
+ * task deletes expired unique locks. Only the
  * process that deletes a payload runs it, so scheduling it again never runs it
  * twice.
  */
@@ -137,6 +139,8 @@ public static function recover(): int
             }
         }
 
+        UniqueLock::prune();
+
         if ($recovered > 0) {
             Async::report(sprintf('WP-Cron: %d asynchronous execution(s) whose event was lost scheduled again.', $recovered));
         }

src/Async/PendingAsync.php

@@ -30,6 +30,8 @@ final class PendingAsync
 
     private ?string $queue = null;
 
+    private ?int $uniqueFor = null;
+
     /** @var (\Closure(mixed...): array<string, mixed>)|null */
     private ?\Closure $capture = null;
 
@@ -160,6 +162,25 @@ public function asUser(bool $asUser = true): self
         return $this;
     }
 
+    /**
+     * Merge identical triggers (same hook, same handler, same arguments) while
+     * the first one has not started running.
+     *
+     * @param  int  $for  Seconds after which the lock expires if it was never released
+     *
+     * @throws \InvalidArgumentException When $for is lower than 1
+     */
+    public function unique(int $for = 86400): self
+    {
+        if ($for < 1) {
+            throw new \InvalidArgumentException(sprintf('unique() expects a lock of at least 1 second, %d given.', $for));
+        }
+
+        $this->uniqueFor = $for;
+
+        return $this;
+    }
+
     /**
      * Queue name: the group with Action Scheduler, the queue with a Laravel queue. WP-Cron ignores it.
      */
@@ -276,6 +297,16 @@ public function runsAsUser(): bool
         return $this->asUser;
     }
 
+    /**
+     * @internal
+     *
+     * @return int|null Lock duration, null when the registration is not unique
+     */
+    public function uniqueFor(): ?int
+    {
+        return $this->uniqueFor;
+    }
+
     /**
      * @internal
      */

src/Async/UniqueLock.php

@@ -0,0 +1,102 @@
+<?php
+
+declare(strict_types=1);
+
+namespace Pollora\Hook\Async;
+
+use Pollora\Hook\Async\Drivers\WpCronDriver;
+use Pollora\Hook\Async\Exceptions\AsyncException;
+
+/**
+ * Time-limited lock that merges identical triggers of a unique registration.
+ *
+ * The lock is an option that is not autoloaded, created with add_option(): the
+ * unique key on option_name makes acquiring it atomic. Its value is the time it
+ * expires. It is released when the first attempt starts, and expired locks are
+ * deleted by the daily maintenance task.
+ *
+ * @internal
+ */
+final class UniqueLock
+{
+    public const string OPTION_PREFIX = 'pollora_unique_';
+
+    /**
+     * Identify a trigger: same hook, same handler, same arguments.
+     *
+     * @param  array<int|string, mixed>  $arguments  Normalized arguments
+     */
+    public static function key(string $hook, string $handler, array $arguments): string
+    {
+        return sha1((string) json_encode([$hook, $handler, $arguments]));
+    }
+
+    /**
+     * Take the lock, unless an identical trigger holds it.
+     *
+     * @param  int  $seconds  How long the lock lasts when it is never released
+     * @return bool False when an identical trigger already holds it
+     *
+     * @throws AsyncException When the lock cannot be stored
+     */
+    public static function acquire(string $key, int $seconds): bool
+    {
+        $option = self::OPTION_PREFIX.$key;
+        $expiresAt = time() + $seconds;
+
+        WpCronDriver::scheduleRecovery();
+
+        if (add_option($option, $expiresAt, '', false)) {
+            return true;
+        }
+
+        wp_cache_delete($option, 'options');
+        $current = get_option($option, null);
+
+        if ($current === null) {
+            throw new AsyncException('The unique lock of an asynchronous action could not be stored.');
+        }
+
+        if ((int) $current >= time()) {
+            return false;
+        }
+
+        // Expired: take it over. Of two processes doing so, add_option() lets one win.
+        delete_option($option);
+
+        return add_option($option, $expiresAt, '', false);
+    }
+
+    public static function release(string $key): void
+    {
+        delete_option(self::OPTION_PREFIX.$key);
+    }
+
+    /**
+     * Delete expired locks.
+     *
+     * @return int Number of locks deleted
+     */
+    public static function prune(int $limit = 500): int
+    {
+        global $wpdb;
+
+        if (! is_object($wpdb)) {
+            return 0;
+        }
+
+        $names = $wpdb->get_col($wpdb->prepare(
+            "SELECT option_name FROM {$wpdb->options} WHERE option_name LIKE %s AND CAST(option_value AS UNSIGNED) < %d LIMIT %d",
+            $wpdb->esc_like(self::OPTION_PREFIX).'%',
+            time(),
+            $limit,
+        ));
+
+        $deleted = 0;
+        foreach ((array) $names as $name) {
+            $deleted += delete_option((string) $name) ? 1 : 0;
+        }
+
+        return $deleted;
+    }
+}

tests/Stubs/wordpress.php

@@ -346,5 +346,32 @@ public function get_results(array $prepared, string $output): array
 
             return array_slice($rows, 0, $prepared['args'][1]);
         }
+
+        /**
+         * Answers the expired-lock query: prefix, time, limit.
+         */
+        public function get_col(array $prepared): array
+        {
+            [$like, $now, $limit] = $prepared['args'];
+            $prefix = stripslashes(rtrim($like, '%'));
+            $names = [];
+
+            foreach ($GLOBALS['wp_options'] ?? [] as $name => $option) {
+                if (str_starts_with($name, $prefix) && (int) $option['value'] < $now) {
+                    $names[] = $name;
+                }
+            }
+
+            return array_slice($names, 0, $limit);
+        }
     };
 }
+
+if (! function_exists('wp_cache_delete')) {
+    function wp_cache_delete(int|string $key, string $group = ''): bool
+    {
+        $GLOBALS['wp_cache_deleted'][] = [$key, $group];
+
+        return true;
+    }
+}