From e903bf4deb03a39b7184ef46eb490ab1da10383b Mon Sep 17 00:00:00 2001 From: Dariusz Gafka Date: Tue, 20 Oct 2026 17:42:00 +0200 Subject: [PATCH] test(flow): poll multi-message catalogue stages until the queue is empty, not for 100ms ExecutionPollingMetadata::createWithTestingSetup() limits a run to 100ms of wall-clock time, and the consumer stops after any poll that ends past it. Five catalogue stages need more than one poll in a single run: the round-robin dynamic channel (orders_a, then orders_b), two commands on one queue, a handler feeding its own channel, a retry before the dead letter, and an error channel send that fails on the redelivery. On a slow runner the first message alone took over 100ms, the run stopped, and the second message was never received. These stages now run with createWithFinishWhenNoMessages(), which stops only once the queue is empty. --- packages/Ecotone/src/Test/Flow/AsynchronousFlowScenarios.php | 2 +- packages/Ecotone/src/Test/Flow/ChannelFailureFlowScenarios.php | 2 +- packages/Ecotone/src/Test/Flow/ChannelFlowScenarios.php | 2 +- packages/Ecotone/src/Test/Flow/RecoverabilityFlowScenarios.php | 2 +- packages/Ecotone/src/Test/Flow/TraceFlowScenarios.php | 2 +- 5 files changed, 5 insertions(+), 5 deletions(-) diff --git a/packages/Ecotone/src/Test/Flow/AsynchronousFlowScenarios.php b/packages/Ecotone/src/Test/Flow/AsynchronousFlowScenarios.php index ae6414e82..d7e42fdab 100644 --- a/packages/Ecotone/src/Test/Flow/AsynchronousFlowScenarios.php +++ b/packages/Ecotone/src/Test/Flow/AsynchronousFlowScenarios.php @@ -151,7 +151,7 @@ public function notify(string $userId): void ) ->expectNoSpanNamed('process class@anonymous::register') ->withNoOtherSpans(), - FlowStage::when(fn (FlowTestSupport $ecotone) => $ecotone->run('async_channel', ExecutionPollingMetadata::createWithTestingSetup(handledMessageLimit: 2))) + FlowStage::when(fn (FlowTestSupport $ecotone) => $ecotone->run('async_channel', ExecutionPollingMetadata::createWithFinishWhenNoMessages())) ->expect( ExpectedSpan::named('receive async_channel')->ofKind(SpanKind::Consumer)->containing( ExpectedSpan::named('process class@anonymous::register')->withAttributes([SpanAttribute::ENDPOINT_ID => 'user.registration'])->containing( diff --git a/packages/Ecotone/src/Test/Flow/ChannelFailureFlowScenarios.php b/packages/Ecotone/src/Test/Flow/ChannelFailureFlowScenarios.php index 80eb3a300..402fb9d7e 100644 --- a/packages/Ecotone/src/Test/Flow/ChannelFailureFlowScenarios.php +++ b/packages/Ecotone/src/Test/Flow/ChannelFailureFlowScenarios.php @@ -326,7 +326,7 @@ public function register(string $userId): void ServiceConfiguration::createWithDefaults()->withDefaultErrorChannel('error_channel'), ], [ self::registrationSentToTheQueue(), - FlowStage::when(fn (FlowTestSupport $ecotone) => $ecotone->run('users', ExecutionPollingMetadata::createWithTestingSetup(stopOnError: false))) + FlowStage::when(fn (FlowTestSupport $ecotone) => $ecotone->run('users', ExecutionPollingMetadata::createWithFinishWhenNoMessages(stopOnError: false))) ->failingWith('Exception on send') ->expect( ExpectedSpan::named('receive users')->ofKind(SpanKind::Consumer)->withStatus(SpanStatus::Error)->containing( diff --git a/packages/Ecotone/src/Test/Flow/ChannelFlowScenarios.php b/packages/Ecotone/src/Test/Flow/ChannelFlowScenarios.php index 58c566ca8..b45c9792c 100644 --- a/packages/Ecotone/src/Test/Flow/ChannelFlowScenarios.php +++ b/packages/Ecotone/src/Test/Flow/ChannelFlowScenarios.php @@ -125,7 +125,7 @@ public function place(string $orderId): void ) ->withNoOtherSpans() ->leavingWorkPending(), - FlowStage::when(fn (FlowTestSupport $ecotone) => $ecotone->run('orders', ExecutionPollingMetadata::createWithTestingSetup(handledMessageLimit: 2))) + FlowStage::when(fn (FlowTestSupport $ecotone) => $ecotone->run('orders', ExecutionPollingMetadata::createWithFinishWhenNoMessages())) ->expect( ExpectedSpan::named('send orders_a')->containing(ExpectedSpan::named('receive orders')->withAttributes(['flowId' => '1'])->containing(ExpectedSpan::named('process class@anonymous::place'))), ExpectedSpan::named('send orders_b')->containing(ExpectedSpan::named('receive orders')->withAttributes(['flowId' => '2'])->containing(ExpectedSpan::named('process class@anonymous::place'))), diff --git a/packages/Ecotone/src/Test/Flow/RecoverabilityFlowScenarios.php b/packages/Ecotone/src/Test/Flow/RecoverabilityFlowScenarios.php index 93596cfce..90d28bf7a 100644 --- a/packages/Ecotone/src/Test/Flow/RecoverabilityFlowScenarios.php +++ b/packages/Ecotone/src/Test/Flow/RecoverabilityFlowScenarios.php @@ -81,7 +81,7 @@ private static function failureRetriedWithDelayThenSentToDeadLetter(): FlowScena ->expect(self::queued()) ->withNoOtherSpans() ->leavingWorkPending(), - FlowStage::when(fn (FlowTestSupport $ecotone) => $ecotone->run('orders', ExecutionPollingMetadata::createWithTestingSetup(stopOnError: false))) + FlowStage::when(fn (FlowTestSupport $ecotone) => $ecotone->run('orders', ExecutionPollingMetadata::createWithFinishWhenNoMessages(stopOnError: false))) ->expect( ExpectedSpan::named('receive orders')->withStatus(SpanStatus::Error)->containing( ExpectedSpan::named('process class@anonymous::place')->withStatus(SpanStatus::Error, 'Order cannot be placed'), diff --git a/packages/Ecotone/src/Test/Flow/TraceFlowScenarios.php b/packages/Ecotone/src/Test/Flow/TraceFlowScenarios.php index 6edb0deb3..ba74a7963 100644 --- a/packages/Ecotone/src/Test/Flow/TraceFlowScenarios.php +++ b/packages/Ecotone/src/Test/Flow/TraceFlowScenarios.php @@ -85,7 +85,7 @@ private static function twoAsynchronousCommandsKeepTheirOwnTraces(): FlowScenari return new FlowScenario('two asynchronous commands consumed in one run keep their own traces', [$users::class], [$users], [SimpleMessageChannelBuilder::createQueueChannel('users')], [ $send('1'), $send('2'), - FlowStage::when(fn (FlowTestSupport $ecotone) => $ecotone->run('users', ExecutionPollingMetadata::createWithTestingSetup(handledMessageLimit: 2))) + FlowStage::when(fn (FlowTestSupport $ecotone) => $ecotone->run('users', ExecutionPollingMetadata::createWithFinishWhenNoMessages())) ->expect($consumed('1'), $consumed('2')), ]); }