Repository navigation
fix: preserve push notification order per task - #1156
Conversation
sendPushNotification submitted each push independently to the shared ForkJoinPool, so delivery order for the same task depended on scheduling rather than production order. Chain pushes onto a per-task tail future instead: each one only starts after the previous one for the same task finishes, while different tasks stay fully independent. This fixes a2aproject#775
| pushNotificationChains.compute(taskId, (id, previous) -> { | ||
| CompletableFuture<Void> previousOrDone = previous != null ? previous : CompletableFuture.completedFuture(null); | ||
| CompletableFuture<Void> next = previousOrDone.thenRunAsync(pushTask, executor); | ||
| next.whenComplete((v, t) -> pushNotificationChains.remove(id, next)); |
There was a problem hiding this comment.
P2: When pushNotificationExecutor is Runnable::run, thenRunAsync completes synchronously. The whenComplete callback therefore removes next during compute, after which compute installs the completed future back into the map. This leaves one completed entry per task, so the map can grow indefinitely and contradicts the cleanup guarantee.\n\nCould the completion callback be registered after compute returns?\n\njava\nCompletableFuture<Void> next = pushNotificationChains.compute(...);\nnext.whenComplete((v, t) -> pushNotificationChains.remove(taskId, next));\n
There was a problem hiding this comment.
Fixed in 8515d09, along the lines you suggested: whenComplete is now attached after compute() returns, so the map always holds next before the callback can try to remove it.
There was a problem hiding this comment.
Thank you too
With a synchronous executor, thenRunAsync completes the chained future before compute()'s remapping function returns, so the whenComplete callback registered inside it fires before the future is even stored in the map -- its remove() is a no-op, leaving one completed entry per task forever. Register whenComplete after compute() returns instead. Found by kabir's review on a2aproject#1156.
Description
CONTRIBUTINGGuide.sendPushNotificationsubmitted each push independently to the sharedForkJoinPool, so delivery order for the same task depended on scheduling rather than production order — a client could seeworking → completedarrive ascompleted → working. Chains pushes onto a per-task tail future instead: each one only starts after the previous one for the same task finishes, while different tasks stay fully independent (no dedicated thread per task, no manual queue to synchronize).Added
MainEventBusProcessorPushNotificationOrderTest: one test proving same-task ordering under a delay pattern that would expose the old race, one proving a slow task never blocks a fast one. Confirmed red-then-green: both fail against the previous implementation (scrambled order), pass against this one. Fullserver-commonsuite: 519 tests, 0 failures.There was an earlier attempt at this in #776 (per-task bounded queue + dedicated consumer thread), closed for inactivity after Gemini flagged a blocking
join()on the critical processing thread and a race in the queue's overflow handling. This avoids both by not using dedicated threads at all.Fixes #775 🦕