fix(flow-php/telemetry): bound profiler signal store to prevent worker OO

- add opt-in per-signal cap to MemoryExporter (drops oldest)
- cap profiler store and reset it per messenger worker cycle

Author: norberttech

src/bridge/symfony/telemetry-bundle/src/Flow/Bridge/Symfony/TelemetryBundle/Profiler/ProfilerStoreWorkerResetSubscriber.php

@@ -0,0 +1,38 @@
+<?php
+
+declare(strict_types=1);
+
+namespace Flow\Bridge\Symfony\TelemetryBundle\Profiler;
+
+use Flow\Telemetry\Provider\Memory\MemoryExporter;
+use Symfony\Component\EventDispatcher\EventSubscriberInterface;
+use Symfony\Component\Messenger\Event\WorkerRunningEvent;
+
+/**
+ * The profiler store accumulates every signal teed into it and is normally emptied by the data collector
+ * between HTTP requests. A messenger worker is a single long-lived process with no request cycle, so the
+ * collector never resets the store and it grows until the process runs out of memory. Resetting it after
+ * each worker cycle bounds it to a single message, mirroring how Symfony resets data collectors per message.
+ *
+ * The event only fires when symfony/messenger is installed; otherwise this subscriber is inert.
+ */
+final readonly class ProfilerStoreWorkerResetSubscriber implements EventSubscriberInterface
+{
+    private const RESET_PRIORITY = 512;
+
+    public function __construct(
+        private MemoryExporter $store,
+    ) {}
+
+    public static function getSubscribedEvents(): array
+    {
+        return [
+            WorkerRunningEvent::class => ['onWorkerRunning', self::RESET_PRIORITY],
+        ];
+    }
+
+    public function onWorkerRunning(WorkerRunningEvent $event): void
+    {
+        $this->store->reset();
+    }
+}

src/bridge/symfony/telemetry-bundle/src/Flow/Bridge/Symfony/TelemetryBundle/Resources/config/profiler.php

@@ -3,6 +3,7 @@
 declare(strict_types=1);
 
 use Flow\Bridge\Symfony\TelemetryBundle\Profiler\FlowTelemetryDataCollector;
+use Flow\Bridge\Symfony\TelemetryBundle\Profiler\ProfilerStoreWorkerResetSubscriber;
 use Flow\Telemetry\Provider\Memory\MemoryExporter;
 use Flow\Telemetry\Telemetry;
 use Symfony\Component\DependencyInjection\Loader\Configurator\ContainerConfigurator;
