feat: vizualize the message flow graph, and prove tracing, OpenTelemetry, flow tracking and the graph against one catalogue of flows

Monorepo/CrossModuleTests/Fixture/Flow/CrossModuleFlowScenarioCatalogue.php

@@ -0,0 +1,35 @@
+<?php
+
+declare(strict_types=1);
+
+namespace Monorepo\CrossModuleTests\Fixture\Flow;
+
+use Ecotone\Test\Flow\FlowScenario;
+
+/**
+ * licence Apache-2.0
+ */
+final class CrossModuleFlowScenarioCatalogue
+{
+    /**
+     * @return iterable<string, array{FlowScenario}>
+     */
+    public static function dataProvider(): iterable
+    {
+        foreach (self::all() as $scenario) {
+            yield $scenario->name => [$scenario];
+        }
+    }
+
+    /**
+     * @return list<FlowScenario>
+     */
+    public static function all(): array
+    {
+        return [
+            ...DbalFlowScenarios::all(),
+            ...EventSourcingFlowScenarios::all(),
+            ...MultiTenantFlowScenarios::all(),
+        ];
+    }
+}

Monorepo/CrossModuleTests/Fixture/Flow/EventSourcingFlowScenarios.php

@@ -0,0 +1,126 @@
+<?php
+
+declare(strict_types=1);
+
+namespace Monorepo\CrossModuleTests\Fixture\Flow;
+
+use Ecotone\Api\Attribute\EventHandler;
+use Ecotone\Api\Dbal\ExtensionObject\DbalBackedMessageChannelBuilder;
+use Ecotone\Api\ExtensionObject\ExecutionPollingMetadata;
+use Ecotone\Api\Lite\Test\FlowTestSupport;
+use Ecotone\Api\Projecting\FromAggregateStream;
+use Ecotone\Api\Projecting\Partitioned;
+use Ecotone\Api\Projecting\Projection;
+use Ecotone\Api\Projecting\ProjectionRebuild;
+use Ecotone\Api\Tracing\SpanAttribute;
+use Ecotone\Api\Tracing\SpanKind;
+use Ecotone\Test\Flow\ExpectedSpan;
+use Ecotone\Test\Flow\FlowScenario;
+use Ecotone\Test\Flow\FlowStage;
+use Ecotone\Test\Flow\Fixture\FlowOrder;
+use Ecotone\Test\Flow\Fixture\FlowOrderPlaced;
+
+/**
+ * licence Apache-2.0
+ */
+final class EventSourcingFlowScenarios
+{
+    /**
+     * @return list<FlowScenario>
+     */
+    public static function all(): array
+    {
+        return [
+            self::projectionOnTheDatabaseEventStoreRebuilt(),
+            self::partitionedProjectionRebuiltThroughADatabaseChannel(),
+            self::projectionDeletedInitializedAndBackfilledFromTheDatabaseEventStore(),
+        ];
+    }
+
+    private static function projectionOnTheDatabaseEventStoreRebuilt(): FlowScenario
+    {
+        $projection = new #[Projection('order_list'), FromAggregateStream(FlowOrder::class)] class () {
+            #[EventHandler(endpointId: 'order_list.placed')]
+            public function whenPlaced(FlowOrderPlaced $event): void
+            {
+            }
+        };
+
+        return new FlowScenario('projection on the database event store rebuilt from the console', [FlowOrder::class, FlowOrderPlaced::class, $projection::class], [$projection], [], [
+            self::orderPlaced(),
+            FlowStage::when(fn (FlowTestSupport $ecotone) => $ecotone->runConsoleCommand('ecotone:projection:rebuild', ['name' => 'order_list']))
+                ->expect(self::operation('rebuild')->containing(self::projectedOrder())),
+        ]);
+    }
+
+    private static function partitionedProjectionRebuiltThroughADatabaseChannel(): FlowScenario
+    {
+        $projection = new #[Projection('order_list'), Partitioned, FromAggregateStream(FlowOrder::class), ProjectionRebuild(asyncChannelName: 'order_list_rebuild')] class () {
+            #[EventHandler(endpointId: 'order_list.placed')]
+            public function whenPlaced(FlowOrderPlaced $event): void
+            {
+            }
+        };
+
+        return new FlowScenario('partitioned projection rebuilt through a database channel', [FlowOrder::class, FlowOrderPlaced::class, $projection::class], [$projection], [DbalBackedMessageChannelBuilder::create('order_list_rebuild')], [
+            self::orderPlaced(),
+            FlowStage::when(fn (FlowTestSupport $ecotone) => $ecotone->runConsoleCommand('ecotone:projection:rebuild', ['name' => 'order_list']))
+                ->expect(self::operation('rebuild')->containing(ExpectedSpan::named('send order_list_rebuild')->ofKind(SpanKind::Producer)))
+                ->expectNoSpanNamed('receive order_list_rebuild')
+                ->leavingWorkPending(),
+            FlowStage::when(fn (FlowTestSupport $ecotone) => $ecotone->run('order_list_rebuild', ExecutionPollingMetadata::createWithTestingSetup()))
+                ->expect(
+                    self::operation('rebuild')->containing(
+                        ExpectedSpan::named('send order_list_rebuild')->containing(
+                            ExpectedSpan::named('receive order_list_rebuild')->ofKind(SpanKind::Consumer)->containing(self::projectedOrder()),
+                        ),
+                    ),
+                ),
+        ]);
+    }
+
+    private static function projectionDeletedInitializedAndBackfilledFromTheDatabaseEventStore(): FlowScenario
+    {
+        $projection = new #[Projection('order_list'), FromAggregateStream(FlowOrder::class)] class () {
+            #[EventHandler(endpointId: 'order_list.placed')]
+            public function whenPlaced(FlowOrderPlaced $event): void
+            {
+            }
+        };
+
+        return new FlowScenario('projection deleted, initialized and backfilled from the database event store', [FlowOrder::class, FlowOrderPlaced::class, $projection::class], [$projection], [], [
+            self::orderPlaced(),
+            FlowStage::when(function (FlowTestSupport $ecotone): void {
+                $ecotone->deleteProjection('order_list');
+                $ecotone->initializeProjection('order_list');
+                $ecotone->runConsoleCommand('ecotone:projection:backfill', ['name' => 'order_list']);
+            })
+                ->expect(self::operation('delete'), self::operation('initialize'), self::operation('backfill')->containing(self::projectedOrder())),
+        ]);
+    }
+
+    private static function orderPlaced(): FlowStage
+    {
+        return FlowStage::when(fn (FlowTestSupport $ecotone, array $metadata) => $ecotone->sendCommandWithRouting('flow_order.place', ['orderId' => 'order-1'], metadata: $metadata))
+            ->expect(
+                ExpectedSpan::named('send command')->containing(
+                    ExpectedSpan::named('process FlowOrder::place')->withAttributes([SpanAttribute::ENDPOINT_ID => 'flow_order.placing']),
+                    ExpectedSpan::named('send event')->containing(
+                        ExpectedSpan::named('route order_list')->containing(
+                            ExpectedSpan::named('process order_list')->containing(self::projectedOrder()),
+                        ),
+                    ),
+                ),
+            );
+    }
+
+    private static function operation(string $operation): ExpectedSpan
+    {
+        return ExpectedSpan::named($operation . ' order_list')->asRoot()->ofKind(SpanKind::Server)->withAttributes([SpanAttribute::PROJECTION_NAME => 'order_list', SpanAttribute::PROJECTION_OPERATION => $operation]);
+    }
+
+    private static function projectedOrder(): ExpectedSpan
+    {
+        return ExpectedSpan::named('process class@anonymous::whenPlaced')->withAttributes([SpanAttribute::ENDPOINT_ID => 'order_list.placed']);
+    }
+}

