fleet/fleetd
3
0
Fork 0
Code Issues 73 Pull Requests Actions Packages Projects Releases 1 Wiki Activity

fleetd#329: fix silent async-ticket bugs in MessageService (F1/F2/F3)
CI / contract (pull_request) Successful in 50s
Details
CI / build (pull_request) Failing after 1m28s
Details

Browse Source 
F2 (sendAsync executor catch): log when completeExceptionally returns
false, so an exception thrown after finishAsyncTask already completed
the ticket's future is no longer silently lost.

F1 (answer()'s stranded async ticket): reuse the Task reference answer()
already looked up before rendezvous.answerAsk(), instead of a second
asyncTasksByTurn lookup by turnId in finishAsyncTask. The second lookup
raced ask()'s unlocked timeout cleanup, which could forget turnId first
and leave the ticket stuck PENDING even though answer() itself returned
REPLIED. The #282 chained-ask guard is unaffected: it is still keyed on
result.outcome() == QUESTION, not on this lookup. Removed the now-unused
finishAsyncTask(String, Reply) overload.

F3 (reply()'s orphan recovery path): read orphan.turnId once instead of
twice, closing the same double-read shape fleetd #324 fixed in
finishAsyncTask.

Each fix has its own test plus a test-only race hook (mirroring #324's
finishAsyncTaskRaceHook) to force the exact interleaving deterministically.
Mutation-tested each fix by reverting it, confirming the real failure
(swallowed exception / PENDING ticket / NullPointerException), then
restoring it.

mvn clean install: Tests run: 1345, Failures: 0, Errors: 0, Skipped: 0,
BUILD SUCCESS.
This commit is contained in:
Dai Ha
2026-09-04 15:19:38 +07:00
parent 823976c1b5
commit ea41bbf6b9
2 changed files with 267 additions and 19 deletions
Download Patch File Download Diff File Expand all files Collapse all files
fleetd/src/main/java/dev/ltms/fleet/msg/MessageService.java
+106 -19
View File
@@ -466,8 +466,18 @@ public final class MessageService {
if (candidates.size() == 1) {
Task orphan = candidates.get(0);
if (orphan.future.complete(new Reply(Outcome.REPLIED, content))) {
if (orphan.turnId != null) {
asyncTasksByTurn.remove(orphan.turnId, orphan);
// fleetd #329 (F3): read orphan.turnId once. It used to be read twice, under no
// lock — once for this null check, once as the removal key — the same double-read
// shape fleetd #324 fixed in finishAsyncTask. clearAsyncQuestion's unlocked
// forgetTurn=true path (ask()'s timeout cleanup) can null the field between the two
// reads; capturing it once removes the torn read here too.
String turnId = orphan.turnId;
if (turnId != null) {
if (replyOrphanTurnIdRaceHookForTest != null) {
// Test-only (fleetd #329, F3): see the field's own javadoc.
replyOrphanTurnIdRaceHookForTest.run();
}
asyncTasksByTurn.remove(turnId, orphan);
}
count(FleetMetrics.REPLIES, "path", "async-recovered");
return true; // the ticket itself took it — no inbox stranding at all
@@ -1002,13 +1012,36 @@ public final class MessageService {
// Measured when #282 was merged: this guard is DEFENCE IN DEPTH, not the thing
// that makes the chained ask work. ask() calls markAsyncQuestion (:860) before
// resolveQuestion (:861), so by the time this thread wakes, the task has already
// moved to the new turnId and finishAsyncTask(oldTurnId, ...) finds nothing. Removing
// this guard alone leaves the test green. Keep it anyway: it mirrors sendAsync's
// sibling guard, and that sibling's own comment (:1017) warns the two orderings are
// not something to rely on. Do NOT delete it as dead code without re-checking that
// ordering, and do not treat it as the sole protection either.
if (result.outcome() != Outcome.QUESTION) {
finishAsyncTask(turnId, result);
// moved to the new turnId — this outcome check, not a Task lookup, is what tells the
// two cases apart (see fleetd #329 below). Removing this guard alone leaves the test
// green. Keep it anyway: it mirrors sendAsync's sibling guard, and that sibling's own
// comment (:1017) warns the two orderings are not something to rely on. Do NOT delete
// it as dead code without re-checking that ordering, and do not treat it as the sole
// protection either.
//
// fleetd #329 (F1): complete the SAME Task object this method already looked up at
// :991, rather than re-resolving it from turnId a second time. The old
// finishAsyncTask(turnId, result) did its own asyncTasksByTurn.get(turnId) here, and
// that second lookup races ask()'s own timeout path: ask()'s ticket.answer().get(...)
// can time out (or lose that exact race) at essentially the same instant this method's
// rendezvous.answerAsk(turnId, ...) above already succeeded, running
// markAskTimedOut + clearAsyncQuestion(turnId, true) with no lock at all — which
// forgets turnId (removes it from asyncTasksByTurn, nulls Task.turnId) before this
// thread ever gets here. The worker's real reply then arrived, this method's own wait
// woke up with it, and the by-turnId lookup found nothing: the async ticket's future
// was never completed, so fleet_poll{ticket} stayed PENDING forever even though
// answer() itself correctly returned REPLIED. Reusing the reference captured at :991
// — before that race window opens — sidesteps the second lookup entirely: it is the
// identical Task whatever asyncTasksByTurn or Task.turnId say by the time we reach
// this point, so finishAsyncTask(Task, Reply) can still complete its future and detach
// it using whatever turnId it now reads. The #282 chained-ask case is unaffected
// because it is still gated purely by result.outcome() == QUESTION above, which does
// not depend on this lookup — widening what "task" means here cannot complete a ticket
// the chained ask deliberately left open. When task is null (this was never an async
// ticket — a blocking fleet_ask's answer() call has no Task at all), there is nothing
// to complete, matching the old lookup-miss behaviour.
if (result.outcome() != Outcome.QUESTION && task != null) {
finishAsyncTask(task, result);
}
return result;
} catch (TimeoutException e) {
@@ -1059,7 +1092,7 @@ public final class MessageService {
// this fires exactly once, from whichever path completes it: finishAsyncTask(task, result)
// below on any non-QUESTION outcome of send() — a worker's fleet_reply, the CB-106
// completion fallback, a CB-109 wedge, TIMED_OUT, BUSY, or BACKEND_EXHAUSTED — the same
// finishAsyncTask reached via answer()'s finishAsyncTask(turnId, result) once a QUESTION
// finishAsyncTask reached via answer()'s own finishAsyncTask(task, result) once a QUESTION
// is resolved, completeExceptionally(t) just below when send() itself throws, or a CB-516
// abandon() on teardown. Without this, MessageService.reply's rendezvous fast path (the
// one an async ticket always takes) never told the push loop anything happened — see the
@@ -1079,7 +1112,22 @@ public final class MessageService {
finishAsyncTask(task, result);
}
} catch (Throwable t) {
task.future.completeExceptionally(t);
// fleetd #329 (F2): finishAsyncTask above already completes task.future — on its very
// first line — before doing anything else, so anything that throws afterward (inside
// finishAsyncTask's own cleanup, or from a future addition to this try block) lands
// here with the future already resolved. completeExceptionally on an already-completed
// future is a silent no-op: it returns false and does nothing, so without the check
// below the exception simply vanished — no log, no metric, nothing. Measured (see the
// ticket): temporarily reintroducing the fleetd #324 NPE reproduced 19 real exceptions
// on the ordinary path, with 74/74 tests staying green and not one log line produced.
// Do not stop completing the future first — finishAsyncTask completing before it
// cleans up is what makes a late failure harmless to the ticket's own result — only
// add the missing visibility for the case where that step, or whatever ran after it,
// has already lost the race to report through the future.
if (!task.future.completeExceptionally(t)) {
log.error("async send {} -> {} threw after its ticket was already resolved",
ticket, target, t);
}
}
});
pruneTerminalTickets();
@@ -1247,6 +1295,10 @@ public final class MessageService {
*/
private void finishAsyncTask(Task task, Reply result) {
task.future.complete(result);
if (afterFinishAsyncTaskCompleteHookForTest != null) {
// Test-only (fleetd #329, F2): see the field's own javadoc.
afterFinishAsyncTaskCompleteHookForTest.run();
}
String turnId = task.turnId;
if (turnId != null) {
if (finishAsyncTaskRaceHook != null) {
@@ -1276,6 +1328,49 @@ public final class MessageService {
this.finishAsyncTaskRaceHook = hook;
}
/**
* Null in production; test seam for fleetd #329 (F2) — invoked from {@link #finishAsyncTask(Task,
* Reply)} unconditionally, immediately after {@code task.future.complete(result)} runs (before
* {@link Task#turnId} is even read, so it fires regardless of whether this task was ever asked).
* A test installs this to force an exception into exactly the shape fleetd #329 identified:
* something throws inside {@link #sendAsync}'s executor task after the async ticket's future is
* already resolved, so the surrounding {@code catch (Throwable t)} can only report failure through
* {@code completeExceptionally} — a silent no-op on an already-completed future. Engineering a
* real exception to land in that exact post-completion window is what the ticket itself had to do
* by temporarily deleting a production guard (fleetd #324's {@code turnId != null} check); this
* hook drives the identical shape deterministically instead.
*/
private volatile Runnable afterFinishAsyncTaskCompleteHookForTest;
/**
* Test-only (fleetd #329, F2): install {@link #afterFinishAsyncTaskCompleteHookForTest}.
* Package-private so the test, in the same package, can reach it without widening any production
* API.
*/
void setAfterFinishAsyncTaskCompleteHookForTest(Runnable hook) {
this.afterFinishAsyncTaskCompleteHookForTest = hook;
}
/**
* Null in production; test seam for fleetd #329 (F3) — invoked from {@link #reply} right after
* the single local read of {@code orphan.turnId} passes its null-check and before that (now-local)
* value is used as the {@code asyncTasksByTurn} removal key. Mirrors {@link
* #finishAsyncTaskRaceHook} exactly, for the structurally identical double-read fleetd #324 fixed
* in {@link #finishAsyncTask}: a test installs this to force, deterministically, {@code
* clearAsyncQuestion}'s unlocked {@code forgetTurn=true} path nulling {@link Task#turnId} in that
* exact window, and to confirm the single-read fix tolerates it (the captured local is used
* unconditionally, so a hook that nulls the field afterward cannot affect this call).
*/
private volatile Runnable replyOrphanTurnIdRaceHookForTest;
/**
* Test-only (fleetd #329, F3): install {@link #replyOrphanTurnIdRaceHookForTest}. Package-private
* so the test, in the same package, can reach it without widening any production API.
*/
void setReplyOrphanTurnIdRaceHookForTest(Runnable hook) {
this.replyOrphanTurnIdRaceHookForTest = 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
@@ -1285,14 +1380,6 @@ public final class MessageService {
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);
if (task != null) {
finishAsyncTask(task, result);
}
}
/** A new send must not open a waiter while an async ticket owns this worker's paused turn. */
private boolean hasAsyncQuestion(String target) {
return asyncTasksByTurn.values().stream().anyMatch(task -> target.equals(task.target));
fleetd/src/test/java/dev/ltms/fleet/msg/MessageServiceTest.java
+161
View File
@@ -1,5 +1,9 @@
package dev.ltms.fleet.msg;
import ch.qos.logback.classic.Level;
import ch.qos.logback.classic.Logger;
import ch.qos.logback.classic.spi.ILoggingEvent;
import ch.qos.logback.core.read.ListAppender;
import dev.ltms.fleet.herdr.AgentControl;
import dev.ltms.fleet.herdr.AgentStatus;
import dev.ltms.fleet.herdr.FakeHerdr;
@@ -11,6 +15,7 @@ import dev.ltms.fleet.mcp.PrimaryRegistry;
import dev.ltms.fleet.inject.Injector;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
import org.slf4j.LoggerFactory;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.TimeUnit;
@@ -1041,6 +1046,105 @@ class MessageServiceTest {
"the reply completed its own ticket directly and never touched the inbox");
}
/**
* fleetd #329 (F1). {@code answer()} completes the async ticket by looking {@code turnId} up in
* {@code asyncTasksByTurn} a SECOND time (the first is at :991, purely to re-register the
* {@code asyncTasksByWaiter} entry for #282's chained-ask case). That second lookup races
* {@code ask()}'s own unlocked timeout cleanup ({@code clearAsyncQuestion(turnId, true)}, run from
* {@code markAskTimedOut} + forgetting): {@code ask()}'s {@code ticket.answer().get(timeoutMillis)}
* can time out at essentially the same instant {@code answer()}'s {@code rendezvous.answerAsk}
* call above already succeeded and unblocked the worker. fleetd #324's single-read fix does not
* help here — that fixed a torn read of one already-held {@link MessageService} internal
* {@code Task}; this is a second, independent map lookup by a caller that no longer holds the
* {@code Task} it already found once.
*
* <p>This test does not wait for that real race to land on its own schedule — it drives the exact
* sequence the ticket describes (worker asks, primary answers, worker's real reply arrives) and
* fires the identical production cleanup {@code ask()}'s timeout path runs
* ({@code clearAsyncQuestion(turnId, true)}, via the {@code forgetTurnForTest} seam fleetd #324
* already merged) at the point between "the worker's own {@code ask()} call has unblocked" and
* "the worker's real {@code fleet_reply} arrives" — the exact window fleetd #329 names.
*
* <p>What this proves: given that exact interleaving, the async ticket must still resolve to the
* worker's real reply, not stay stuck {@link MessageService.Phase#PENDING} forever. What it does
* not prove: that the interleaving itself is reachable in production on its own timing — that is
* established by reading the code (see the ticket's "path in"), not by this test, for the same
* reason fleetd #324's own race test says so.
*/
@Test
void aReplyRacingAsksTimeoutCleanupStillCompletesTheAsyncTicket() throws Exception {
String ticket = messages.sendAsync(T, "task that asks");
awaitWaiting();
injectDelivery();
CompletableFuture<MessageService.AskResult> ask =
CompletableFuture.supplyAsync(() -> messages.ask(T, "which config?", 5000));
MessageService.TaskView asking = awaitTicketPhase(ticket, MessageService.Phase.ASKING);
String turnId = asking.turnId();
CompletableFuture<MessageService.Reply> answer =
CompletableFuture.supplyAsync(() -> messages.answer(turnId, "config.yaml", 5000));
assertEquals("config.yaml", ask.get(5, TimeUnit.SECONDS).answer(),
"the worker's own ask() call must have already unblocked with the primary's answer "
+ "before we force the race below");
// ask()'s own unlocked timeout cleanup can forget this exact turnId at essentially the same
// instant answer() has already unblocked the worker and is now waiting on the resumed turn's
// real reply — reproduce that interleaving directly instead of trying to win a real race.
messages.forgetTurnForTest(turnId);
assertTrue(messages.reply(T, "PR opened: https://example/pulls/42"));
assertEquals(MessageService.Outcome.REPLIED, answer.get(5, TimeUnit.SECONDS).outcome(),
"the primary's own answer() call must still see the worker's real reply");
MessageService.TaskView done = awaitTicketPhase(ticket, MessageService.Phase.DONE);
assertEquals("PR opened: https://example/pulls/42", done.reply(),
"fleet_poll{ticket} must return the worker's real reply, not stay PENDING forever "
+ "just because ask()'s timeout cleanup forgot this turnId first");
}
/**
* fleetd #329 (F3). {@link MessageService#reply} reads the same orphan {@code Task}'s {@code
* turnId} twice on this path (once to check it is non-null, once as the {@code
* asyncTasksByTurn.remove} key) — the exact double-read shape fleetd #324 fixed in {@code
* finishAsyncTask}. This test drives the same real sequence as {@code
* aReplyAfterAnswerTimesOutStillCompletesTheAsyncTicket} (worker asks, primary answers, the
* primary's own bounded wait for the resumed turn expires) to reach a task with {@code turnId}
* genuinely stamped and no live rendezvous waiter open — the state {@code askAnsweredAsyncTasks}
* matches here — then fires the identical production cleanup {@code ask()}'s own timeout path
* runs ({@code clearAsyncQuestion(turnId, true)}, via the {@code forgetTurnForTest} seam fleetd
* #324 merged) at the point between the check and the removal use, via a dedicated test hook
* mirroring {@code finishAsyncTaskRaceHook}.
*/
@Test
void replyToAnOrphanedTaskSurvivesTurnIdGoingNullBetweenItsTwoReads() throws Exception {
String ticket = messages.sendAsync(T, "task that asks");
awaitWaiting();
injectDelivery();
CompletableFuture<MessageService.AskResult> ask =
CompletableFuture.supplyAsync(() -> messages.ask(T, "which config?", 5000));
MessageService.TaskView asking = awaitTicketPhase(ticket, MessageService.Phase.ASKING);
String turnId = asking.turnId();
MessageService.Reply answerReply = messages.answer(turnId, "config.yaml", 150);
assertEquals("config.yaml", ask.get(5, TimeUnit.SECONDS).answer());
assertEquals(MessageService.Outcome.TIMED_OUT_WORKING, answerReply.outcome(),
"the primary's own bounded wait must give up first, leaving turnId stamped with no "
+ "live waiter — the state reply()'s F3 code path matches");
messages.setReplyOrphanTurnIdRaceHookForTest(() -> messages.forgetTurnForTest(turnId));
try {
assertTrue(messages.reply(T, "PR opened: https://example/pulls/42"));
MessageService.TaskView done = awaitTicketPhase(ticket, MessageService.Phase.DONE);
assertEquals("PR opened: https://example/pulls/42", done.reply(),
"the ticket must still resolve to the worker's real reply despite the forced race");
} finally {
messages.setReplyOrphanTurnIdRaceHookForTest(null);
}
}
/**
* fleetd #307's ambiguity guard: an ask timeout frees its target ({@code hasAsyncQuestion}
* becomes false the instant it lapses — proven above), so a second, independent delegation can
@@ -1151,6 +1255,63 @@ class MessageServiceTest {
assertEquals("done", done.reply());
}
// --- fleetd #329 (F2): an exception after the ticket's future completes must reach a log -----
//
// sendAsync's own executor task ends with catch (Throwable t) { task.future.completeExceptionally(t); }
// — but finishAsyncTask completes that same future on its first line, so anything that throws
// afterward hits an already-completed future: completeExceptionally returns false and does
// nothing, and (before this fix) nothing logged it either. Measured on the ticket: temporarily
// reintroducing the fleetd #324 NPE reproduced 19 real exceptions on the ordinary path with
// 74/74 tests staying green and zero log lines. There is no reachable production call site where
// finishAsyncTask throws after completing the future (fleetd #324 already closed the one that
// used to), so this test drives the shape directly via a dedicated test-only hook
// (afterFinishAsyncTaskCompleteHookForTest) rather than trying to engineer a real exception into
// that narrow window — the same technique fleetd #324's own race test and this ticket's F1/F3
// tests use for their own races.
private static ListAppender<ILoggingEvent> attachMessageServiceLog() {
Logger logger = (Logger) LoggerFactory.getLogger(MessageService.class);
ListAppender<ILoggingEvent> appender = new ListAppender<>();
appender.start();
logger.addAppender(appender);
return appender;
}
private static void detachMessageServiceLog(ListAppender<ILoggingEvent> appender) {
((Logger) LoggerFactory.getLogger(MessageService.class)).detachAppender(appender);
}
@Test
void anExceptionAfterTheTicketFutureCompletesStillReachesTheLog() throws Exception {
ListAppender<ILoggingEvent> appender = attachMessageServiceLog();
try {
messages.setAfterFinishAsyncTaskCompleteHookForTest(() -> {
throw new RuntimeException("PROBE-329-F2");
});
String ticket = messages.sendAsync(T, "do the task");
awaitWaiting();
assertTrue(rendezvous.resolve(T, "done"));
// finishAsyncTask's own first line already completed the future with the real result
// BEFORE the hook threw — F2 is a visibility gap, not a correctness gap for the ticket
// itself, so the ticket's own outcome must be unaffected by the swallowed exception.
MessageService.TaskView done = awaitTicketPhase(ticket, MessageService.Phase.DONE);
assertEquals("done", done.reply());
assertTrue(appender.list.stream().anyMatch(e ->
e.getLevel() == Level.ERROR
&& e.getFormattedMessage().contains(ticket)
&& e.getThrowableProxy() != null
&& "PROBE-329-F2".equals(e.getThrowableProxy().getMessage())),
"an exception thrown after the ticket's future already completed must still "
+ "reach the log, not vanish silently");
} finally {
messages.setAfterFinishAsyncTaskCompleteHookForTest(null);
detachMessageServiceLog(appender);
}
}
// --- CB-582: fleet_status pendingAsk() ------------------------------------------------------
@Test