From a3463264f809eefa8f56696c4f82faa9ab72dec3 Mon Sep 17 00:00:00 2001 From: Dai Ha Date: Sat, 15 Aug 2026 06:10:24 +0200 Subject: [PATCH] CB-568: fail queued async tickets on teardown --- .../dev/ltms/bridged/msg/MessageService.java | 51 +++++++++++++----- .../ltms/bridged/msg/MessageServiceTest.java | 53 +++++++++++++++++++ 2 files changed, 91 insertions(+), 13 deletions(-) diff --git a/bridged/src/main/java/dev/ltms/bridged/msg/MessageService.java b/bridged/src/main/java/dev/ltms/bridged/msg/MessageService.java index 58a0df8..f949539 100644 --- a/bridged/src/main/java/dev/ltms/bridged/msg/MessageService.java +++ b/bridged/src/main/java/dev/ltms/bridged/msg/MessageService.java @@ -9,6 +9,7 @@ import org.slf4j.Logger; import org.slf4j.LoggerFactory; import java.util.List; +import java.util.Set; import java.util.UUID; import java.util.concurrent.CompletableFuture; import java.util.concurrent.CompletionException; @@ -170,8 +171,8 @@ public final class MessageService { private final Metrics metrics; // CB-502: nullable — no registry in unit tests private final ConcurrentHashMap sessionLocks = new ConcurrentHashMap<>(); private final ConcurrentHashMap tasks = new ConcurrentHashMap<>(); - /** The async task that currently owns a target's send lock. */ - private final ConcurrentHashMap asyncTasksByTarget = new ConcurrentHashMap<>(); + /** Async tasks that have accepted delivery for a target. */ + private final ConcurrentHashMap> asyncTasksByTarget = new ConcurrentHashMap<>(); /** Async tickets paused on a specific {@code bridge_ask} turn. */ private final ConcurrentHashMap asyncTasksByTurn = new ConcurrentHashMap<>(); private final AtomicLong ticketSeq = new AtomicLong(); @@ -300,14 +301,14 @@ public final class MessageService { */ public boolean abandon(String target, String reason) { CompletableFuture waiter = rendezvous.currentWaiter(target); - if (waiter == null || waiter.isDone()) { - return false; // nobody is blocked on this worker — nothing to abandon - } - boolean failed = rendezvous.resolveFailure(waiter, reason); + boolean failed = waiter != null && !waiter.isDone() && rendezvous.resolveFailure(waiter, reason); + boolean asyncFailed = tasks.values().stream() + .filter(task -> target.equals(task.target) && task.question == null) + .anyMatch(task -> task.future.complete(new Reply(Outcome.WORKER_FAILED, reason))); if (failed) { log.warn("abandoning the blocked send to {}: {}", target, reason); } - return failed; + return failed || asyncFailed; } /** @@ -354,6 +355,11 @@ public final class MessageService { * never earned. {@code null} disables the hook. */ public Reply send(String target, String content, long timeoutMillis, Runnable onAccepted) { + return send(target, content, timeoutMillis, onAccepted, null); + } + + /** Run a send, optionally stopping an async task that teardown already failed before acceptance. */ + private Reply send(String target, String content, long timeoutMillis, Runnable onAccepted, Task task) { long deadlineNanos = System.nanoTime() + timeoutMillis * 1_000_000L; ReentrantLock lock = sessionLocks.computeIfAbsent(target, _ -> new ReentrantLock()); @@ -361,6 +367,9 @@ public final class MessageService { return new Reply(Outcome.BUSY, null); // another send held the session the whole window } try { + if (task != null && task.future.isDone()) { + return task.future.getNow(null); + } if (hasAsyncQuestion(target)) { return new Reply(Outcome.BUSY, null); // the worker's current turn is paused for its lead } @@ -527,17 +536,18 @@ public final class MessageService { if (onAccepted != null) { onAccepted.run(); } - asyncTasksByTarget.put(target, task); + asyncTasksByTarget.computeIfAbsent(target, _ -> ConcurrentHashMap.newKeySet()).add(task); }; - Reply result = send(target, content, ASYNC_TIMEOUT_MS, trackingAccepted); + Reply result = send(target, content, ASYNC_TIMEOUT_MS, trackingAccepted, task); if (result.outcome() == Outcome.QUESTION) { - asyncTasksByTarget.remove(target, task); + // Keep the accepted owner until answer() finishes it. markAsyncQuestion may run + // just after resolveQuestion wakes this thread. } else { finishAsyncTask(task, result); } } catch (Throwable t) { task.future.completeExceptionally(t); - asyncTasksByTarget.remove(target, task); + untrackAsyncTarget(task); } }); pruneTerminalTickets(); @@ -600,7 +610,11 @@ public final class MessageService { /** Record the active question for an async ticket; blocking sends have no entry and stay unchanged. */ private void markAsyncQuestion(String target, String text, String turnId) { - Task task = asyncTasksByTarget.get(target); + Set targetTasks = asyncTasksByTarget.get(target); + Task task = targetTasks == null ? null : targetTasks.stream() + .filter(candidate -> !candidate.future.isDone()) + .findFirst() + .orElse(null); if (task != null) { task.question = new Reply(Outcome.QUESTION, text, turnId); task.turnId = turnId; @@ -623,7 +637,7 @@ public final class MessageService { /** Complete and detach an async ticket after its worker's actual terminal reply. */ private void finishAsyncTask(Task task, Reply result) { task.future.complete(result); - asyncTasksByTarget.remove(task.target, task); + untrackAsyncTarget(task); if (task.turnId != null) { asyncTasksByTurn.remove(task.turnId, task); } @@ -642,6 +656,17 @@ public final class MessageService { return asyncTasksByTurn.values().stream().anyMatch(task -> target.equals(task.target)); } + /** Stop tracking a task once it no longer owns an accepted target turn. */ + private void untrackAsyncTarget(Task task) { + Set targetTasks = asyncTasksByTarget.get(task.target); + if (targetTasks != null) { + targetTasks.remove(task); + if (targetTasks.isEmpty()) { + asyncTasksByTarget.remove(task.target, targetTasks); + } + } + } + /** Release the async executor. */ public void close() { asyncExecutor.shutdown(); diff --git a/bridged/src/test/java/dev/ltms/bridged/msg/MessageServiceTest.java b/bridged/src/test/java/dev/ltms/bridged/msg/MessageServiceTest.java index 5743816..7072a7d 100644 --- a/bridged/src/test/java/dev/ltms/bridged/msg/MessageServiceTest.java +++ b/bridged/src/test/java/dev/ltms/bridged/msg/MessageServiceTest.java @@ -639,4 +639,57 @@ class MessageServiceTest { assertTrue(view.detail() != null && view.detail().contains("released"), "and the detail says why, rather than 'worker unknown'"); } + + @Test + void abandonFailsEveryPendingAsyncTicketForTheReleasedTarget() throws Exception { + String first = messages.sendAsync(T, "first task"); + awaitWaiting(); // first task owns the target lock and rendezvous waiter + String second = messages.sendAsync(T, "second task"); // parked on the same lock, not yet queued + + assertTrue(messages.abandon(T, "agent target term_a not found")); + + assertFailedTicket(first, "agent target term_a not found"); + assertFailedTicket(second, "agent target term_a not found"); + } + + @Test + void abandonDoesNotFailAnAsyncTicketWaitingForAnAnswer() throws Exception { + String ticket = messages.sendAsync(T, "task that asks"); + awaitWaiting(); + injectDelivery(); + + CompletableFuture ask = + CompletableFuture.supplyAsync(() -> messages.ask(T, "which config?", 5000)); + MessageService.TaskView asking = awaitTicketPhase(ticket, MessageService.Phase.ASKING); + + assertFalse(messages.abandon(T, "agent target term_a not found"), + "an asking ticket is an active turn, not a pending send to sweep"); + assertEquals(MessageService.Phase.ASKING, messages.poll(ticket).phase()); + + CompletableFuture answer = CompletableFuture.supplyAsync( + () -> messages.answer(asking.turnId(), "config.yaml", 5000)); + assertEquals("config.yaml", ask.get(5, TimeUnit.SECONDS).answer()); + awaitWaiting(); + assertTrue(rendezvous.resolve(T, "done")); + assertEquals(MessageService.Outcome.REPLIED, answer.get(5, TimeUnit.SECONDS).outcome()); + } + + private void assertFailedTicket(String ticket, String reason) throws Exception { + MessageService.TaskView view = awaitTicketPhase(ticket, MessageService.Phase.FAILED); + assertEquals(reason, view.detail()); + } + + private MessageService.TaskView awaitTicketPhase(String ticket, MessageService.Phase phase) throws Exception { + long deadline = System.currentTimeMillis() + 3000; + MessageService.TaskView view; + do { + view = messages.poll(ticket); + if (view.phase() == phase) { + return view; + } + Thread.sleep(5); + } while (System.currentTimeMillis() < deadline); + assertEquals(phase, view.phase()); + return view; + } }