Monorepo/CrossModuleTests/Fixture/Flow/MultiTenantFlowScenarios.php

@@ -0,0 +1,74 @@
+<?php
+
+declare(strict_types=1);
+
+namespace Monorepo\CrossModuleTests\Fixture\Flow;
+
+use Ecotone\Api\Attribute\Asynchronous;
+use Ecotone\Api\Attribute\CommandHandler;
+use Ecotone\Api\Dbal\DbalConnectionFactory;
+use Ecotone\Api\Dbal\ExtensionObject\DbalBackedMessageChannelBuilder;
+use Ecotone\Api\Dbal\ExtensionObject\MultiTenantConfiguration;
+use Ecotone\Api\ExtensionObject\ExecutionPollingMetadata;
+use Ecotone\Api\Lite\Test\FlowTestSupport;
+use Ecotone\Api\Tracing\SpanAttribute;
+use Ecotone\Api\Tracing\SpanKind;
+use Ecotone\Test\Flow\ExpectedSpan;
+use Ecotone\Test\Flow\FlowScenario;
+use Ecotone\Test\Flow\FlowStage;
+use Test\Ecotone\EventSourcing\SecondaryTestDatabaseConnectionFactory;
+
+/**
+ * licence Apache-2.0
+ */
+final class MultiTenantFlowScenarios
+{
+    /**
+     * @return list<FlowScenario>
+     */
+    public static function all(): array
+    {
+        return [
+            self::commandsOfTwoTenantsConsumedFromTheirOwnDatabases(),
+        ];
+    }
+
+    private static function commandsOfTwoTenantsConsumedFromTheirOwnDatabases(): FlowScenario
+    {
+        $orders = new class () {
+            #[Asynchronous('orders')]
+            #[CommandHandler('order.place', endpointId: 'order.placing')]
+            public function place(string $orderId): void
+            {
+            }
+        };
+        $sentFor = fn (string $tenant) => ExpectedSpan::named('send command')->asRoot()->withAttributes(['tenant' => $tenant])->containing(
+            ExpectedSpan::named('send orders')->ofKind(SpanKind::Producer),
+        );
+        $handledFor = fn (string $tenant) => ExpectedSpan::named('send command')->asRoot()->withAttributes(['tenant' => $tenant])->containing(
+            ExpectedSpan::named('send orders')->containing(
+                ExpectedSpan::named('receive orders')->ofKind(SpanKind::Consumer)->containing(
+                    ExpectedSpan::named('process class@anonymous::place')->withAttributes([SpanAttribute::ENDPOINT_ID => 'order.placing']),
+                ),
+            ),
+        );
+
+        return new FlowScenario('commands of two tenants consumed from their own databases', [$orders::class], [
+            $orders,
+            'tenant_a_connection' => new DbalConnectionFactory(getenv('DATABASE_DSN') ?: 'pgsql://ecotone:secret@127.0.0.1:5432/ecotone'),
+            'tenant_b_connection' => new DbalConnectionFactory(SecondaryTestDatabaseConnectionFactory::resolveDsn()),
+        ], [
+            MultiTenantConfiguration::create('tenant', ['tenant_a' => 'tenant_a_connection', 'tenant_b' => 'tenant_b_connection']),
+            DbalBackedMessageChannelBuilder::create('orders'),
+        ], [
+            FlowStage::when(fn (FlowTestSupport $ecotone, array $metadata) => $ecotone->sendCommandWithRouting('order.place', '1', metadata: [...$metadata, 'tenant' => 'tenant_a']))
+                ->expect($sentFor('tenant_a'))
+                ->leavingWorkPending(),
+            FlowStage::when(fn (FlowTestSupport $ecotone, array $metadata) => $ecotone->sendCommandWithRouting('order.place', '2', metadata: [...$metadata, 'tenant' => 'tenant_b']))
+                ->expect($sentFor('tenant_b'))
+                ->leavingWorkPending(),
+            FlowStage::when(fn (FlowTestSupport $ecotone) => $ecotone->run('orders', ExecutionPollingMetadata::createWithTestingSetup()))
+                ->expect($handledFor('tenant_a'), $handledFor('tenant_b')),
+        ]);
+    }
+}

