Merge pull request #562 from hypervel/fix/pre-outbox-framework-correctness

Fix queue publication and worker lifecycle correctness

Author: binaryfire

docs/plans/2026-09-04-1527-components-pre-outbox-framework-correctness.md

@@ -21,7 +21,7 @@
 ## Framework-wide
 
 - Resolve the production dependency cycle between `hypervel/foundation` and `hypervel/testing`, which currently causes test-only classes, PHPUnit integration, and Mockery to propagate through `hypervel/support` into every package that depends on Support. Decide whether Foundation's testing namespace moves to `hypervel/testing` or Foundation changes its dependency to development-only while preserving split-package installation and public namespaces.
-- Design a connection-owned service identity and capability API for external backends that packages can query without repeated hot-path probes. Redis/Valkey and database connections already expose fragments of this information in different forms; prefer lazy detection cached for the current connection or pool generation, with invalidation on reconnect and purge, over an eager process-global startup registry that performs unused I/O or survives a backend change. Start with concrete consumers and capability checks rather than a universal version-comparison abstraction. Database Queue's `SKIP LOCKED` selection is one current consumer: MySQL/MariaDB engine and version detection runs on every pop and must not be cached on the worker-lived queue object.
+- Design a connection-owned service identity and capability API for external backends that packages can query without repeated hot-path probes. Redis/Valkey and database connections already expose fragments of this information in different forms; prefer lazy detection cached for the current connection or pool generation, with invalidation on reconnect and purge, over an eager process-global startup registry that performs unused I/O or survives a backend change. Start with concrete consumers and capability checks rather than a universal version-comparison abstraction.
 - Find a clean, simple framework-wide solution for configuration-dependent services resolved before worker configuration reload. `server:reload` refreshes the existing configuration repository, but objects that have already copied configuration into their own state remain stale. For example, `SentryServiceProvider` eagerly resolves a worker-lifetime Hub and client during boot, so DSN, environment, and sampling changes are not applied until a full restart; resolving `Cache::store('some-store')` from a service provider populates `CacheManager`'s store cache before reload, so changes to that store's driver, connection, prefix, or other captured configuration are likewise not applied. Define the reload contract, audit framework-owned eager resolutions and manager caches, and solve the lifecycle at their shared owning boundary instead of adding package-specific refresh hooks or application workarounds.
 - Investigate where requiring and directly using a PHP extension would make framework code significantly faster than its current pure-PHP implementation. The framework already declares bundled extensions it depends on, so the question is which hot paths are doing in PHP what a C extension does natively. The worked example is `ext-gmp` for identifier encoding: UUID and ULID string conversion and any base32/base58/base62 short-id work exceed 64 bits, so `ramsey/uuid` and `symfony/uid` convert them digit by digit in PHP, while `gmp_init()`/`gmp_strval()` do arbitrary-base conversion natively — a hand-rolled base-36 UUID conversion measured 14.6 µs against 0.5 µs for the GMP equivalent with byte-identical output. Anything that fits in a 64-bit int (snowflakes, timestamps, counters) needs no extension, and hashing, encryption, and signatures are already C. Measure `Str::uuid()`/`Str::ulid()` and the other candidates before adding a requirement, and weigh each new extension against installation cost.
 - Convert untyped `$config->get()` calls across `src/` to the typed getters (`string()`, `integer()`, `float()`, `boolean()`, `array()`) without call-site defaults, for every key that isn't genuinely nullable. Defaults live in the merged config files — declare any key currently defaulted only at a call site in its package's config file as part of the conversion. Typed getters throw `InvalidArgumentException` naming the key on misconfiguration instead of letting a wrong type propagate silently, and give phpstan real return types. Bootstrap code that runs before config merging keeps its call-site defaults. Approved modernization per the Porting Packages policy in `AGENTS.md`; new code already follows the rule.

docs/todo.md

@@ -12,6 +12,7 @@
 use Hypervel\Broadcasting\Broadcasters\NullBroadcaster;
 use Hypervel\Broadcasting\Broadcasters\PusherBroadcaster;
 use Hypervel\Broadcasting\Broadcasters\RedisBroadcaster;
+use Hypervel\Bus\DispatchLockContext;
 use Hypervel\Bus\UniqueLock;
 use Hypervel\Contracts\Broadcasting\Broadcaster;
 use Hypervel\Contracts\Broadcasting\Factory as BroadcastingFactoryContract;
@@ -237,6 +238,17 @@ public function queue(mixed $event): void
             )
             ->pushOn($queue, $broadcastEvent);
 
+        if ($event instanceof ShouldBeUnique) {
+            $dispatch = $push;
+            $push = static function () use ($broadcastEvent, $dispatch): mixed {
+                try {
+                    return $dispatch();
+                } finally {
+                    DispatchLockContext::release($broadcastEvent);
+                }
+            };
+        }
+
         $event instanceof ShouldRescue
             ? $this->rescue($push)
             : $push();
@@ -245,13 +257,10 @@ public function queue(mixed $event): void
     /**
      * Determine if the broadcastable event must be unique and determine if we can acquire the necessary lock.
      */
-    protected function mustBeUniqueAndCannotAcquireLock(mixed $event): bool
+    protected function mustBeUniqueAndCannotAcquireLock(object $event): bool
     {
-        return ! (new UniqueLock(
-            method_exists($event, 'uniqueVia')
-                ? $event->uniqueVia()
-                : $this->app->make(Cache::class)
-        ))->acquire($event);
+        return ! (new UniqueLock($this->app->make(Cache::class)))
+            ->acquireForDispatch($event);
     }
 
     /**

src/broadcasting/src/BroadcastManager.php

@@ -43,7 +43,7 @@ public function __construct(mixed $event)
     /**
      * Resolve the cache implementation that should manage the event's uniqueness.
      */
