feat: improve deferred event processing

Signed-off-by: Sebastian Krupinski <krupinski01@gmail.com>

core/lib/Application/Execution/TerminationReport.php

@@ -15,6 +15,9 @@ final readonly class TerminationReport
         public array $failures = [],
         public bool $deadlineExceeded = false,
         public bool $limitExceeded = false,
+        public int $deferredListenerInvocations = 0,
+        public bool $deferredEventLimitExceeded = false,
+        public bool $deferredListenerInvocationLimitExceeded = false,
     ) {
     }
 }

core/lib/Event/DeferredProcessingResult.php

@@ -11,6 +11,9 @@ final readonly class DeferredProcessingResult
         public int $remaining,
         public bool $deadlineExceeded,
         public bool $limitExceeded = false,
+        public int $listenerInvocations = 0,
+        public bool $eventLimitExceeded = false,
+        public bool $listenerInvocationLimitExceeded = false,
     ) {
     }
 }

core/lib/Event/EventDispatcher.php

@@ -13,6 +13,10 @@ use Psr\Log\LoggerInterface;
 
 final class EventDispatcher implements EventDispatcherInterface, DeferredEventProcessorInterface
 {
+    private const DEFAULT_DEFERRED_PROCESSING_TIMEOUT_SECONDS = 300.0;
+    private const DEFAULT_MAX_DEFERRED_EVENTS = 1000;
+    private const DEFAULT_MAX_DEFERRED_LISTENER_INVOCATIONS = 50000;
+
     /** @var array<string, list<Event>> */
     private array $deferred = [];
     private ?string $activeExecution = null;
@@ -22,7 +26,19 @@ final class EventDispatcher implements EventDispatcherInterface, DeferredEventPr
         private readonly EventListenerRegistry $registry,
         private readonly ContainerInterface $container,
         private readonly LoggerInterface $logger,
+        private readonly float $deferredProcessingTimeoutSeconds = self::DEFAULT_DEFERRED_PROCESSING_TIMEOUT_SECONDS,
+        private readonly int $maxDeferredEvents = self::DEFAULT_MAX_DEFERRED_EVENTS,
+        private readonly int $maxDeferredListenerInvocations = self::DEFAULT_MAX_DEFERRED_LISTENER_INVOCATIONS,
     ) {
+        if ($this->deferredProcessingTimeoutSeconds <= 0) {
+            throw new \InvalidArgumentException('The deferred processing timeout must be greater than zero.');
+        }
+        if ($this->maxDeferredEvents <= 0) {
+            throw new \InvalidArgumentException('The deferred event limit must be greater than zero.');
+        }
+        if ($this->maxDeferredListenerInvocations <= 0) {
+            throw new \InvalidArgumentException('The deferred listener invocation limit must be greater than zero.');
+        }
     }
 
     public function dispatch(Event $event): void
@@ -61,13 +77,15 @@ final class EventDispatcher implements EventDispatcherInterface, DeferredEventPr
         }
 
         try {
-            $processed = 0;
+            $processedEvents = 0;
-            $deadline = microtime(true) + 1.0;
+            $listenerInvocations = 0;
+            $deadline = microtime(true) + $this->deferredProcessingTimeoutSeconds;
             $deadlineExceeded = false;
-            $limitExceeded = false;
+            $eventLimitExceeded = false;
+            $listenerInvocationLimitExceeded = false;
             while (($event = array_shift($this->deferred[$executionId])) !== null) {
-                if ($processed >= 1000) {
+                if ($processedEvents >= $this->maxDeferredEvents) {
-                    $limitExceeded = true;
+                    $eventLimitExceeded = true;
                     array_unshift($this->deferred[$executionId], $event);
                     break;
                 }
@@ -76,14 +94,29 @@ final class EventDispatcher implements EventDispatcherInterface, DeferredEventPr
                     array_unshift($this->deferred[$executionId], $event);
                     break;
                 }
-                $processed += $this->invoke($event, DeliveryMode::Deferred);
+
+                $eventListenerCount = count($this->registry->listeners(
+                    $event->getName(),
+                    DeliveryMode::Deferred,
+                ));
+                if ($listenerInvocations + $eventListenerCount > $this->maxDeferredListenerInvocations) {
+                    $listenerInvocationLimitExceeded = true;
+                    array_unshift($this->deferred[$executionId], $event);
+                    break;
+                }
+
+                $listenerInvocations += $this->invoke($event, DeliveryMode::Deferred);
+                $processedEvents++;
             }
 
             return new DeferredProcessingResult(
-                $processed,
+                processed: $processedEvents,
-                count($this->deferred[$executionId]),
+                remaining: count($this->deferred[$executionId]),
-                $deadlineExceeded,
+                deadlineExceeded: $deadlineExceeded,
-                $limitExceeded,
+                limitExceeded: $eventLimitExceeded || $listenerInvocationLimitExceeded,
+                listenerInvocations: $listenerInvocations,
+                eventLimitExceeded: $eventLimitExceeded,
+                listenerInvocationLimitExceeded: $listenerInvocationLimitExceeded,
             );
         } finally {
             $this->discardDeferred($executionId);
@@ -106,10 +139,10 @@ final class EventDispatcher implements EventDispatcherInterface, DeferredEventPr
                 break;
             }
 