Monorepo/CrossModuleTests/Tests/Tracing/CrossModuleFlowConformanceTest.php

@@ -0,0 +1,74 @@
+<?php
+
+declare(strict_types=1);
+
+namespace Monorepo\CrossModuleTests\Tests\Tracing;
+
+use Ecotone\Api\Dbal\DbalConnectionFactory;
+use Ecotone\Api\Dbal\Tracing\FlowTrackingConfiguration;
+use Ecotone\Api\Dbal\Tracing\TracingGateway;
+use Ecotone\Api\ExtensionObject\ModulePackageList;
+use Ecotone\Api\Lite\EcotoneLite;
+use Ecotone\Api\Tracing\InMemorySpanRecorder;
+use Ecotone\Api\Tracing\TracingConfiguration;
+use Ecotone\Dbal\Test\StoredFlowVerifier;
+use Ecotone\OpenTelemetry\Test\ExportedSpans;
+use Ecotone\Test\Flow\FlowScenario;
+use Ecotone\Test\Flow\MessageFlowGraphVerifier;
+use Ecotone\Test\LicenceTesting;
+use Monorepo\CrossModuleTests\Fixture\Flow\CrossModuleFlowScenarioCatalogue;
+use OpenTelemetry\API\Trace\TracerProviderInterface;
+use OpenTelemetry\SDK\Trace\SpanExporter\InMemoryExporter;
+use OpenTelemetry\SDK\Trace\SpanProcessor\SimpleSpanProcessor;
+use OpenTelemetry\SDK\Trace\TracerProvider;
+use PHPUnit\Framework\Attributes\DataProviderExternal;
+use PHPUnit\Framework\TestCase;
+use Test\Ecotone\Dbal\DbalMessagingTestCase;
+use Test\Ecotone\EventSourcing\EventSourcingMessagingTestCase;
+use Test\Ecotone\EventSourcing\SecondaryTestDatabaseConnectionFactory;
+
+/**
+ * licence Apache-2.0
+ * @internal
+ */
+final class CrossModuleFlowConformanceTest extends TestCase
+{
+    protected function setUp(): void
+    {
+        EventSourcingMessagingTestCase::clearDataTables(DbalMessagingTestCase::prepareConnection()->createContext()->getDbalConnection());
+        EventSourcingMessagingTestCase::clearDataTables((new DbalConnectionFactory(SecondaryTestDatabaseConnectionFactory::resolveDsn()))->createContext()->getDbalConnection());
+    }
+
+    #[DataProviderExternal(CrossModuleFlowScenarioCatalogue::class, 'dataProvider')]
+    public function test_every_cross_module_flow_is_traced_exported_stored_and_drawn_alike(FlowScenario $scenario): void
+    {
+        $recorder = new InMemorySpanRecorder();
+        $exporter = new InMemoryExporter();
+        $ecotone = EcotoneLite::bootstrapFlowTesting(
+            $scenario->classesToResolve,
+            [
+                ...$scenario->services,
+                InMemorySpanRecorder::class => $recorder,
+                TracerProviderInterface::class => new TracerProvider(new SimpleSpanProcessor($exporter)),
+                DbalConnectionFactory::class => new DbalConnectionFactory(getenv('DATABASE_DSN') ?: 'pgsql://ecotone:secret@127.0.0.1:5432/ecotone'),
+            ],
+            [
+                FlowTrackingConfiguration::createWithDefaults(),
+                TracingConfiguration::createWithDefaults()->withSpanRecorder(InMemorySpanRecorder::class),
+                ...$scenario->extensionObjects,
+            ],
+            enabledModules: [ModulePackageList::DBAL_PACKAGE, ModulePackageList::EVENT_SOURCING_PACKAGE, ModulePackageList::JMS_CONVERTER_PACKAGE, ModulePackageList::TRACING_PACKAGE],
+            licenceKey: LicenceTesting::VALID_LICENCE,
+        )->setupDatabaseTables();
+        $tracing = $ecotone->getGateway(TracingGateway::class);
+
+        foreach (array_keys($scenario->stages) as $index) {
+            self::assertNull($scenario->runStage($index, $ecotone));
+
+            self::assertNull($scenario->mismatchAfterStage($index, $recorder->recordedSpans()));
+            self::assertNull($scenario->mismatchAfterStage($index, ExportedSpans::of($exporter)));
+            self::assertNull(StoredFlowVerifier::mismatchAfterStage($scenario, $index, $tracing, $recorder->recordedSpans()));
+        }
+        self::assertNull(MessageFlowGraphVerifier::mismatchIn($ecotone->getMessageFlowGraph(), $recorder->recordedSpans()));
+    }
+}

