From 5fe859593bf5e081e2ec1a87e02cf4326724745f Mon Sep 17 00:00:00 2001 From: Jason Viljoen Date: Sun, 17 May 2026 22:55:05 +0200 Subject: [PATCH] feat: add LoggingListenerDispatcher and promote recoverStuck* onto store interfaces MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit LoggingListenerDispatcher logs every failed dispatch via PSR-3 with event + listener context and rethrows so upstream redelivery still triggers — gives operators per-attempt visibility into retries that SequentialRedeliveryProcessor previously exhausted silently. recoverStuckEvents() and recoverStuckRedeliveries() move from the SQL classes to their respective interfaces so recovery jobs can type-hint against the contract instead of instanceof-checking down. Co-Authored-By: Claude Opus 4.7 (1M context) --- README.md | 62 +++++++++++++---- src/EventStore.php | 7 +- .../Dispatch/LoggingListenerDispatcher.php | 34 +++++++++ src/Infrastructure/InMemoryEventStore.php | 6 ++ .../Redelivery/InMemoryRedeliveryStore.php | 19 +++++ .../Redelivery/SqlRedeliveryStore.php | 1 + src/Infrastructure/SqlEventStore.php | 1 + src/Redelivery/RedeliveryStore.php | 7 +- .../LoggingListenerDispatcherTest.php | 69 +++++++++++++++++++ tests/_Fixtures/CapturingEventStore.php | 6 ++ 10 files changed, 196 insertions(+), 16 deletions(-) create mode 100644 src/Infrastructure/Dispatch/LoggingListenerDispatcher.php create mode 100644 tests/Unit/Dispatch/LoggingListenerDispatcherTest.php diff --git a/README.md b/README.md index e3bf45d..7c15f64 100644 --- a/README.md +++ b/README.md @@ -62,6 +62,7 @@ composer require hivesper/php-events | `SilentEventProcessor` | Decorator — wraps any `EventProcessor` and logs (via PSR-3) any `Throwable` that escapes `process()` instead of letting it propagate. Use in production around the whole batch. | | `ListenerDispatcher` | Interface — invokes one (subscriber, event) pair. Returns `void`; lets the listener exception propagate. | | `DefaultListenerDispatcher` | Built-in — resolves the handler, hydrates the payload, calls the listener. Swallows exceptions listed in `ignoredExceptions`; otherwise rethrows. Stateless and side-effect-free. | +| `LoggingListenerDispatcher` | Decorator — wraps any `ListenerDispatcher` and on a throw logs at error level via PSR-3 (with the event name and listener key) then rethrows so upstream redelivery scheduling still triggers. Wrap **outside** `DefaultListenerDispatcher` so its `ignoredExceptions` never reach the logger. | | `RedeliveringListenerDispatcher` | Decorator — wraps any `ListenerDispatcher` and on a throw schedules a fresh row in the `RedeliveryStore` (attempt 1, retry-now). Used by the event-processing flow only — the redelivery processor must run a plain dispatcher to avoid double-scheduling. | | `RedeliveryProcessor` | Interface — drains a `RedeliveryStore`, mirroring `EventProcessor::process()` | | `SequentialRedeliveryProcessor` | Built-in redelivery processor — owns the `RetryPolicy` and decides per due row whether to reschedule, mark succeeded, or mark failed permanently | @@ -134,8 +135,10 @@ mostly-disjoint tables and rows, so they can run in parallel without contending. | `$eventStore->recoverStuckEvents(CarbonInterval::minutes(30))` | Every 5–15 minutes | Resets `event_outbox` rows wedged in `processing` (worker crash victims) back to `pending`. Threshold should be comfortably longer than your longest healthy dispatch — see [Recovering stuck events](#recovering-stuck-events). | | `$redeliveryStore->recoverStuckRedeliveries(CarbonInterval::minutes(30))` | Every 5–15 minutes | Resets `event_outbox_redelivery` rows wedged in `dispatching` (worker crash victims) back to `pending_retry`. Same threshold guidance as above. | -`recoverStuckEvents` and `recoverStuckRedeliveries` are SQL-specific (live on `SqlEventStore` / -`SqlRedeliveryStore`). The other two work against any `EventStore` / `RedeliveryStore` implementation. +`recoverStuckEvents` and `recoverStuckRedeliveries` live on the `EventStore` and +`RedeliveryStore` interfaces, so recovery jobs can type-hint against the interface. Only the SQL +implementations do meaningful work — `InMemoryEventStore::recoverStuckEvents()` returns `0` +(there is no persisted `processing` state to recover from). The four jobs do not compete for the same rows: @@ -409,12 +412,12 @@ The processor advances each event through three states: If a worker dies between `next()` and `markProcessed()`, the row stays in `processing` — intentionally. Any redelivery rows that *did* get persisted before the crash remain durable, so listener-level retries still fire when their time comes. To recover the wedged `processing` row -itself, call `SqlEventStore::recoverStuckEvents()` from a separate scheduled job (see +itself, call `EventStore::recoverStuckEvents()` from a separate scheduled job (see [Recovering stuck events](#recovering-stuck-events) below). ### Recovering stuck events -`SqlEventStore::recoverStuckEvents(CarbonInterval $olderThan): int` resets events that are stuck +`EventStore::recoverStuckEvents(CarbonInterval $olderThan): int` resets events that are stuck in `processing` back to `pending` so the next worker can claim them again. An event is "stuck" when its most recent `processing` audit row is older than `$olderThan`. The recovery writes a `pending` audit row tagged `Recovered from stuck processing state` so dashboards can distinguish @@ -439,7 +442,7 @@ unnecessary work is unnecessary work. ### Recovering stuck redeliveries The same shape applies to the redelivery table: -`SqlRedeliveryStore::recoverStuckRedeliveries(CarbonInterval $olderThan): int` resets rows +`RedeliveryStore::recoverStuckRedeliveries(CarbonInterval $olderThan): int` resets rows wedged in `dispatching` back to `pending_retry` so a worker can claim them again on the next `processNextRedelivery()` tick. A row is "stuck" when its `updated_at` is older than `$olderThan`. @@ -535,19 +538,17 @@ class MyProcessor implements EventProcessor ### Custom EventStore -If you implement your own `EventStore`, you need to satisfy the new `markProcessed()` method -alongside `add()` and `next()`. For an in-memory or queue-style store with no persisted status, -this is a one-liner: +If you implement your own `EventStore`, you need to satisfy `add()`, `next()`, `markProcessed()`, +and `recoverStuckEvents()`. For an in-memory or queue-style store with no persisted `processing` +status, the last two collapse to no-ops: ```php class MyEventStore implements EventStore { - public function add(RawEvent $event): void { /* ... */ } - public function next(): ?RawEvent { /* ... */ } - public function markProcessed(string $eventId): void - { - // No-op when there's no persisted status to flip. - } + public function add(RawEvent $event): void { /* ... */ } + public function next(): ?RawEvent { /* ... */ } + public function markProcessed(RawEvent $event): void { /* No-op when there's no persisted status to flip. */ } + public function recoverStuckEvents(CarbonInterval $olderThan): int { return 0; /* No rows to recover when there's no `processing` state. */ } } ``` @@ -631,6 +632,39 @@ $processor = new SilentEventProcessor( | Local / CI | `SequentialEventProcessor` + `DefaultListenerDispatcher` | A listener throw propagates out of `process()`. Nothing is hidden. | | Production | `SilentEventProcessor` wrapping `SequentialEventProcessor`, dispatcher = `RedeliveringListenerDispatcher(DefaultListenerDispatcher, $redeliveryStore)` | A listener throw is written to `event_outbox_redelivery` and the next listener runs. Anything else (DB blip, etc.) is logged and the batch aborts; the next scheduled tick picks it up. | +### Per-attempt failure logging + +`SequentialRedeliveryProcessor` exhausts retries silently — when the `RetryPolicy` returns +`null`, the row moves to `failed` with `last_error` set, but nothing is logged. To get +per-attempt visibility, wrap the inner dispatcher in `LoggingListenerDispatcher`. It logs every +failed dispatch at error level via PSR-3 with `['exception', 'event', 'listener']` context and +**rethrows**, so upstream redelivery scheduling/rescheduling still fires. + +Wrap it **outside** `DefaultListenerDispatcher` so its `ignoredExceptions` never reach the +logger, and **inside** `RedeliveringListenerDispatcher` (event path) / +`SequentialRedeliveryProcessor` (redelivery path) so every attempt is logged: + +```php +use Vesper\Tool\Event\Infrastructure\Dispatch\DefaultListenerDispatcher; +use Vesper\Tool\Event\Infrastructure\Dispatch\LoggingListenerDispatcher; +use Vesper\Tool\Event\Infrastructure\Dispatch\RedeliveringListenerDispatcher; + +$logged = new LoggingListenerDispatcher(new DefaultListenerDispatcher(), $logger); + +// Event path: log + persist failure as a redelivery row. +$eventDispatcher = new RedeliveringListenerDispatcher($logged, $redeliveryStore); + +// Redelivery path: pass $logged directly — the redelivery processor handles reschedule/fail. +$redeliveryProcessor = new SequentialRedeliveryProcessor( + $subscribers, + $logged, + new ExponentialBackoffRetryPolicy(), +); +``` + +With exponential backoff and a deduplicating log backend, per-attempt log volume stays bounded +while still giving a trail of differing errors across retries. + --- ## Automatic retry & failure tracking diff --git a/src/EventStore.php b/src/EventStore.php index 29fef03..c6db18d 100644 --- a/src/EventStore.php +++ b/src/EventStore.php @@ -2,6 +2,8 @@ namespace Vesper\Tool\Event; +use Carbon\CarbonInterval; + interface EventStore { /** Call inside the caller's business transaction so event + business change commit together. */ @@ -10,9 +12,12 @@ public function add(RawEvent $event): void; /** * Claim the next pending event, transitioning it from `pending` to `processing`. Returns null * when nothing is due. A worker that dies before calling markProcessed() leaves the row in - * `processing` — call SqlEventStore::recoverStuckEvents() from a separate cron to recover it. + * `processing` — call EventStore::recoverStuckEvents() from a separate cron to recover it. */ public function next(): ?RawEvent; public function markProcessed(RawEvent $event): void; + + /** Resets events wedged in `processing` back to `pending`; returns the count recovered. */ + public function recoverStuckEvents(CarbonInterval $olderThan): int; } diff --git a/src/Infrastructure/Dispatch/LoggingListenerDispatcher.php b/src/Infrastructure/Dispatch/LoggingListenerDispatcher.php new file mode 100644 index 0000000..f02ff2f --- /dev/null +++ b/src/Infrastructure/Dispatch/LoggingListenerDispatcher.php @@ -0,0 +1,34 @@ +inner->dispatch($event, $subscriber); + } catch (Throwable $e) { + $this->logger->error('Listener dispatch failed.', [ + 'exception' => $e, + 'event' => $event->name, + 'listener' => ListenerKey::of($subscriber), + ]); + + throw $e; + } + } +} diff --git a/src/Infrastructure/InMemoryEventStore.php b/src/Infrastructure/InMemoryEventStore.php index 0cf9cdc..0690a1d 100644 --- a/src/Infrastructure/InMemoryEventStore.php +++ b/src/Infrastructure/InMemoryEventStore.php @@ -2,6 +2,7 @@ namespace Vesper\Tool\Event\Infrastructure; +use Carbon\CarbonInterval; use Override; use Vesper\Tool\Event\EventStore; use Vesper\Tool\Event\RawEvent; @@ -27,4 +28,9 @@ class InMemoryEventStore implements EventStore { // No-op: the in-memory queue discards events on next(); there is no persisted status to flip. } + + #[Override] public function recoverStuckEvents(CarbonInterval $olderThan): int + { + return 0; + } } diff --git a/src/Infrastructure/Redelivery/InMemoryRedeliveryStore.php b/src/Infrastructure/Redelivery/InMemoryRedeliveryStore.php index 739e0e6..2be9611 100644 --- a/src/Infrastructure/Redelivery/InMemoryRedeliveryStore.php +++ b/src/Infrastructure/Redelivery/InMemoryRedeliveryStore.php @@ -3,6 +3,7 @@ namespace Vesper\Tool\Event\Infrastructure\Redelivery; use Carbon\CarbonImmutable; +use Carbon\CarbonInterval; use Override; use Vesper\Tool\Event\Redelivery\Redelivery; use Vesper\Tool\Event\Redelivery\RedeliveryStore; @@ -75,6 +76,24 @@ public function retryNow(string $eventId, string $listener): void $this->rows[$key] = $this->rows[$key]->queueForImmediateRetry(); } + #[Override] + public function recoverStuckRedeliveries(CarbonInterval $olderThan): int + { + $threshold = CarbonImmutable::now()->sub($olderThan); + $recovered = 0; + + foreach ($this->rows as $key => $row) { + if (!$row->isDispatching() || !$row->updatedAt->lessThan($threshold)) { + continue; + } + + $this->rows[$key] = $row->queueForImmediateRetry(); + $recovered++; + } + + return $recovered; + } + private static function key(string $eventId, string $listener): string { return $eventId . '|' . $listener; diff --git a/src/Infrastructure/Redelivery/SqlRedeliveryStore.php b/src/Infrastructure/Redelivery/SqlRedeliveryStore.php index 6719df5..acf0a24 100644 --- a/src/Infrastructure/Redelivery/SqlRedeliveryStore.php +++ b/src/Infrastructure/Redelivery/SqlRedeliveryStore.php @@ -129,6 +129,7 @@ public function retryNow(string $eventId, string $listener): void * when its updated_at is older than $olderThan. Returns the number of rows recovered. * Call from a separate scheduled job; safe to run alongside the redelivery worker. */ + #[Override] public function recoverStuckRedeliveries(CarbonInterval $olderThan): int { $now = CarbonImmutable::now()->format('Y-m-d H:i:s.u'); diff --git a/src/Infrastructure/SqlEventStore.php b/src/Infrastructure/SqlEventStore.php index 78ebd9f..7661372 100644 --- a/src/Infrastructure/SqlEventStore.php +++ b/src/Infrastructure/SqlEventStore.php @@ -110,6 +110,7 @@ public function markProcessed(RawEvent $event): void * transitions apart. Returns the number of events recovered. Call from a separate scheduled * job; safe to run alongside the main worker. */ + #[Override] public function recoverStuckEvents(CarbonInterval $olderThan): int { $thresholdAt = CarbonImmutable::now()->sub($olderThan)->format('Y-m-d H:i:s.u'); diff --git a/src/Redelivery/RedeliveryStore.php b/src/Redelivery/RedeliveryStore.php index 5b9ae13..a62c55c 100644 --- a/src/Redelivery/RedeliveryStore.php +++ b/src/Redelivery/RedeliveryStore.php @@ -2,6 +2,8 @@ namespace Vesper\Tool\Event\Redelivery; +use Carbon\CarbonInterval; + interface RedeliveryStore { /** @@ -16,7 +18,7 @@ public function schedule(Redelivery $redelivery): void; * Claim the next due redelivery, atomically transitioning it from pending_retry to * dispatching so concurrent workers cannot pick up the same row. Returns null when * nothing is due. A worker that dies before calling update() leaves the row in - * dispatching — call SqlRedeliveryStore::recoverStuckRedeliveries() from a separate + * dispatching — call RedeliveryStore::recoverStuckRedeliveries() from a separate * cron to recover it. */ public function next(): ?Redelivery; @@ -33,4 +35,7 @@ public function update(Redelivery $redelivery): void; * is preserved, so the retry policy's max-attempts ceiling still applies. */ public function retryNow(string $eventId, string $listener): void; + + /** Resets redeliveries wedged in `dispatching` back to `pending_retry`; returns the count recovered. */ + public function recoverStuckRedeliveries(CarbonInterval $olderThan): int; } diff --git a/tests/Unit/Dispatch/LoggingListenerDispatcherTest.php b/tests/Unit/Dispatch/LoggingListenerDispatcherTest.php new file mode 100644 index 0000000..6df3c9f --- /dev/null +++ b/tests/Unit/Dispatch/LoggingListenerDispatcherTest.php @@ -0,0 +1,69 @@ +createMock(LoggerInterface::class); + $logger->expects($this->never())->method('error'); + + $dispatcher = new LoggingListenerDispatcher(self::passthroughDispatcher(), $logger); + + $dispatcher->dispatch($event, function () {}); + } + + public function test_logs_and_rethrows_when_inner_throws(): void + { + $event = TestEventFactory::retrieveOrderPlaced(); + $exception = new RuntimeException('boom'); + + $logger = $this->createMock(LoggerInterface::class); + $logger->expects($this->once()) + ->method('error') + ->with( + 'Listener dispatch failed.', + [ + 'exception' => $exception, + 'event' => $event->name, + 'listener' => ThrowingListener::class, + ], + ); + + $dispatcher = new LoggingListenerDispatcher(self::throwingDispatcher($exception), $logger); + + $this->expectExceptionObject($exception); + + $dispatcher->dispatch($event, ThrowingListener::class); + } + + private static function passthroughDispatcher(): ListenerDispatcher + { + return new readonly class implements ListenerDispatcher { + public function dispatch(RawEvent $event, callable|string $subscriber): void {} + }; + } + + private static function throwingDispatcher(Throwable $error): ListenerDispatcher + { + return new readonly class ($error) implements ListenerDispatcher { + public function __construct(private Throwable $error) {} + public function dispatch(RawEvent $event, callable|string $subscriber): void + { + throw $this->error; + } + }; + } +} diff --git a/tests/_Fixtures/CapturingEventStore.php b/tests/_Fixtures/CapturingEventStore.php index 9594985..20a8901 100644 --- a/tests/_Fixtures/CapturingEventStore.php +++ b/tests/_Fixtures/CapturingEventStore.php @@ -2,6 +2,7 @@ namespace Test\Vesper\Tool\Event\_Fixtures; +use Carbon\CarbonInterval; use Override; use Vesper\Tool\Event\EventStore; use Vesper\Tool\Event\RawEvent; @@ -22,4 +23,9 @@ class CapturingEventStore implements EventStore } #[Override] public function markProcessed(RawEvent $event): void {} + + #[Override] public function recoverStuckEvents(CarbonInterval $olderThan): int + { + return 0; + } }