+            $processed++;
             try {
                 $service = $this->container->get($listener->service);
                 $service->{$listener->method}($event);
-                $processed++;
             } catch (\Throwable $error) {
                 $this->logger->error('Event listener failed.', [
                     'event' => $event->getName(),

core/lib/Kernel.php

@@ -238,6 +238,9 @@ class Kernel implements KernelInterface
         $remaining = 0;
         $deadlineExceeded = false;
         $limitExceeded = false;
+        $listenerInvocations = 0;
+        $eventLimitExceeded = false;
+        $listenerInvocationLimitExceeded = false;
         $failures = [];
 
         try {
@@ -249,6 +252,9 @@ class Kernel implements KernelInterface
                 $remaining = $result->remaining;
                 $deadlineExceeded = $result->deadlineExceeded;
                 $limitExceeded = $result->limitExceeded;
+                $listenerInvocations = $result->listenerInvocations;
+                $eventLimitExceeded = $result->eventLimitExceeded;
+                $listenerInvocationLimitExceeded = $result->listenerInvocationLimitExceeded;
             }
         } catch (\Throwable $e) {
             $failures[] = $e;
@@ -287,6 +293,9 @@ class Kernel implements KernelInterface
             failures: $failures,
             deadlineExceeded: $deadlineExceeded,
             limitExceeded: $limitExceeded,
+            deferredListenerInvocations: $listenerInvocations,
+            deferredEventLimitExceeded: $eventLimitExceeded,
+            deferredListenerInvocationLimitExceeded: $listenerInvocationLimitExceeded,
         );
     }
 

tests/php/Unit/Event/EventDispatcherTest.php

@@ -230,9 +230,57 @@ final class EventDispatcherTest extends TestCase
         $result = $dispatcher->processDeferred('test');
 
         self::assertSame(1000, $result->processed);
+        self::assertSame(1000, $result->listenerInvocations);
         self::assertSame(1, $result->remaining);
         self::assertFalse($result->deadlineExceeded);
         self::assertTrue($result->limitExceeded);
+        self::assertTrue($result->eventLimitExceeded);
+        self::assertFalse($result->listenerInvocationLimitExceeded);
+    }
+
+    #[Test]
+    #[TestDox('Deferred event and listener invocation limits are tracked separately')]
+    public function boundsDeferredListenerInvocations(): void
+    {
+        $recursive = new RecursiveListener();
+        $recording = new RecordingListener();
+        $registry = new EventListenerRegistry();
+        $registry->listen(
+            'test',
+            'test.event',
+            RecursiveListener::class,
+            'deferred',
+            DeliveryMode::Deferred,
+        );
+        $registry->listen(
+            'test',
+            'test.event',
+            RecordingListener::class,
+            'deferred',
+            DeliveryMode::Deferred,
+        );
+        $registry->freeze();
+        $dispatcher = new EventDispatcher(
+            $registry,
+            new RecordingContainer([
+                RecursiveListener::class => $recursive,
+                RecordingListener::class => $recording,
+            ]),
+            new NullLogger(),
+            maxDeferredListenerInvocations: 3,
+        );
+        $recursive->dispatcher = $dispatcher;
+        $dispatcher->beginExecution('test');
+        $dispatcher->dispatch(new Event('test.event'));
+
+        $result = $dispatcher->processDeferred('test');
+
+        self::assertSame(1, $result->processed);
+        self::assertSame(2, $result->listenerInvocations);
+        self::assertSame(1, $result->remaining);
+        self::assertTrue($result->limitExceeded);
+        self::assertFalse($result->eventLimitExceeded);
+        self::assertTrue($result->listenerInvocationLimitExceeded);
     }
 }