-    public function uniqueVia(): Repository
+    public function uniqueVia(): ?Repository
     {
         return method_exists($this->event, 'uniqueVia')
             ? $this->event->uniqueVia()

src/broadcasting/src/UniqueBroadcastEvent.php

@@ -3,10 +3,12 @@ Bus for Hypervel
 
 [![Ask DeepWiki](https://deepwiki.com/badge.svg)](https://deepwiki.com/hypervel/bus)
 
-Ported from: https://github.com/laravel/framework/tree/13.x/src/Illuminate/Bus
+Documentation: https://hypervel.org/docs/queues
 
 ## Differences From Laravel
 
 Hypervel does not include Laravel's DynamoDB batch repository because DynamoDB is not a supported database backend.
 
 `DatabaseBatchRepository::setConnection()` is intentionally omitted. The repository is shared for the worker lifetime, so mutating its connection would race across coroutines. Configure `queue.batching.database` instead; each repository operation resolves that connection when it runs.
+
+Ported from: https://github.com/laravel/framework/tree/13.x/src/Illuminate/Bus

src/bus/README.md

@@ -9,6 +9,7 @@
 use Hypervel\Queue\Attributes\ReadsQueueAttributes;
 use Hypervel\Support\CarbonImmutable;
 use Hypervel\Support\Str;
+use Throwable;
 
 class DebounceLock
 {
@@ -31,20 +32,72 @@ public function __construct(
      */
     public function acquire(mixed $job, ?int $debounceFor = null, ?int $maxWait = null): array
     {
-        $cache = $this->resolveCache($job);
+        [$cache, $key, $ttl, $maxWait] = $this->resolveLockValues($job, $debounceFor, $maxWait);
+
+        return $this->acquireResolvedLock($cache, $key, $ttl, $maxWait);
+    }
+
+    /**
+     * Store and register a debounce owner token for a dispatch operation.
+     *
+     * @return array{owner: string, maxWaitExceeded: bool}
+     */
+    public function acquireForDispatch(object $job, ?int $debounceFor = null, ?int $maxWait = null): array
+    {
+        [$cache, $key, $ttl, $maxWait] = $this->resolveLockValues($job, $debounceFor, $maxWait);
+        $result = $this->acquireResolvedLock($cache, $key, $ttl, $maxWait);
+
+        if (isset(class_uses_recursive($job)[Queueable::class])) {
+            $job->debounceOwner = $result['owner'];
+        }
+
+        DispatchLockContext::registerDebounce($job, $cache, $key, $result['owner']);
 
+        return $result;
+    }
+
+    /**
+     * Resolve the values needed to acquire a debounce lock.
+     *
+     * @return array{Cache, string, int, null|int}
+     */
+    protected function resolveLockValues(mixed $job, ?int $debounceFor, ?int $maxWait): array
+    {
+        $cache = $this->resolveCache($job);
         $ttl = max(($debounceFor ?? $this->getDebounceDelay($job)) * 10, 300);
 
-        $cache->put($key = static::getKey($job), $owner = Str::random(40), $ttl);
+        return [
+            $cache,
+            static::getKey($job),
+            $ttl,
+            $maxWait ?? $this->getMaxDebounceWait($job),
+        ];
+    }
+
+    /**
+     * Store a debounce owner token using resolved values.
+     *
+     * @return array{owner: string, maxWaitExceeded: bool}
+     */
+    protected function acquireResolvedLock(Cache $cache, string $key, int $ttl, ?int $maxWait): array
+    {
+        $cache->put($key, $owner = Str::random(40), $ttl);
+
+        try {
+            $maxWaitExceeded = $this->maxWaitExceeded($cache, $key, $ttl, $maxWait);
+        } catch (Throwable $exception) {
+            try {
+                static::releaseOwned($cache, $key, $owner);
+            } catch (Throwable) {
+                // A cleanup failure must not replace the acquisition failure.
+            }
+
+            throw $exception;
+        }
 
         return [
             'owner' => $owner,
-            'maxWaitExceeded' => $this->maxWaitExceeded(
-                $cache,
-                $key,
-                $ttl,
-                $maxWait ?? $this->getMaxDebounceWait($job)
-            ),
+            'maxWaitExceeded' => $maxWaitExceeded,
         ];
     }
 
@@ -93,16 +146,47 @@ public function getCurrentOwner(mixed $job): ?string
      */
     public function release(mixed $job, string $owner = ''): void
     {
-        $key = static::getKey($job);
-
         $cache = $this->resolveCache($job);
 
+        static::releaseOwned($cache, static::getKey($job), $owner);
+    }
+
+    /**
+     * Remove the maximum wait timestamp for the given job.
+     */
+    public function releaseMaxWait(mixed $job): void
+    {
+        $this->resolveCache($job)->forget(static::getKey($job) . ':first_dispatched_at');
+    }
+
+    /**
+     * Release a debounce lock from already-resolved provenance.
+     *
+     * @internal
+     */
+    public static function releaseOwned(Cache $cache, string $key, string $owner = ''): void
+    {
         if ($owner !== '' && $cache->get($key) !== $owner) {
             return;
         }
 
-        $cache->forget($key);
-        $cache->forget($key . ':first_dispatched_at');
+        $exception = null;
+
+        try {
+            $cache->forget($key);
+        } catch (Throwable $throwable) {
+            $exception = $throwable;
+        }
+
+        try {
+            $cache->forget($key . ':first_dispatched_at');
+        } catch (Throwable $throwable) {
+            $exception ??= $throwable;
+        }
+
+        if ($exception !== null) {
+            throw $exception;
+        }
     }
 
     /**