feat(javaTest): arbitrate corridor direction ownership

javaTest/src/main/java/com/ft546/wcs/shadow/history/simulation/CorridorReservationManager.java

@@ -0,0 +1,155 @@
+package com.ft546.wcs.shadow.history.simulation;
+
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.Comparator;
+import java.util.LinkedHashMap;
+import java.util.LinkedHashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+
+public final class CorridorReservationManager {
+    private final Map<String, EdgeReservation> edges =
+            new LinkedHashMap<String, EdgeReservation>();
+    private final Map<String, Set<String>> taskEdges =
+            new LinkedHashMap<String, Set<String>>();
+    private final Map<String, Rank> ranks = new LinkedHashMap<String, Rank>();
+
+    public ReservationDecision reserve(String taskId, List<String> route,
+                                       int effectivePriority, long arrivalMillis) {
+        if (taskId == null || taskId.trim().isEmpty() || route == null || route.isEmpty()) {
+            throw new IllegalArgumentException("task id and route are required");
+        }
+        Rank rank = new Rank(taskId, effectivePriority, arrivalMillis);
+        for (int i = 1; i < route.size(); i++) {
+            EdgeRequest request = request(route.get(i - 1), route.get(i));
+            EdgeReservation existing = edges.get(request.edgeId);
+            if (existing != null && existing.direction != request.direction) {
+                return ReservationDecision.denied(bestOwner(existing.holders),
+                        "CORRIDOR_DIRECTION_OWNED");
+            }
+        }
+        ranks.put(taskId, rank);
+        Set<String> reserved = taskEdges.computeIfAbsent(
+                taskId, ignored -> new LinkedHashSet<String>());
+        for (int i = 1; i < route.size(); i++) {
+            EdgeRequest request = request(route.get(i - 1), route.get(i));
+            EdgeReservation reservation = edges.computeIfAbsent(
+                    request.edgeId, ignored -> new EdgeReservation(request.direction));
+            reservation.holders.add(taskId);
+            reserved.add(request.edgeId);
+        }
+        return ReservationDecision.granted();
+    }
+
+    public void releasePassedEdge(String taskId, String from, String to) {
+        String edgeId = request(from, to).edgeId;
+        EdgeReservation reservation = edges.get(edgeId);
+        if (reservation != null) {
+            reservation.holders.remove(taskId);
+            if (reservation.holders.isEmpty()) edges.remove(edgeId);
+        }
+        Set<String> reserved = taskEdges.get(taskId);
+        if (reserved != null) {
+            reserved.remove(edgeId);
+            if (reserved.isEmpty()) {
+                taskEdges.remove(taskId);
+                ranks.remove(taskId);
+            }
+        }
+    }
+
+    public void releaseTask(String taskId) {
+        Set<String> reserved = taskEdges.remove(taskId);
+        if (reserved != null) {
+            for (String edgeId : new ArrayList<String>(reserved)) {
+                EdgeReservation reservation = edges.get(edgeId);
+                if (reservation == null) continue;
+                reservation.holders.remove(taskId);
+                if (reservation.holders.isEmpty()) edges.remove(edgeId);
+            }
+        }
+        ranks.remove(taskId);
+    }
+
+    public String ownerFor(String left, String right) {
+        EdgeReservation reservation = edges.get(request(left, right).edgeId);
+        return reservation == null ? null : bestOwner(reservation.holders);
+    }
+
+    private String bestOwner(Set<String> holders) {
+        List<Rank> candidates = new ArrayList<Rank>();
+        for (String holder : holders) {
+            Rank rank = ranks.get(holder);
+            if (rank != null) candidates.add(rank);
+        }
+        if (candidates.isEmpty()) return null;
+        Collections.sort(candidates, Comparator.comparingInt((Rank value) -> value.priority)
+                .thenComparingLong(value -> value.arrivalMillis)
+                .thenComparing(value -> value.taskId));
+        return candidates.get(0).taskId;
+    }
+
+    private EdgeRequest request(String from, String to) {
+        if (from == null || to == null || from.equals(to)) {
+            throw new IllegalArgumentException("corridor edge endpoints must differ");
+        }
+        boolean canonical = from.compareTo(to) <= 0;
+        return new EdgeRequest(canonical ? from + "|" + to : to + "|" + from,
+                canonical ? 1 : -1);
+    }
+
+    public static final class ReservationDecision {
+        private final boolean granted;
+        private final String ownerTaskId;
+        private final String reason;
+
+        private ReservationDecision(boolean granted, String ownerTaskId, String reason) {
+            this.granted = granted;
+            this.ownerTaskId = ownerTaskId;
+            this.reason = reason;
+        }
+
+        private static ReservationDecision granted() {
+            return new ReservationDecision(true, null, "GRANTED");
+        }
+
+        private static ReservationDecision denied(String ownerTaskId, String reason) {
+            return new ReservationDecision(false, ownerTaskId, reason);
+        }
+
+        public boolean isGranted() { return granted; }
+        public String getOwnerTaskId() { return ownerTaskId; }
+        public String getReason() { return reason; }
+    }
+
+    private static final class EdgeReservation {
+        private final int direction;
+        private final Set<String> holders = new LinkedHashSet<String>();
+
+        private EdgeReservation(int direction) { this.direction = direction; }
+    }
+
+    private static final class EdgeRequest {
+        private final String edgeId;
+        private final int direction;
+
+        private EdgeRequest(String edgeId, int direction) {
+            this.edgeId = edgeId;
+            this.direction = direction;
+        }
+    }
+
+    private static final class Rank {
+        private final String taskId;
+        private final int priority;
+        private final long arrivalMillis;
+
+        private Rank(String taskId, int priority, long arrivalMillis) {
+            this.taskId = taskId;
+            this.priority = priority;
+            this.arrivalMillis = arrivalMillis;
+        }
+    }
+}