packages/Dbal/Api/ExtensionObject/DbalDeadLetterBuilder.php

@@ -5,6 +5,7 @@
 use Ecotone\Api\Messaging\MessageHeaders;
 use Ecotone\Dbal\Recoverability\DeadLetterHandler;
 use Ecotone\Dbal\Recoverability\DeadLetterStore;
+use Ecotone\Dbal\Tracing\FlowTrackingScope;
 use Ecotone\Messaging\Config\Container\Definition;
 use Ecotone\Messaging\Config\Container\InterfaceToCallReference;
 use Ecotone\Messaging\Config\Container\MessagingContainerBuilder;
@@ -158,6 +159,7 @@ public function compile(MessagingContainerBuilder $builder): Definition
             $builder->register($deadLetterHandlerReference, new Definition(DeadLetterHandler::class, [
                 Reference::to(self::storeReferenceFor($this->referenceName)),
                 Reference::to(MessageBuilder::class),
+                Reference::to(FlowTrackingScope::class),
             ]));
         }
 

packages/Dbal/src/BatchForwarding/DbalBatchForwardingModule.php

@@ -13,6 +13,7 @@
 use Ecotone\Api\ExtensionObject\ModulePackageList;
 use Ecotone\Api\Gateway\EcotoneClockInterface;
 use Ecotone\Dbal\DbalReconnectableConnectionFactory;
