diff --git a/docs/content/configuration.md b/docs/content/configuration.md index b164bf3cf..70d337bc8 100644 --- a/docs/content/configuration.md +++ b/docs/content/configuration.md @@ -40,6 +40,9 @@ a2a.blocking.agent.timeout.seconds=30 # Timeout for event consumption in blocking calls (default: 5 seconds) a2a.blocking.consumption.timeout.seconds=5 + +# Timeout for TaskStore reconciliation polling in blocking calls (default: 1 second) +a2a.blocking.reconciliation.timeout.seconds=1 ``` ### Tuning Guidelines @@ -48,6 +51,7 @@ a2a.blocking.consumption.timeout.seconds=5 - **Resource Management**: The dedicated executor prevents streaming operations from competing with the ForkJoinPool. - **Concurrency**: In production with high concurrent streaming, increase pool sizes accordingly. - **Agent Timeouts**: LLM-based agents may need longer timeouts (60-120s) compared to simple agents. +- **Reconciliation Timeout**: Increase if blocking calls fail with "Could not find a Task/Message" under heavy load or with slow TaskStore implementations. ## MicroProfile Config Integration diff --git a/integrations/microprofile-config/README.md b/integrations/microprofile-config/README.md index ed9fd74ac..351218641 100644 --- a/integrations/microprofile-config/README.md +++ b/integrations/microprofile-config/README.md @@ -37,6 +37,7 @@ a2a.executor.max-pool-size=100 # Timeout configuration a2a.blocking.agent.timeout.seconds=60 a2a.blocking.consumption.timeout.seconds=10 +a2a.blocking.reconciliation.timeout.seconds=2 ``` **Environment variables:** diff --git a/server-common/src/main/java/org/a2aproject/sdk/server/requesthandlers/DefaultRequestHandler.java b/server-common/src/main/java/org/a2aproject/sdk/server/requesthandlers/DefaultRequestHandler.java index db7478fb0..addac8e69 100644 --- a/server-common/src/main/java/org/a2aproject/sdk/server/requesthandlers/DefaultRequestHandler.java +++ b/server-common/src/main/java/org/a2aproject/sdk/server/requesthandlers/DefaultRequestHandler.java @@ -142,8 +142,9 @@ *
  • Blocking (configuration.blocking=true): Client waits for first event or final task state
    • *
  • Streaming: Client receives events as they arrive via reactive streams
    • *
  • Both modes support fire-and-forget (agent continues after client disconnect)
    • - *
  • Configurable timeouts via {@code a2a.blocking.agent.timeout.seconds} and - * {@code a2a.blocking.consumption.timeout.seconds}
    • + *
  • Configurable timeouts via {@code a2a.blocking.agent.timeout.seconds}, + * {@code a2a.blocking.consumption.timeout.seconds}, and + * {@code a2a.blocking.reconciliation.timeout.seconds}
    • * * *

    CDI Dependencies

    @@ -189,6 +190,7 @@ public class DefaultRequestHandler implements RequestHandler { private static final String A2A_BLOCKING_AGENT_TIMEOUT_SECONDS = "a2a.blocking.agent.timeout.seconds"; private static final String A2A_BLOCKING_CONSUMPTION_TIMEOUT_SECONDS = "a2a.blocking.consumption.timeout.seconds"; + private static final String A2A_BLOCKING_RECONCILIATION_TIMEOUT_SECONDS = "a2a.blocking.reconciliation.timeout.seconds"; @Inject A2AConfigProvider configProvider; @@ -215,6 +217,19 @@ public class DefaultRequestHandler implements RequestHandler { */ int consumptionCompletionTimeoutSeconds; + /** + * Timeout in seconds for TaskStore reconciliation polling in blocking calls. + * When the in-memory event capture is empty, the aggregator polls TaskStore + * with a bounded timeout to handle the race where MainEventBusProcessor + * has not yet persisted the task. + *

    + * Property: {@code a2a.blocking.reconciliation.timeout.seconds}
    + * Default: 1 second
    + * Note: Property override requires a configurable {@link A2AConfigProvider} on the classpath + * (e.g., MicroProfileConfigProvider in reference implementations). + */ + int reconciliationTimeoutSeconds; + // Fields set by constructor injection cannot be final. We need a noargs constructor for // Jakarta compatibility, and it seems that making fields set by constructor injection // final, is not proxyable in all runtimes @@ -276,6 +291,8 @@ void initConfig() { configProvider.getValue(A2A_BLOCKING_AGENT_TIMEOUT_SECONDS)); consumptionCompletionTimeoutSeconds = Integer.parseInt( configProvider.getValue(A2A_BLOCKING_CONSUMPTION_TIMEOUT_SECONDS)); + reconciliationTimeoutSeconds = Integer.parseInt( + configProvider.getValue(A2A_BLOCKING_RECONCILIATION_TIMEOUT_SECONDS)); } @@ -291,6 +308,7 @@ public static DefaultRequestHandler create(AgentExecutor agentExecutor, TaskStor mainEventBusProcessor, executor, eventConsumerExecutor); handler.agentCompletionTimeoutSeconds = 5; handler.consumptionCompletionTimeoutSeconds = 2; + handler.reconciliationTimeoutSeconds = 1; return handler; } @@ -377,7 +395,8 @@ public Task onCancelTask(CancelTaskParams params, ServerCallContext context) thr taskStore, null); - ResultAggregator resultAggregator = new ResultAggregator(taskManager, null, executor, eventConsumerExecutor); + ResultAggregator resultAggregator = new ResultAggregator(taskManager, null, executor, eventConsumerExecutor, + SECONDS.toNanos(reconciliationTimeoutSeconds)); EventQueue queue = queueManager.createOrTap(task.id()); EventConsumer consumer = new EventConsumer(queue, eventConsumerExecutor); @@ -450,7 +469,8 @@ public EventKind onMessageSend(MessageSendParams params, ServerCallContext conte // Create queue with real taskId (no tempId parameter needed) EventQueue queue = queueManager.createOrTap(queueTaskId); final java.util.concurrent.atomic.AtomicReference<@NonNull String> taskId = new java.util.concurrent.atomic.AtomicReference<>(queueTaskId); - ResultAggregator resultAggregator = new ResultAggregator(mss.taskManager, null, executor, eventConsumerExecutor); + ResultAggregator resultAggregator = new ResultAggregator(mss.taskManager, null, executor, eventConsumerExecutor, + SECONDS.toNanos(reconciliationTimeoutSeconds)); // Default to blocking per A2A spec (returnImmediately defaults to false, meaning wait for completion) boolean returnImmediately = params.configuration() != null && Boolean.TRUE.equals(params.configuration().returnImmediately()); @@ -668,7 +688,8 @@ public Flow.Publisher onMessageSendStream( .taskId(taskId.get()).build(), version); } - ResultAggregator resultAggregator = new ResultAggregator(mss.taskManager, null, executor, eventConsumerExecutor); + ResultAggregator resultAggregator = new ResultAggregator(mss.taskManager, null, executor, eventConsumerExecutor, + SECONDS.toNanos(reconciliationTimeoutSeconds)); // Create consumer BEFORE starting agent - callback is registered inside registerAndExecuteAgentAsync EventConsumer consumer = new EventConsumer(queue, eventConsumerExecutor); @@ -857,7 +878,8 @@ public Flow.Publisher onSubscribeToTask(TaskIdParams params, } TaskManager taskManager = new TaskManager(task.id(), task.contextId(), taskStore, null); - ResultAggregator resultAggregator = new ResultAggregator(taskManager, null, executor, eventConsumerExecutor); + ResultAggregator resultAggregator = new ResultAggregator(taskManager, null, executor, eventConsumerExecutor, + SECONDS.toNanos(reconciliationTimeoutSeconds)); EventQueue queue = queueManager.tap(task.id()); LOGGER.debug("onSubscribeToTask - tapped queue: {}", queue != null ? System.identityHashCode(queue) : "null"); diff --git a/server-common/src/main/java/org/a2aproject/sdk/server/tasks/ResultAggregator.java b/server-common/src/main/java/org/a2aproject/sdk/server/tasks/ResultAggregator.java index 82d9f8e5d..32c0b4ad0 100644 --- a/server-common/src/main/java/org/a2aproject/sdk/server/tasks/ResultAggregator.java +++ b/server-common/src/main/java/org/a2aproject/sdk/server/tasks/ResultAggregator.java @@ -29,19 +29,21 @@ public class ResultAggregator { private static final Logger LOGGER = LoggerFactory.getLogger(ResultAggregator.class); - private static final long TASK_STORE_RECONCILIATION_TIMEOUT_NANOS = TimeUnit.SECONDS.toNanos(1); private static final long TASK_STORE_RECONCILIATION_POLL_MILLIS = 10; private final TaskManager taskManager; private final Executor executor; private final Executor eventConsumerExecutor; + private final long reconciliationTimeoutNanos; private volatile @Nullable Message message; - public ResultAggregator(TaskManager taskManager, @Nullable Message message, Executor executor, Executor eventConsumerExecutor) { + public ResultAggregator(TaskManager taskManager, @Nullable Message message, Executor executor, + Executor eventConsumerExecutor, long reconciliationTimeoutNanos) { this.taskManager = taskManager; this.message = message; this.executor = executor; this.eventConsumerExecutor = eventConsumerExecutor; + this.reconciliationTimeoutNanos = reconciliationTimeoutNanos; } public @Nullable EventKind getCurrentResult() { @@ -238,7 +240,8 @@ else if (blocking) { Utils.rethrow(error); } - // Return Message if captured, otherwise Task if captured, otherwise reconcile with TaskStore. + // Return Message if captured, otherwise Task if captured, + // otherwise poll TaskStore with bounded timeout (blocking) or single read (non-blocking). EventKind eventKind = message.get(); if (eventKind == null) { eventKind = capturedTask.get(); @@ -268,7 +271,10 @@ else if (blocking) { return task; } - long deadline = System.nanoTime() + TASK_STORE_RECONCILIATION_TIMEOUT_NANOS; + LOGGER.debug("TaskStore reconciliation: task not found on first read, polling for up to {}ms", + TimeUnit.NANOSECONDS.toMillis(reconciliationTimeoutNanos)); + + long deadline = System.nanoTime() + reconciliationTimeoutNanos; while (System.nanoTime() < deadline) { try { Thread.sleep(TASK_STORE_RECONCILIATION_POLL_MILLIS); @@ -279,6 +285,7 @@ else if (blocking) { task = taskManager.getTask(); if (task != null) { + LOGGER.debug("TaskStore reconciliation: task {} found after polling", task.id()); return task; } } diff --git a/server-common/src/main/resources/META-INF/a2a-defaults.properties b/server-common/src/main/resources/META-INF/a2a-defaults.properties index 719be9e7a..9e0d13480 100644 --- a/server-common/src/main/resources/META-INF/a2a-defaults.properties +++ b/server-common/src/main/resources/META-INF/a2a-defaults.properties @@ -10,6 +10,10 @@ a2a.blocking.agent.timeout.seconds=30 # Ensures TaskStore is fully updated before returning to client a2a.blocking.consumption.timeout.seconds=5 +# Timeout for TaskStore reconciliation polling in blocking calls (seconds) +# When in-memory event capture is empty, polls TaskStore with this bounded timeout +a2a.blocking.reconciliation.timeout.seconds=1 + # AsyncExecutorProducer - Thread pool configuration # Core pool size for async agent execution a2a.executor.core-pool-size=5 diff --git a/server-common/src/test/java/org/a2aproject/sdk/server/tasks/ResultAggregatorTest.java b/server-common/src/test/java/org/a2aproject/sdk/server/tasks/ResultAggregatorTest.java index 329cdb617..d3adc986e 100644 --- a/server-common/src/test/java/org/a2aproject/sdk/server/tasks/ResultAggregatorTest.java +++ b/server-common/src/test/java/org/a2aproject/sdk/server/tasks/ResultAggregatorTest.java @@ -62,7 +62,7 @@ public class ResultAggregatorTest { @BeforeEach void setUp() { MockitoAnnotations.openMocks(this); - aggregator = new ResultAggregator(mockTaskManager, null, testExecutor, testExecutor); + aggregator = new ResultAggregator(mockTaskManager, null, testExecutor, testExecutor, TimeUnit.SECONDS.toNanos(1)); } // Helper methods for creating sample data @@ -120,7 +120,7 @@ public void onTaskFinalized(String taskId) { @Test void testConstructorWithMessage() { Message initialMessage = createSampleMessage("initial", "msg1", Message.Role.ROLE_USER); - ResultAggregator aggregatorWithMessage = new ResultAggregator(mockTaskManager, initialMessage, testExecutor, testExecutor); + ResultAggregator aggregatorWithMessage = new ResultAggregator(mockTaskManager, initialMessage, testExecutor, testExecutor, TimeUnit.SECONDS.toNanos(1)); // Test that the message is properly stored by checking getCurrentResult assertEquals(initialMessage, aggregatorWithMessage.getCurrentResult()); @@ -131,7 +131,7 @@ void testConstructorWithMessage() { @Test void testGetCurrentResultWithMessageSet() { Message sampleMessage = createSampleMessage("hola", "msg1", Message.Role.ROLE_USER); - ResultAggregator aggregatorWithMessage = new ResultAggregator(mockTaskManager, sampleMessage, testExecutor, testExecutor); + ResultAggregator aggregatorWithMessage = new ResultAggregator(mockTaskManager, sampleMessage, testExecutor, testExecutor, TimeUnit.SECONDS.toNanos(1)); EventKind result = aggregatorWithMessage.getCurrentResult(); @@ -166,7 +166,7 @@ void testConstructorStoresTaskManagerCorrectly() { @Test void testConstructorWithNullMessage() { - ResultAggregator aggregatorWithNullMessage = new ResultAggregator(mockTaskManager, null, testExecutor, testExecutor); + ResultAggregator aggregatorWithNullMessage = new ResultAggregator(mockTaskManager, null, testExecutor, testExecutor, TimeUnit.SECONDS.toNanos(1)); Task expectedTask = createSampleTask("null_msg_task", TaskState.TASK_STATE_WORKING, "ctx1"); when(mockTaskManager.getTask()).thenReturn(expectedTask); @@ -226,7 +226,7 @@ void testMultipleGetCurrentResultCalls() { void testGetCurrentResultWithMessageTakesPrecedence() { // Test that when both message and task are available, message takes precedence Message message = createSampleMessage("priority message", "pri1", Message.Role.ROLE_USER); - ResultAggregator messageAggregator = new ResultAggregator(mockTaskManager, message, testExecutor, testExecutor); + ResultAggregator messageAggregator = new ResultAggregator(mockTaskManager, message, testExecutor, testExecutor, TimeUnit.SECONDS.toNanos(1)); // Even if we set up the task manager to return something, message should take precedence Task task = createSampleTask("should_not_be_returned", TaskState.TASK_STATE_WORKING, "ctx1"); @@ -297,7 +297,7 @@ void testBlockingReturnsTaskWhenStoreBecomesVisibleAfterEmptyCapture() throws Ex TaskManager taskManager = new TaskManager(taskId, "ctx1", taskStore, null); ResultAggregator blockingAggregator = - new ResultAggregator(taskManager, null, testExecutor, testExecutor); + new ResultAggregator(taskManager, null, testExecutor, testExecutor, TimeUnit.SECONDS.toNanos(1)); MainEventBus mainEventBus = new MainEventBus(); InMemoryQueueManager queueManager = @@ -325,7 +325,7 @@ void testBlockingStillRaisesMissingTaskErrorWhenStoreRemainsEmpty() { TaskStore taskStore = mock(TaskStore.class); TaskManager taskManager = new TaskManager(taskId, "ctx1", taskStore, null); ResultAggregator blockingAggregator = - new ResultAggregator(taskManager, null, testExecutor, testExecutor); + new ResultAggregator(taskManager, null, testExecutor, testExecutor, TimeUnit.MILLISECONDS.toNanos(100)); InternalError error = assertThrows(InternalError.class, () -> blockingAggregator.consumeAndBreakOnInterrupt(createClosedEventConsumer(taskId), true)); @@ -339,7 +339,7 @@ void testNonBlockingDoesNotPollTaskStoreWhenCaptureIsEmpty() { TaskStore taskStore = mock(TaskStore.class); TaskManager taskManager = new TaskManager(taskId, "ctx1", taskStore, null); ResultAggregator nonBlockingAggregator = - new ResultAggregator(taskManager, null, testExecutor, testExecutor); + new ResultAggregator(taskManager, null, testExecutor, testExecutor, TimeUnit.SECONDS.toNanos(1)); assertThrows(InternalError.class, () -> nonBlockingAggregator.consumeAndBreakOnInterrupt(createClosedEventConsumer(taskId), false));