Skip to content

Commit dabca6c

Browse files
authored
fix(server): recover failed terminal enqueues and fix EventQueue close races
- Roll back terminalStateReached on enqueue failure so error fallbacks work - Add timed enqueue with interrupt-preserving retry in DefaultRequestHandler - Replace boolean awaitingFinalEvent with two-phase counters to prevent premature close - Invoke onClose() inside synchronized block for atomic state transitions - Skip closed child queues during event dispatch This fixes #1183 Co-Authored-By: Emmanuel Hugonnet <ehugonne@redhat.com> Signed-off-by: Kabir Khan <kkhan@redhat.com>
1 parent 93e521a commit dabca6c

6 files changed

Lines changed: 552 additions & 81 deletions

File tree

‎server-common/src/main/java/org/a2aproject/sdk/server/events/EventConsumer.java‎

Lines changed: 8 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -179,10 +179,14 @@ public Flow.Publisher<EventQueueItem> consumeAll() {
179179
if (pollTimeoutsAfterAgentCompleted >= MAX_POLL_TIMEOUTS_AFTER_AGENT_COMPLETED) {
180180
LOGGER.debug("Agent completed with {} consecutive poll timeouts and empty queue, closing for graceful completion (queue={})",
181181
pollTimeoutsAfterAgentCompleted, System.identityHashCode(queue));
182-
queue.close();
183-
completed = true;
184-
tube.complete();
185-
return;
182+
if (queue.closeIfNotAwaitingFinalEvent()) {
183+
completed = true;
184+
tube.complete();
185+
return;
186+
}
187+
// A final event started while this consumer was timing out.
188+
// Keep polling until it is distributed or its producer rolls back.
189+
pollTimeoutsAfterAgentCompleted = 0;
186190
} else {
187191
LOGGER.debug("Agent completed but grace period active ({}/{} timeouts), continuing to poll (queue={})",
188192
pollTimeoutsAfterAgentCompleted, MAX_POLL_TIMEOUTS_AFTER_AGENT_COMPLETED, System.identityHashCode(queue));

0 commit comments

Comments
 (0)