fix: resolve event stream race conditions in EventConsumer and SSE tr… (#782)

…ansport

Fixed multiple race conditions causing intermittent test failures in
streaming and subscription scenarios. These fixes reduced the failure
rate from ~33% to 0% across 1,500+ test iterations in CI.

Removed executor.execute() wrapper in REST/JSONRPC routes that delayed
subscription by 100-600ms, causing events to be lost when EventConsumer
started emitting before subscriber was ready.

Changed onCancelTask to use consumeAndBreakOnInterrupt() instead of
consumeAll(). Removed unused ResultAggregator.consumeAll() method since
cancel was its only caller.

Moved EventConsumer polling loop to executor thread to prevent blocking
caller, ensuring subscription happens immediately without delay.

Fixed race in onSubscribeToTask where initial task snapshot was enqueued
but EventConsumer polling hadn't started yet. Added insertingProcessor()
utility to AsyncUtils that prepends initial items synchronously to
reactive streams using mutiny-zero ZeroPublisher, ensuring subscriber
receives initial task snapshot immediately on subscription.

Updated transport layer unit tests (JSONRPCHandlerTest, GrpcHandlerTest)
to expect initial task snapshot per A2A Protocol Specification 3.1.6.

Fixed insertingProcessor() to respect reactive streams semantics by
sending inserted items in the source's onSubscribe() callback after
subscription is established, rather than immediately in the
ZeroPublisher creation lambda.

Fixed two test race conditions where tests checked for received events
immediately after subscription was established (server-side metric),
without waiting for consumer callbacks to actually process events:
- testSubscribeToTaskWithInterruptedStateKeepsStreamOpen: Added
initialTaskLatch to wait for initial TaskEvent reception
- testNonBlockingWithMultipleMessages: Added streamConsumerReadyLatch to
wait for streaming consumer to start receiving events

Enhanced awaitingFinalEvent tracking with timeout guards (max 3s wait)
to prevent infinite waiting if final event never arrives due to
distribution delays in replicated scenarios.

Increased sleep delay from 50ms to 150ms to account for CI environment
latency and ensure buffered events flush before stream ends.

Improved pollTimeoutsWhileAwaitingFinal reset logic to only reset when
not awaiting final event. Calculated timeout constant from base timeout
value for better maintainability.

Tests fixed:
- testNonBlockingWithMultipleMessages
- testCancelTaskSuccess
- testSubscribeToTaskWithInterruptedStateKeepsStreamOpen

Author: kabir

reference/jsonrpc/src/main/java/io/a2a/server/apps/quarkus/A2AServerRoutes.java

@@ -299,15 +299,15 @@ public void invokeJSONRPCHandler(@Body String body, RoutingContext rc) {
                         .putHeader(CONTENT_TYPE, APPLICATION_JSON)
                         .end(serializeResponse(error));
             } else if (streaming) {
-                final Multi<? extends A2AResponse<?>> finalStreamingResponse = streamingResponse;
-                executor.execute(() -> {
-                    // Convert Multi<A2AResponse> to Multi<String> with SSE formatting
-                    AtomicLong eventIdCounter = new AtomicLong(0);
-                    Multi<String> sseEvents = finalStreamingResponse
-                            .map(response -> SseFormatter.formatResponseAsSSE(response, eventIdCounter.getAndIncrement()));
-                    // Write SSE-formatted strings to HTTP response
-                    MultiSseSupport.writeSseStrings(sseEvents, rc, context);
-                });
+                // Convert Multi<A2AResponse> to Multi<String> with SSE formatting
+                // CRITICAL: Subscribe synchronously to avoid race condition where EventConsumer
+                // starts emitting events before MultiSseSupport subscribes. The executor.execute()
+                // wrapper caused 100-600ms delays before subscription, causing events to be lost.
+                AtomicLong eventIdCounter = new AtomicLong(0);
+                Multi<String> sseEvents = streamingResponse
+                        .map(response -> SseFormatter.formatResponseAsSSE(response, eventIdCounter.getAndIncrement()));
+                // Write SSE-formatted strings to HTTP response
+                MultiSseSupport.writeSseStrings(sseEvents, rc, context);
 
             } else {
                 rc.response()
@@ -783,7 +783,17 @@ public void onNext(String sseEvent) {
                         if (headers.get(CONTENT_TYPE) == null) {
                             headers.set(CONTENT_TYPE, SERVER_SENT_EVENTS);
                         }
+                        // Additional SSE headers to prevent buffering
+                        headers.set("Cache-Control", "no-cache");
+                        headers.set("X-Accel-Buffering", "no");  // Disable nginx buffering
                         response.setChunked(true);
+
+                        // CRITICAL: Disable write queue max size to prevent buffering
+                        // Vert.x buffers writes by default - we need immediate flushing for SSE
+                        response.setWriteQueueMaxSize(1);
+
+                        // Send initial SSE comment to kickstart the stream
+                        response.write(": SSE stream started\n\n");
                     }
 
                     // Write SSE-formatted string to response

reference/rest/src/main/java/io/a2a/server/rest/quarkus/A2AServerRoutes.java

@@ -221,15 +221,15 @@ public void sendMessageStreaming(@Body String body, RoutingContext rc) {
             if (error != null) {
                 sendResponse(rc, error);
             } else if (streamingResponse != null) {
-                final HTTPRestStreamingResponse finalStreamingResponse = streamingResponse;
-                executor.execute(() -> {
-                    // Convert Flow.Publisher<String> (JSON) to Multi<String> (SSE-formatted)
-                    AtomicLong eventIdCounter = new AtomicLong(0);
-                    Multi<String> sseEvents = Multi.createFrom().publisher(finalStreamingResponse.getPublisher())
-                            .map(json -> SseFormatter.formatJsonAsSSE(json, eventIdCounter.getAndIncrement()));
-                    // Write SSE-formatted strings to HTTP response
-                    MultiSseSupport.writeSseStrings(sseEvents, rc, context);
-                });
+                // Convert Flow.Publisher<String> (JSON) to Multi<String> (SSE-formatted)
+                // CRITICAL: Subscribe synchronously to avoid race condition where EventConsumer
+                // starts emitting events before MultiSseSupport subscribes. The executor.execute()
+                // wrapper caused 100-600ms delays before subscription, causing events to be lost.
+                AtomicLong eventIdCounter = new AtomicLong(0);
+                Multi<String> sseEvents = Multi.createFrom().publisher(streamingResponse.getPublisher())
+                        .map(json -> SseFormatter.formatJsonAsSSE(json, eventIdCounter.getAndIncrement()));
+                // Write SSE-formatted strings to HTTP response
+                MultiSseSupport.writeSseStrings(sseEvents, rc, context);
             }
         }
     }
@@ -431,15 +431,15 @@ public void subscribeToTask(RoutingContext rc) {
             if (error != null) {
                 sendResponse(rc, error);
             } else if (streamingResponse != null) {
-                final HTTPRestStreamingResponse finalStreamingResponse = streamingResponse;
-                executor.execute(() -> {
-                    // Convert Flow.Publisher<String> (JSON) to Multi<String> (SSE-formatted)
-                    AtomicLong eventIdCounter = new AtomicLong(0);
-                    Multi<String> sseEvents = Multi.createFrom().publisher(finalStreamingResponse.getPublisher())
-                            .map(json -> SseFormatter.formatJsonAsSSE(json, eventIdCounter.getAndIncrement()));
-                    // Write SSE-formatted strings to HTTP response
-                    MultiSseSupport.writeSseStrings(sseEvents, rc, context);
-                });
+                // Convert Flow.Publisher<String> (JSON) to Multi<String> (SSE-formatted)
+                // CRITICAL: Subscribe synchronously to avoid race condition where EventConsumer
+                // starts emitting events before MultiSseSupport subscribes. The executor.execute()
+                // wrapper caused 100-600ms delays before subscription, causing events to be lost.
+                AtomicLong eventIdCounter = new AtomicLong(0);
+                Multi<String> sseEvents = Multi.createFrom().publisher(streamingResponse.getPublisher())
+                        .map(json -> SseFormatter.formatJsonAsSSE(json, eventIdCounter.getAndIncrement()));
+                // Write SSE-formatted strings to HTTP response
+                MultiSseSupport.writeSseStrings(sseEvents, rc, context);
             }
         }
     }