@@ -12,7 +13,12 @@
 return static function (ContainerConfigurator $container): void {
     $services = $container->services();
 
-    $services->set('flow.telemetry.profiler.store', MemoryExporter::class)->public();
+    $services->set('flow.telemetry.profiler.store', MemoryExporter::class)->args([10000])->public();
+
+    $services
+        ->set('flow.telemetry.profiler.store_worker_reset_subscriber', ProfilerStoreWorkerResetSubscriber::class)
+        ->args([service('flow.telemetry.profiler.store')])
+        ->tag('kernel.event_subscriber');
 
     $services->set('flow.telemetry.profiler.collector', FlowTelemetryDataCollector::class)->args([
         service(Telemetry::class),

src/bridge/symfony/telemetry-bundle/tests/Flow/Bridge/Symfony/TelemetryBundle/Tests/Unit/Profiler/ProfilerStoreWorkerResetSubscriberTest.php

@@ -0,0 +1,50 @@
+<?php
+
+declare(strict_types=1);
+
+namespace Flow\Bridge\Symfony\TelemetryBundle\Tests\Unit\Profiler;
+
+use Flow\Bridge\Symfony\TelemetryBundle\Profiler\ProfilerStoreWorkerResetSubscriber;
+use Flow\Telemetry\Provider\Memory\MemoryExporter;
+use Flow\Telemetry\Signal\Signals;
+use Flow\Telemetry\Tests\Mother\SpanMother;
+use PHPUnit\Framework\Attributes\CoversClass;
+use PHPUnit\Framework\TestCase;
+use Symfony\Component\Messenger\Event\WorkerRunningEvent;
+use Symfony\Component\Messenger\MessageBus;
+use Symfony\Component\Messenger\Worker;
+
+use function class_exists;
+
+#[CoversClass(ProfilerStoreWorkerResetSubscriber::class)]
+final class ProfilerStoreWorkerResetSubscriberTest extends TestCase
+{
+    protected function setUp(): void
+    {
+        if (!class_exists(Worker::class)) {
+            self::markTestSkipped('symfony/messenger is not installed');
+        }
+    }
+
+    public function test_it_empties_the_store_after_a_worker_cycle(): void
+    {
+        $store = new MemoryExporter();
+        $store->export(Signals::traces([SpanMother::withName('span')]));
+        static::assertCount(1, $store->spans());
+
+        $subscriber = new ProfilerStoreWorkerResetSubscriber($store);
+        $subscriber->onWorkerRunning(new WorkerRunningEvent(new Worker([], new MessageBus()), false));
+
+        static::assertSame([], $store->spans());
+    }
+
+    public function test_it_subscribes_to_the_worker_running_event_after_the_cycle_flush(): void
+    {
+        // WorkerReceiveCycleSubscriber flushes the cycle into the store at priority 1024; resetting must run
+        // afterwards (priority 512 < 1024) so the just-flushed signals are cleared and never accumulate.
+        static::assertSame(
+            [WorkerRunningEvent::class => ['onWorkerRunning', 512]],
+            ProfilerStoreWorkerResetSubscriber::getSubscribedEvents(),
+        );
+    }
+}

src/lib/telemetry/src/Flow/Telemetry/DSL/functions.php

@@ -341,11 +341,13 @@ function void_exporter(): VoidExporter
  *
  * Unified exporter that stores logs, metrics, and spans in memory for direct access.
  * Useful for testing and inspection without serialization.
+ *
+ * @param null|int $maxEntriesPerSignal maximum entries retained per signal type; null keeps everything
  */
 #[DocumentationDSL(module: Module::TELEMETRY, type: DSLType::HELPER)]
-function memory_exporter(): MemoryExporter
+function memory_exporter(?int $maxEntriesPerSignal = null): MemoryExporter
 {
-    return new MemoryExporter();
+    return new MemoryExporter($maxEntriesPerSignal);
 }
 
 /**

src/lib/telemetry/src/Flow/Telemetry/Provider/Memory/MemoryExporter.php

@@ -10,6 +10,11 @@
 use Flow\Telemetry\Signal\Signals;
 use Flow\Telemetry\Signal\SignalType;
 use Flow\Telemetry\Tracer\Span;
+use InvalidArgumentException;
+
+use function array_slice;
+use function count;
+use function sprintf;
 
 /**
  * Exporter that stores telemetry batches in memory for direct access.
@@ -34,12 +39,28 @@ final class MemoryExporter implements Exporter
      */
     private array $spans = [];
 
+    /**
+     * @param null|int $maxEntriesPerSignal maximum number of entries retained per signal type; once exceeded the
+     *                                      oldest entries are dropped. Null (default) keeps everything, which is
+     *                                      unbounded and unsafe in long-running processes that never reset().
+     */
+    public function __construct(
+        private readonly ?int $maxEntriesPerSignal = null,
+    ) {
+        if ($maxEntriesPerSignal !== null && $maxEntriesPerSignal < 1) {
+            throw new InvalidArgumentException(sprintf(
+                'MemoryExporter maxEntriesPerSignal must be a positive integer, got %d',
+                $maxEntriesPerSignal,
+            ));
+        }
+    }
+
     public function export(Signals $signal): bool
     {
         match ($signal->type) {
-            SignalType::LOGS => $this->logs = [...$this->logs, ...$signal->allLogs()],
-            SignalType::METRICS => $this->metrics = [...$this->metrics, ...$signal->allMetrics()],
-            SignalType::TRACES => $this->spans = [...$this->spans, ...$signal->allSpans()],
+            SignalType::LOGS => $this->logs = $this->cap([...$this->logs, ...$signal->allLogs()]),
+            SignalType::METRICS => $this->metrics = $this->cap([...$this->metrics, ...$signal->allMetrics()]),
+            SignalType::TRACES => $this->spans = $this->cap([...$this->spans, ...$signal->allSpans()]),
         };
 
         return true;
@@ -77,4 +98,20 @@ public function spans(): array
     {
         return $this->spans;
     }
+
+    /**
+     * @template T
+     *
+     * @param array<T> $entries
+     *
+     * @return array<T>
+     */
+    private function cap(array $entries): array
+    {
+        if ($this->maxEntriesPerSignal === null || count($entries) <= $this->maxEntriesPerSignal) {
+            return $entries;
+        }
+
+        return array_slice($entries, -$this->maxEntriesPerSignal);
+    }
 }

src/lib/telemetry/tests/Flow/Telemetry/Tests/Unit/Provider/Memory/MemoryExporterTest.php

@@ -14,12 +14,75 @@
 use Flow\Telemetry\Signal\Signals;
 use Flow\Telemetry\Tests\Mother\InstrumentationScopeMother;
 use Flow\Telemetry\Tests\Mother\LogEntryMother;
+use Flow\Telemetry\Tests\Mother\MetricMother;
 use Flow\Telemetry\Tests\Mother\ResourceMother;
 use Flow\Telemetry\Tests\Mother\SpanMother;
+use InvalidArgumentException;
 use PHPUnit\Framework\TestCase;
 
 final class MemoryExporterTest extends TestCase
 {
+    public function test_constructor_rejects_non_positive_max_entries(): void
+    {
+        $this->expectException(InvalidArgumentException::class);
+        $this->expectExceptionMessage('MemoryExporter maxEntriesPerSignal must be a positive integer, got 0');
+
+        new MemoryExporter(0);
+    }
+
+    public function test_export_caps_logs_to_max_entries_dropping_oldest(): void
+    {
+        $exporter = new MemoryExporter(2);
+        $entry1 = LogEntryMother::create('Log 1', Severity::INFO);
+        $entry2 = LogEntryMother::create('Log 2', Severity::WARN);
+        $entry3 = LogEntryMother::create('Log 3', Severity::ERROR);
+
+        $exporter->export(Signals::logs([$entry1, $entry2]));
+        $exporter->export(Signals::logs([$entry3]));
+
+        static::assertCount(2, $exporter->logs());
+        static::assertSame($entry2, $exporter->logs()[0]);
+        static::assertSame($entry3, $exporter->logs()[1]);
+    }
+
+    public function test_export_caps_metrics_to_max_entries_dropping_oldest(): void
+    {
+        $exporter = new MemoryExporter(1);
+        $first = MetricMother::counter('first', 1);
+        $second = MetricMother::counter('second', 2);
+
+        $exporter->export(Signals::metrics([$first]));
+        $exporter->export(Signals::metrics([$second]));
+
+        static::assertCount(1, $exporter->metrics());
+        static::assertSame($second, $exporter->metrics()[0]);
+    }
+
+    public function test_export_caps_spans_to_max_entries_dropping_oldest(): void
+    {
+        $exporter = new MemoryExporter(2);
+        $span1 = SpanMother::withName('span-1');
+        $span2 = SpanMother::withName('span-2');
+        $span3 = SpanMother::withName('span-3');
+
+        $exporter->export(Signals::traces([$span1, $span2, $span3]));
+
+        static::assertCount(2, $exporter->spans());
+        static::assertSame($span2, $exporter->spans()[0]);
+        static::assertSame($span3, $exporter->spans()[1]);
+    }
+
+    public function test_export_without_cap_keeps_every_entry(): void
+    {
+        $exporter = new MemoryExporter();
+
+        for ($i = 0; $i < 100; $i++) {
+            $exporter->export(Signals::traces([SpanMother::withName('span-' . $i)]));
+        }
+
+        static::assertCount(100, $exporter->spans());
+    }
+
     public function test_export_empty_logs_does_not_modify_state(): void
     {
         $exporter = new MemoryExporter();