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 4c0a151..c83c0bd 100644 --- a/bridged/src/main/java/dev/ltms/bridged/msg/MessageService.java +++ b/bridged/src/main/java/dev/ltms/bridged/msg/MessageService.java @@ -9,7 +9,6 @@ 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; @@ -171,8 +170,6 @@ 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<>(); - /** Async tasks that have accepted delivery for a target. */ - private final ConcurrentHashMap> asyncTasksByTarget = new ConcurrentHashMap<>(); /** Async task that owns each exact forward rendezvous waiter. */ private final ConcurrentHashMap, Task> asyncTasksByWaiter = new ConcurrentHashMap<>(); @@ -548,13 +545,7 @@ public final class MessageService { tasks.put(ticket, task); asyncExecutor.submit(() -> { try { - Runnable trackingAccepted = () -> { - if (onAccepted != null) { - onAccepted.run(); - } - asyncTasksByTarget.computeIfAbsent(target, _ -> ConcurrentHashMap.newKeySet()).add(task); - }; - Reply result = send(target, content, ASYNC_TIMEOUT_MS, trackingAccepted, task); + Reply result = send(target, content, ASYNC_TIMEOUT_MS, onAccepted, task); if (result.outcome() == Outcome.QUESTION) { // Keep the accepted owner until answer() finishes it. markAsyncQuestion may run // just after resolveQuestion wakes this thread. @@ -563,7 +554,6 @@ public final class MessageService { } } catch (Throwable t) { task.future.completeExceptionally(t); - untrackAsyncTarget(task); } }); pruneTerminalTickets(); @@ -643,7 +633,6 @@ public final class MessageService { if (forgetTurn) { asyncTasksByTurn.remove(turnId, task); task.turnId = null; - untrackAsyncTarget(task); } } } @@ -651,7 +640,6 @@ 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); - untrackAsyncTarget(task); if (task.turnId != null) { asyncTasksByTurn.remove(task.turnId, task); } @@ -670,17 +658,6 @@ 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 58e3d04..e08a3e7 100644 --- a/bridged/src/test/java/dev/ltms/bridged/msg/MessageServiceTest.java +++ b/bridged/src/test/java/dev/ltms/bridged/msg/MessageServiceTest.java @@ -10,9 +10,6 @@ import dev.ltms.bridged.inject.Injector; import org.junit.jupiter.api.BeforeEach; import org.junit.jupiter.api.Test; -import java.lang.reflect.Field; -import java.util.Map; -import java.util.Set; import java.util.concurrent.CompletableFuture; import java.util.concurrent.TimeUnit; @@ -697,22 +694,10 @@ class MessageServiceTest { } @Test - @SuppressWarnings("unchecked") void asyncQuestionBelongsToTheTaskThatOwnsItsForwardWaiter() throws Exception { String first = messages.sendAsync(T, "first task"); awaitWaiting(); - // Model the resolveQuestion/markAsyncQuestion race: another accepted task reached the target set. - Field byTargetField = MessageService.class.getDeclaredField("asyncTasksByTarget"); - byTargetField.setAccessible(true); - Map> byTarget = (Map>) byTargetField.get(messages); - Set targetTasks = byTarget.get(T); - Class taskClass = Class.forName(MessageService.class.getName() + "$Task"); - var constructor = taskClass.getDeclaredConstructor(String.class); - constructor.setAccessible(true); - targetTasks.clear(); - targetTasks.add(constructor.newInstance(T)); - CompletableFuture ask = CompletableFuture.supplyAsync(() -> messages.ask(T, "which config?", 5000)); assertEquals(MessageService.Phase.ASKING, awaitTicketPhase(first, MessageService.Phase.ASKING).phase());