feat: disable global DbalTransaction and pause Collector during projection batch execution

[dgafka]

Why is this change proposed?

When ProjectionV2 runs asynchronously or during backfill, the global DbalTransactionInterceptor wraps the entire execution in one big transaction — overriding the projection's own per-batch transaction management. This means all batches commit or rollback together instead of independently. Additionally, the CollectorSenderInterceptor collects messages sent during flush() and releases them only after all batches complete, outside the per-batch transaction scope.

Description of Changes

• Disable global DbalTransaction for projection execute and backfill handlers — Added WithoutDbalTransaction endpoint annotation to both the execute() and BackfillExecutorHandler message handlers in ProjectingModule. In async mode, the projection's own executeSingleBatch() now manages real per-batch transactions (since isTransactionActive() returns false). In synchronous mode (within CommandBus), the existing transaction is still detected and reused via NoOpTransaction.

• Pause Collector during ProjectionFlush — Added a static pause()/resume() mechanism to CollectorStorage and a new CollectorPauseInterceptor (around interceptor on ProjectionFlush pointcut). When a projection batch executes, collector is paused so messages sent during flush() go directly to the channel within the batch transaction, rather than being collected and released later.

• Updated MessageCollectorChannelInterceptor — preSend() now checks CollectorStorage::isPaused() and lets messages through when paused.

• Tests — Unit tests for CollectorStorage pause/resume mechanism, integration test verifying messages sent during projection flush bypass the collector.

Flow: Async Projection with Collector (after fix)

sequenceDiagram
    participant AsyncConsumer
    participant ProjectingManager
    participant executeSingleBatch
    participant Flush
    participant Channel

    AsyncConsumer->>ProjectingManager: execute() [no global DbalTransaction]
    loop Per batch
        ProjectingManager->>executeSingleBatch: [ProjectionFlush pointcut]
        Note over executeSingleBatch: beginTransaction() → real DB transaction
        Note over executeSingleBatch: CollectorStorage::pause()
        executeSingleBatch->>Flush: flush()
        Flush->>Channel: send message (bypasses collector)
        Note over executeSingleBatch: commit transaction
        Note over executeSingleBatch: CollectorStorage::resume()
    end

Loading

Pull Request Contribution Terms

• I have read and agree to the contribution terms outlined in CONTRIBUTING.