Skip to content

Commit ad9571c

Browse files
authored
test(server): fix flaky testSynchronousExecutorDoesNotLeakChainEntries (a2aproject#1189)
Two bugs caused intermittent failures: 1. Latch count was taskCount (10) but each task generates TWO push notifications -> one for the Task event and one for the StatusUpdate event, both implementing StreamingEventKind -> so the latch could fire with 10 notifications still in flight. Fixed to taskCount * 2. 2. TOCTOU race between the while-loop exit and assertEquals: with a synchronous executor, pushTask runs inside compute()'s lambda before the chain entry is stored, so the latch can reach 0 when count is transiently 0. The while exited on count==0 but assertEquals re-read count and could see 1 if the processor stored the entry in between. Fixed by switching to a do-while that always sleeps before the first check and asserts on the captured count, not a fresh read. Signed-off-by: Emmanuel Hugonnet <ehugonne@redhat.com>
1 parent 09fa265 commit ad9571c

1 file changed

Lines changed: 19 additions & 4 deletions

File tree

‎server-common/src/test/java/org/a2aproject/sdk/server/events/MainEventBusProcessorPushNotificationOrderTest.java‎

Lines changed: 19 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -146,8 +146,17 @@ public void testSynchronousExecutorDoesNotLeakChainEntries() throws InterruptedE
146146
// One entry per DISTINCT task leaks, not repeated pushes for the same one --
147147
// a single task's own map slot just gets overwritten each time. Only several
148148
// different task IDs reveal growth.
149+
//
150+
// Each task produces TWO push notifications: one for the initial Task event and one for the
151+
// StatusUpdate event (both implement StreamingEventKind). With a synchronous executor,
152+
// pushTask runs inside compute()'s lambda -- before the entry is stored in the map -- so
153+
// the latch fires before the entry is even visible. To avoid a TOCTOU race between the
154+
// while-loop exit (count==0 transiently) and the assertEquals re-reading count, we:
155+
// 1. Set the latch to cover ALL push notifications (taskCount * 2).
156+
// 2. Always sleep before the first count check so the last cleanup has time to run.
157+
// 3. Assert on the captured count, not a fresh read.
149158
int taskCount = 10;
150-
CountDownLatch latch = new CountDownLatch(taskCount);
159+
CountDownLatch latch = new CountDownLatch(taskCount * 2);
151160
PushNotificationSender sender = (event, snapshot) -> latch.countDown();
152161

153162
mainEventBusProcessor = new MainEventBusProcessor(mainEventBus, taskStore, sender, queueManager);
@@ -171,10 +180,16 @@ public void testSynchronousExecutorDoesNotLeakChainEntries() throws InterruptedE
171180
assertTrue(latch.await(5, TimeUnit.SECONDS), "All push notifications should complete");
172181

173182
long deadline = System.nanoTime() + TimeUnit.SECONDS.toNanos(5);
174-
while (mainEventBusProcessor.pushNotificationChainCount() != 0 && System.nanoTime() < deadline) {
183+
int count;
184+
do {
185+
// Always sleep before checking: with the sync executor, pushTask fires inside
186+
// compute()'s lambda (before the entry is stored), so the latch may reach 0
187+
// slightly before the final whenComplete removal runs. A brief sleep lets the
188+
// processor thread finish that last cleanup step.
175189
Thread.sleep(10);
176-
}
177-
assertEquals(0, mainEventBusProcessor.pushNotificationChainCount(),
190+
count = mainEventBusProcessor.pushNotificationChainCount();
191+
} while (count != 0 && System.nanoTime() < deadline);
192+
assertEquals(0, count,
178193
"A completed task's chain entry must not be left in the map");
179194
}
180195

0 commit comments

Comments
 (0)