fix(events): close deferred scope after listener failure

Signed-off-by: Sebastian Krupinski <krupinski01@gmail.com>
This commit is contained in:
2026-08-06 00:02:15 -04:00
parent 84eb0e2c21
commit 5b3e8f6588
2 changed files with 63 additions and 26 deletions
+5 -5
View File
@@ -60,6 +60,7 @@ final class EventDispatcher implements EventDispatcherInterface, DeferredEventPr
throw new \LogicException('Cannot process deferred events for an inactive execution.');
}
try {
$processed = 0;
$deadline = microtime(true) + 1.0;
$deadlineExceeded = false;
@@ -78,16 +79,15 @@ final class EventDispatcher implements EventDispatcherInterface, DeferredEventPr
$processed += $this->invoke($event, DeliveryMode::Deferred);
}
$remaining = count($this->deferred[$executionId]);
unset($this->deferred[$executionId]);
$this->activeExecution = null;
return new DeferredProcessingResult(
$processed,
$remaining,
count($this->deferred[$executionId]),
$deadlineExceeded,
$limitExceeded,
);
} finally {
$this->discardDeferred($executionId);
}
}
public function discardDeferred(string $executionId): void
@@ -167,6 +167,43 @@ final class EventDispatcherTest extends TestCase
$dispatcher->dispatch(new Event('test.event'));
}
#[Test]
#[TestDox('Propagated deferred failures close their execution scope')]
public function recoversFromDeferredFailure(): void
{
$registry = new EventListenerRegistry();
$registry->listen(
'test',
'test.event',
FailingListener::class,
'fail',
DeliveryMode::Deferred,
failurePolicy: FailurePolicy::Propagate,
);
$registry->freeze();
$dispatcher = new EventDispatcher(
$registry,
new RecordingContainer([FailingListener::class => new FailingListener()]),
new NullLogger(),
);
$dispatcher->beginExecution('failed');
$dispatcher->dispatch(new Event('test.event'));
try {
$dispatcher->processDeferred('failed');
self::fail('Expected deferred listener failure to propagate.');
} catch (\RuntimeException $error) {
self::assertSame('Listener failed.', $error->getMessage());
}
$dispatcher->beginExecution('next');
$result = $dispatcher->processDeferred('next');
self::assertSame(0, $result->processed);
self::assertSame(0, $result->remaining);
}
#[Test]
#[TestDox('Deferred processing stops at its configured count limit')]
public function boundsDeferredWork(): void