javaTest/src/main/java/com/ft546/wcs/shadow/history/simulation/HistoricalDispatchSimulator.java

@@ -74,6 +74,7 @@ public final class HistoricalDispatchSimulator {
         IdempotentWmsCompletionReceiver wms = new IdempotentWmsCompletionReceiver();
         GridPathPlanner planner = new GridPathPlanner();
         VehicleAssignmentService assignment = new VehicleAssignmentService();
+        CorridorReservationManager reservations = new CorridorReservationManager();
 
         long now = 0L;
         long tick = 0L;
@@ -82,13 +83,13 @@ public final class HistoricalDispatchSimulator {
                 config.getNoProgressLimit(), terminal.size());
         while (terminal.size() < expected && tick < config.getTickLimit()) {
             applyVehicleEvents(tick, now, faults, vehicles, active,
-                    queue, executions, events);
+                    queue, executions, reservations, events);
             assignReadyTasks(tick, now, input, queue, vehicles, active,
-                    executions, terminal, planner, assignment, plc, events);
+                    executions, terminal, planner, assignment, plc, reservations, events);
             Set<String> traversedEdges = new LinkedHashSet<String>();
             advanceActiveTasks(tick, now, input, config, queue, vehicles,
                     active, executions, terminal, planner, ledger, plc, archive, wms,
-                    invariants, events, traversedEdges);
+                    invariants, reservations, events, traversedEdges);
             verifyLedger(ledger, invariants);
 
             if (watchdog.observe(terminal.size())) {
@@ -139,6 +140,7 @@ public final class HistoricalDispatchSimulator {
                                        Map<String, ActiveTask> active,
                                        SimulationTaskQueue queue,
                                        Map<String, Execution> executions,
+                                       CorridorReservationManager reservations,
                                        List<SimulationTickEvent> events) {
         boolean changed = false;
         for (DeterministicFaultPlan.VehicleEvent event : faults.eventsAt(tick)) {
@@ -161,6 +163,7 @@ public final class HistoricalDispatchSimulator {
             if (!vehicle.isAssignmentEligible()) {
                 ActiveTask stopped = active.remove(vehicle.getVehicleId());
                 if (stopped != null) {
+                    reservations.releaseTask(stopped.execution.task.getSimulationTaskId());
                     queue.requeue(stopped.execution.task);
                     hold(stopped.execution, event.getType().name());
                     events.add(event(tick, now, "HOLD", stopped.execution.task.getSimulationTaskId(),
@@ -181,6 +184,7 @@ public final class HistoricalDispatchSimulator {
                                      GridPathPlanner planner,
                                      VehicleAssignmentService assignment,
                                      SimulatedPlcGateway plc,
+                                     CorridorReservationManager reservations,
                                      List<SimulationTickEvent> events) {
         boolean progress = false;
         List<HistoricalSimulationTask> ready = new ArrayList<HistoricalSimulationTask>(
@@ -233,10 +237,14 @@ public final class HistoricalDispatchSimulator {
             List<String> route = new ArrayList<String>(approaches.get(vehicleId));
             for (int i = 1; i < servicePath.size(); i++) route.add(servicePath.get(i));
             Execution execution = executions.get(task.getSimulationTaskId());
-            if (hasOppositeRouteConflict(route, active.values())) {
+            CorridorReservationManager.ReservationDecision reservation = reservations.reserve(
-                hold(execution, "OPPOSITE_ROUTE_CONFLICT");
+                    task.getSimulationTaskId(), route,
+                    queue.effectivePriorityFor(task, now),
+                    queue.arrivalMillis(task.getSimulationTaskId()));
+            if (!reservation.isGranted()) {
+                hold(execution, reservation.getReason());
                 events.add(event(tick, now, "HOLD", task.getSimulationTaskId(), vehicleId,
-                        "OPPOSITE_ROUTE_CONFLICT", route.get(0), route.get(route.size() - 1)));
+                        reservation.getReason(), route.get(0), route.get(route.size() - 1)));
                 continue;
             }
             if (execution.firstAssignedMillis == null) execution.firstAssignedMillis = now;
@@ -252,27 +260,6 @@ public final class HistoricalDispatchSimulator {
         return progress;
     }
 
-    private boolean hasOppositeRouteConflict(List<String> proposed,
-                                             Iterable<ActiveTask> activeTasks) {
-        Set<String> directions = directedEdges(proposed, 0);
-        for (ActiveTask activeTask : activeTasks) {
-            Set<String> activeDirections = directedEdges(activeTask.route, activeTask.index);
-            for (String direction : directions) {
-                String[] pair = direction.split("->", -1);
-                if (activeDirections.contains(pair[1] + "->" + pair[0])) return true;
-            }
-        }
-        return false;
-    }
-
-    private Set<String> directedEdges(List<String> path, int startIndex) {
-        Set<String> edges = new LinkedHashSet<String>();
-        for (int i = Math.max(1, startIndex + 1); i < path.size(); i++) {
-            edges.add(path.get(i - 1) + "->" + path.get(i));
-        }
-        return edges;
-    }
-
     private boolean fleetCanReach(HistoricalSimulationTask task,
                                   Map<String, SimulatedVehicleRuntime> vehicles,
                                   HistoricalSimulationInput input,
@@ -298,6 +285,7 @@ public final class HistoricalDispatchSimulator {
                                        SimulationArchive archive,
                                        IdempotentWmsCompletionReceiver wms,
                                        SimulationInvariantSummary invariants,
+                                       CorridorReservationManager reservations,
                                        List<SimulationTickEvent> events,
                                        Set<String> traversedEdges) {
         boolean progress = false;
@@ -310,7 +298,7 @@ public final class HistoricalDispatchSimulator {
             if (!vehicle.isAssignmentEligible()) continue;
             if (current.index >= current.route.size() - 1) {
                 complete(tick, now, current, vehicle, active, terminal, plc,
-                        archive, wms, invariants, events);
+                        archive, wms, invariants, reservations, events);
                 progress = true;
                 continue;
             }
@@ -321,7 +309,7 @@ public final class HistoricalDispatchSimulator {
             String reverse = to + "->" + from;
             if (traversedEdges.contains(reverse) || traversedEdges.contains(direction)) {
                 recordHold(tick, now, current, vehicle, active, queue,
-                        "OPPOSITE_EDGE_WINDOW", events);
+                        "OPPOSITE_EDGE_WINDOW", reservations, events);
                 continue;
             }
             List<ResourceKey> resources = Arrays.asList(
@@ -347,14 +335,14 @@ public final class HistoricalDispatchSimulator {
             ShadowDecision decision = decisionEvent.getDecisions().get(0);
             if (decision.getType() != DecisionType.GRANT) {
                 recordHold(tick, now, current, vehicle, active, queue,
-                        decision.getReasonCode(), events);
+                        decision.getReasonCode(), reservations, events);
                 continue;
             }
             ResourceLedger.ProtectionResult protection = ledger.evaluate(vehicleId, resources);
             if (!protection.isGranted()) {
                 invariants.capacityViolation();
                 recordHold(tick, now, current, vehicle, active, queue,
-                        "AUTHORITATIVE_RESOURCE_REJECTED", events);
+                        "AUTHORITATIVE_RESOURCE_REJECTED", reservations, events);
                 continue;
             }
             SimulatedPlcGateway.PlcResponse sent = plc.send(
@@ -369,6 +357,7 @@ public final class HistoricalDispatchSimulator {
                 if (plc.attemptsFor(current.execution.task.getSimulationTaskId())
                         % config.getRetryLimit() == 0) {
                     active.remove(vehicleId);
+                    reservations.releaseTask(current.execution.task.getSimulationTaskId());
                     queue.requeue(current.execution.task);
                     hold(current.execution, "RETRY_EXHAUSTED_REDISPATCH");
                     events.add(event(tick, now, "HOLD",
@@ -381,6 +370,8 @@ public final class HistoricalDispatchSimulator {
             Site nextSite = input.getGridMap().getSite(to);
             vehicle.moveTo(to, nextSite.getFloor());
             ledger.release(vehicleId, Arrays.asList(ResourceKey.node(from), edge));
+            reservations.releasePassedEdge(
+                    current.execution.task.getSimulationTaskId(), from, to);
             traversedEdges.add(direction);
             current.index++;
             current.holdStreak = 0;
@@ -390,7 +381,7 @@ public final class HistoricalDispatchSimulator {
             progress = true;
             if (current.index >= current.route.size() - 1) {
                 complete(tick, now, current, vehicle, active, terminal, plc,
-                        archive, wms, invariants, events);
+                        archive, wms, invariants, reservations, events);
             }
         }
         return progress;
@@ -406,6 +397,7 @@ public final class HistoricalDispatchSimulator {
                             SimulatedVehicleRuntime vehicle,
                             Map<String, ActiveTask> active,
                             SimulationTaskQueue queue, String reason,
+                            CorridorReservationManager reservations,
                             List<SimulationTickEvent> events) {
         hold(current.execution, reason);
         current.holdStreak++;
@@ -413,6 +405,7 @@ public final class HistoricalDispatchSimulator {
                 vehicle.getVehicleId(), reason, vehicle.getCurrentSite(), null));
         if (current.holdStreak >= HOLD_REDISPATCH_THRESHOLD) {
             active.remove(vehicle.getVehicleId());
+            reservations.releaseTask(current.execution.task.getSimulationTaskId());
             queue.requeue(current.execution.task);
             events.add(event(tick, now, "REDISPATCH",
                     current.execution.task.getSimulationTaskId(), vehicle.getVehicleId(),
@@ -428,8 +421,10 @@ public final class HistoricalDispatchSimulator {
                           SimulationArchive archive,
                           IdempotentWmsCompletionReceiver wms,
                           SimulationInvariantSummary invariants,
+                          CorridorReservationManager reservations,
                           List<SimulationTickEvent> events) {
         String taskId = activeTask.execution.task.getSimulationTaskId();
+        reservations.releaseTask(taskId);
         if (!archive.complete(taskId, vehicle.getVehicleId(), now)) {
             invariants.duplicateArchiveAttempt();
         }

javaTest/src/test/java/com/ft546/wcs/shadow/history/simulation/CorridorReservationManagerTest.java

@@ -0,0 +1,43 @@
+package com.ft546.wcs.shadow.history.simulation;
+
+import org.junit.jupiter.api.Test;
+
+import java.util.Arrays;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+class CorridorReservationManagerTest {
+    @Test
+    void keepsOneDirectionUntilTheOwnerLeavesTheSharedCorridor() {
+        CorridorReservationManager manager = new CorridorReservationManager();
+
+        assertTrue(manager.reserve("HIGH", Arrays.asList("A", "B", "C"), 1, 20L)
+                .isGranted());
+        CorridorReservationManager.ReservationDecision denied = manager.reserve(
+                "LOW", Arrays.asList("C", "B", "A"), 5, 10L);
+
+        assertFalse(denied.isGranted());
+        assertEquals("HIGH", denied.getOwnerTaskId());
+        assertEquals("HIGH", manager.ownerFor("B", "C"));
+        manager.releasePassedEdge("HIGH", "A", "B");
+        assertFalse(manager.reserve("LOW", Arrays.asList("C", "B", "A"), 0, 10L)
+                .isGranted());
+        manager.releasePassedEdge("HIGH", "B", "C");
+        manager.releaseTask("HIGH");
+        assertTrue(manager.reserve("LOW", Arrays.asList("C", "B", "A"), 0, 10L)
+                .isGranted());
+    }
+
+    @Test
+    void admitsSameDirectionWhilePhysicalResourcesRemainCapacityOne() {
+        CorridorReservationManager manager = new CorridorReservationManager();
+
+        assertTrue(manager.reserve("FIRST", Arrays.asList("A", "B", "C"), 1, 0L)
+                .isGranted());
+        assertTrue(manager.reserve("FOLLOWER", Arrays.asList("A", "B", "C"), 2, 1L)
+                .isGranted());
+        assertEquals("FIRST", manager.ownerFor("A", "B"));
+    }
+}

javaTest/src/test/java/com/ft546/wcs/shadow/history/simulation/HistoricalDispatchSimulatorTest.java

@@ -53,7 +53,8 @@ class HistoricalDispatchSimulatorTest {
         assertEquals(0, result.getWaitingCount());
         assertTrue(result.getEvents().stream().anyMatch(event ->
                 "HOLD".equals(event.getType())
-                        && "OPPOSITE_ROUTE_CONFLICT".equals(event.getReason())));
+                        && "CORRIDOR_DIRECTION_OWNED".equals(event.getReason())));
+        assertEquals(0, result.getInvariants().totalSafetyViolations());
     }
 
     @Test