+use Ecotone\Dbal\Tracing\FlowTrackingScope;
 use Ecotone\Enqueue\CachedConnectionFactory;
 use Ecotone\Messaging\Config\Annotation\AnnotationModule;
 use Ecotone\Messaging\Config\Annotation\ModuleConfiguration\ExtensionObjectResolver;
@@ -134,6 +135,7 @@ private function createOutboxPublisherDefinition(DbalBackedMessageChannelBuilder
             $outboxForwardingChannel->getMaxForwardingBatchSize(),
             $outboxForwardingChannel->getFinalFailureStrategy() ?? $inboundChannelAdapter->getFinalFailureStrategy(),
             new Reference(MessageBuilder::class),
+            new Reference(FlowTrackingScope::class),
         ]);
     }
 }

packages/Dbal/src/BatchForwarding/DbalOutboxPublisher.php

@@ -14,6 +14,7 @@
 use Ecotone\Api\Messaging\MessageHeaders;
 use Ecotone\Dbal\Connection\DbalContext;
 use Ecotone\Dbal\Connection\DbalType;
+use Ecotone\Dbal\Tracing\FlowTrackingScope;
 use Ecotone\Enqueue\CachedConnectionFactory;
 use Ecotone\Messaging\Channel\BatchSupportingMessageChannel;
 use Ecotone\Messaging\Channel\MessageChannelInterceptorAdapter;
@@ -59,10 +60,16 @@ public function __construct(
         private int $maxBatchSize,
         private FinalFailureStrategy $finalFailureStrategy,
         private MessageBuilder $messageBuilder,
+        private FlowTrackingScope $flowTracking,
     ) {
     }
 
     public function publishBatch(): int
+    {
+        return $this->flowTracking->publishingStepsAfterCommit(fn (): int => $this->forwardClaimedBatch());
+    }
+
+    private function forwardClaimedBatch(): int
     {
         /** @var DbalContext $context */
         $context = $this->connectionFactory->createContext();
@@ -83,9 +90,11 @@ public function publishBatch(): int
 
             $deliveredRowIds = [];
             $releasedRowIds = [];
-            foreach ($this->groupByTargetChannel($drainedRecords, $releasedRowIds) as $groupTargetChannelName => $groupedRecords) {
-                $this->forwardGroup((string) $groupTargetChannelName, $groupedRecords, $deliveredRowIds, $releasedRowIds);
-            }
+            $this->flowTracking->storingSpansOf(function () use ($drainedRecords, &$deliveredRowIds, &$releasedRowIds): void {
+                foreach ($this->groupByTargetChannel($drainedRecords, $releasedRowIds) as $groupTargetChannelName => $groupedRecords) {
+                    $this->forwardGroup((string) $groupTargetChannelName, $groupedRecords, $deliveredRowIds, $releasedRowIds);
+                }
+            });
 
             $this->deleteRows($context, $connection, $deliveredRowIds);
             $this->releaseRows($context, $connection, $releasedRowIds);