fleetd #324. {@code task.turnId} is read into {@code turnId} exactly once. + * It used to be read twice — once for the null check, once as the removal key — and {@code + * volatile} makes each of those reads individually fresh but does not make the pair atomic. + * {@link #answer} calls this while holding {@code sessionLocks} for the target; {@link #ask}'s + * own timeout path calls {@link #clearAsyncQuestion} (which nulls {@link Task#turnId}) under no + * lock at all. When that unlocked null-out landed between the two reads here, the second read saw + * {@code null} and {@code asyncTasksByTurn.remove(null, task)} threw {@code NullPointerException} + * on the lead's own {@code answer()} call — even though {@code task.future.complete(result)} on + * the line above had already run, so the answer was in fact delivered. Capturing the field once + * removes the torn read; see the ticket for why the wider asymmetry between the locked and + * unlocked sides is not fixed by this alone. + */ private void finishAsyncTask(Task task, Reply result) { task.future.complete(result); - if (task.turnId != null) { - asyncTasksByTurn.remove(task.turnId, task); + String turnId = task.turnId; + if (turnId != null) { + if (finishAsyncTaskRaceHook != null) { + // Test-only (fleetd #324): see the field's own javadoc. + finishAsyncTaskRaceHook.run(); + } + asyncTasksByTurn.remove(turnId, task); } } + /** + * Null in production; test seam for fleetd #324 — invoked from {@link #finishAsyncTask(Task, + * Reply)} right after {@code task.turnId}'s null-check passes and before the (now-local) value is + * used for the removal. A test installs this to force, deterministically, the exact interleaving + * that a real race between this method and {@link #ask}'s unlocked timeout cleanup can otherwise + * only produce by chance: firing it here reproduces "the field went null between the check and the + * use" against the pre-fix code, and demonstrates the fix tolerates it (the captured local is used + * unconditionally, so a hook that nulls the field afterward cannot affect this call). + */ + private volatile Runnable finishAsyncTaskRaceHook; + + /** + * Test-only (fleetd #324): install {@link #finishAsyncTaskRaceHook}. Package-private so the test, + * in the same package, can reach it without widening any production API. + */ + void setFinishAsyncTaskRaceHookForTest(Runnable hook) { + this.finishAsyncTaskRaceHook = hook; + } + + /** + * Test-only (fleetd #324): run the exact production cleanup {@link #ask}'s own timeout path runs + * unlocked — {@link #clearAsyncQuestion(String, boolean)} with {@code forgetTurn=true} — so a test + * can reproduce that specific mutation instead of hand-rolling an approximation of it. + */ + void forgetTurnForTest(String turnId) { + clearAsyncQuestion(turnId, true); + } + /** Complete the async ticket correlated to a specific answered turn. */ private void finishAsyncTask(String turnId, Reply result) { Task task = asyncTasksByTurn.get(turnId); diff --git a/fleetd/src/test/java/dev/ltms/fleet/msg/MessageServiceTest.java b/fleetd/src/test/java/dev/ltms/fleet/msg/MessageServiceTest.java index 7dd20c8..27834dc 100644 --- a/fleetd/src/test/java/dev/ltms/fleet/msg/MessageServiceTest.java +++ b/fleetd/src/test/java/dev/ltms/fleet/msg/MessageServiceTest.java @@ -940,6 +940,54 @@ class MessageServiceTest { assertEquals("PR opened: https://example/pulls/42", view.reply()); } + /** + * fleetd #324: {@code answer()} holds {@code sessionLocks} for the target and, once the worker's + * real terminal reply arrives, calls {@code finishAsyncTask}, which used to read the volatile + * {@code task.turnId} twice — once to check it is non-null, once as the key for + * {@code asyncTasksByTurn.remove}. {@code ask()}'s own timeout path mutates the same field with no + * lock at all. This test does not wait for a real race to land in that narrow window between the + * two reads — instead it drives the exact sequence the ticket describes (worker asks, primary + * answers, worker's real reply arrives) and, via a package-private test hook wired to fire at + * precisely that point, runs the identical production cleanup {@code ask()}'s timeout catch block + * runs ({@code clearAsyncQuestion(turnId, true)}) so the field goes {@code null} between the two + * reads deterministically rather than by chance. + * + *
What this proves: given that exact interleaving, {@code answer()} must not throw and the
+ * ticket must still resolve to the worker's real reply. What it does not prove: that the
+ * interleaving itself is reachable in production — that is established by reading the code (see
+ * the ticket), not by this test, since forcing it via a hook is not the same as two independent
+ * threads racing on their own schedules.
+ */
+ @Test
+ void finishAsyncTaskSurvivesTurnIdGoingNullBetweenItsTwoReads() throws Exception {
+ String ticket = messages.sendAsync(T, "task that asks");
+ awaitWaiting();
+ injectDelivery();
+
+ CompletableFuture