feat(ft546): run shared dispatch kernel in shadow

2026-08-14 15:40:31 +08:00
parent 08003407eb
commit 6864010832
22 changed files with 2346 additions and 8 deletions
--- a/ruoyi-wcs/src/main/java/com/ruoyi/wcs/application/ReadOnlyDispatchCycle.java
+++ b/ruoyi-wcs/src/main/java/com/ruoyi/wcs/application/ReadOnlyDispatchCycle.java
@@ -0,0 +1,34 @@
+package com.ruoyi.wcs.application;
+
+import com.ruoyi.wcs.application.dispatch.DispatchCycleResult;
+import com.ruoyi.wcs.application.dispatch.DispatchSnapshot;
+import com.ruoyi.wcs.application.dispatch.GrantedDispatch;
+
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Objects;
+import java.util.function.Function;
+
+public final class ReadOnlyDispatchCycle<T> {
+    private final Function<DispatchSnapshot, DispatchCycleResult> dispatcher;
+    private final Function<GrantedDispatch, T> intentFactory;
+
+    public ReadOnlyDispatchCycle(
+            Function<DispatchSnapshot, DispatchCycleResult> dispatcher,
+            Function<GrantedDispatch, T> intentFactory) {
+        this.dispatcher = Objects.requireNonNull(dispatcher, "dispatcher");
+        this.intentFactory = Objects.requireNonNull(intentFactory,
+                "intentFactory");
+    }
+
+    public ShadowDispatchResult<T> run(DispatchSnapshot snapshot) {
+        DispatchCycleResult decisions = dispatcher.apply(
+                Objects.requireNonNull(snapshot, "snapshot"));
+        List<T> intents = new ArrayList<T>();
+        for (GrantedDispatch grant : decisions.getGranted()) {
+            intents.add(Objects.requireNonNull(intentFactory.apply(grant),
+                    "intentFactory result"));
+        }
+        return new ShadowDispatchResult<T>(decisions, intents);
+    }
+}
--- a/ruoyi-wcs/src/main/java/com/ruoyi/wcs/application/ShadowDispatchResult.java
+++ b/ruoyi-wcs/src/main/java/com/ruoyi/wcs/application/ShadowDispatchResult.java
@@ -0,0 +1,28 @@
+package com.ruoyi.wcs.application;
+
+import com.ruoyi.wcs.application.dispatch.DispatchCycleResult;
+
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+import java.util.Objects;
+
+public final class ShadowDispatchResult<T> {
+    private final DispatchCycleResult decisions;
+    private final List<T> intents;
+
+    public ShadowDispatchResult(DispatchCycleResult decisions,
+                                List<T> intents) {
+        this.decisions = Objects.requireNonNull(decisions, "decisions");
+        this.intents = Collections.unmodifiableList(
+                new ArrayList<T>(Objects.requireNonNull(intents, "intents")));
+    }
+
+    public DispatchCycleResult getDecisions() {
+        return decisions;
+    }
+
+    public List<T> getIntents() {
+        return intents;
+    }
+}
--- a/ruoyi-wcs/src/main/java/com/ruoyi/wcs/domain/traffic/ResourceLedger.java
+++ b/ruoyi-wcs/src/main/java/com/ruoyi/wcs/domain/traffic/ResourceLedger.java
@@ -53,6 +53,51 @@ public final class ResourceLedger {
        this.nextFencingToken = source.nextFencingToken;
    }

+    public static ResourceLedger restore(ResourceLedgerState persisted) {
+        if (persisted == null) {
+            throw new IllegalArgumentException("persisted state is required");
+        }
+        ResourceLedger restored = new ResourceLedger(
+                persisted.getCapacities());
+        long greatestToken = 0L;
+        for (ResourceReservationState reservationState
+                : persisted.getReservations()) {
+            ResourceReservation reservation = ResourceReservation.restore(
+                    reservationState);
+            if (restored.reservations.put(reservation.getReservationId(),
+                    reservation) != null) {
+                throw new IllegalArgumentException(
+                        "duplicate persisted reservation: "
+                                + reservation.getReservationId());
+            }
+            for (String resourceKey : reservation.getResourceKeys()) {
+                if (!restored.capacities.containsKey(resourceKey)) {
+                    throw new IllegalArgumentException(
+                            "persisted reservation references unknown resource: "
+                                    + resourceKey);
+                }
+                if (reservation.getState() == ReservationState.GRANTED
+                        || reservation.getState() == ReservationState.OCCUPIED) {
+                    int next = restored.usage.get(resourceKey) + 1;
+                    if (next > restored.capacities.get(resourceKey)) {
+                        throw new IllegalArgumentException(
+                                "persisted resource usage exceeds capacity: "
+                                        + resourceKey);
+                    }
+                    restored.usage.put(resourceKey, next);
+                }
+            }
+            greatestToken = Math.max(greatestToken,
+                    reservation.getFencingToken());
+        }
+        if (persisted.getNextFencingToken() <= greatestToken) {
+            throw new IllegalArgumentException(
+                    "nextFencingToken must exceed every persisted token");
+        }
+        restored.nextFencingToken = persisted.getNextFencingToken();
+        return restored;
+    }
+
    public synchronized ReservationDecision reserve(ReservationRequest request) {
        if (request == null) {
            throw new IllegalArgumentException("request must not be null");
@@ -192,6 +237,15 @@ public final class ResourceLedger {
        return new ResourceLedger(this);
    }

+    public synchronized ResourceLedgerState snapshotState() {
+        List<ResourceReservationState> states =
+                new ArrayList<ResourceReservationState>(reservations.size());
+        for (ResourceReservation reservation : reservations.values()) {
+            states.add(reservation.snapshotState());
+        }
+        return new ResourceLedgerState(capacities, states, nextFencingToken);
+    }
+
    private void decrementUsage(ResourceReservation reservation) {
        for (String resourceKey : reservation.getResourceKeys()) {
            int current = usage.get(resourceKey);
--- a/ruoyi-wcs/src/main/java/com/ruoyi/wcs/domain/traffic/ResourceLedgerState.java
+++ b/ruoyi-wcs/src/main/java/com/ruoyi/wcs/domain/traffic/ResourceLedgerState.java
@@ -0,0 +1,36 @@
+package com.ruoyi.wcs.domain.traffic;
+
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.LinkedHashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Objects;
+
+public final class ResourceLedgerState {
+    private final Map<String, Integer> capacities;
+    private final List<ResourceReservationState> reservations;
+    private final long nextFencingToken;
+
+    public ResourceLedgerState(Map<String, Integer> capacities,
+                               List<ResourceReservationState> reservations,
+                               long nextFencingToken) {
+        this.capacities = Collections.unmodifiableMap(
+                new LinkedHashMap<String, Integer>(Objects.requireNonNull(
+                        capacities, "capacities")));
+        this.reservations = Collections.unmodifiableList(
+                new ArrayList<ResourceReservationState>(Objects.requireNonNull(
+                        reservations, "reservations")));
+        if (nextFencingToken <= 0L) {
+            throw new IllegalArgumentException(
+                    "nextFencingToken must be positive");
+        }
+        this.nextFencingToken = nextFencingToken;
+    }
+
+    public Map<String, Integer> getCapacities() { return capacities; }
+    public List<ResourceReservationState> getReservations() {
+        return reservations;
+    }
+    public long getNextFencingToken() { return nextFencingToken; }
+}
--- a/ruoyi-wcs/src/main/java/com/ruoyi/wcs/domain/traffic/ResourceReservation.java
+++ b/ruoyi-wcs/src/main/java/com/ruoyi/wcs/domain/traffic/ResourceReservation.java
@@ -28,6 +28,22 @@ public final class ResourceReservation {
        this.recoveryUnresolved = source.recoveryUnresolved;
    }

+    static ResourceReservation restore(ResourceReservationState persisted) {
+        ReservationRequest request = new ReservationRequest(
+                persisted.getReservationId(), persisted.getOrderId(),
+                persisted.getStepId(), persisted.getVehicleId(),
+                persisted.getResourceKeys());
+        ResourceReservation restored = new ResourceReservation(request);
+        restored.state = persisted.getState();
+        restored.fencingToken = persisted.getFencingToken();
+        restored.terminalReason = persisted.getTerminalReason();
+        restored.releaseEvidenceId = persisted.getReleaseEvidenceId();
+        restored.commandUnknown = persisted.isCommandUnknown();
+        restored.recoveryUnresolved = persisted.isRecoveryUnresolved();
+        restored.validateRestoredState();
+        return restored;
+    }
+
    void grant(long token) {
        requireState(ReservationState.REQUESTED);
        if (token <= 0L) {
@@ -97,6 +113,13 @@ public final class ResourceReservation {
        return new ResourceReservation(this);
    }

+    ResourceReservationState snapshotState() {
+        return new ResourceReservationState(getReservationId(), getOrderId(),
+                getStepId(), getVehicleId(), getResourceKeys(), state,
+                fencingToken, terminalReason, releaseEvidenceId,
+                commandUnknown, recoveryUnresolved);
+    }
+
    public String getReservationId() {
        return request.getReservationId();
    }
@@ -145,6 +168,25 @@ public final class ResourceReservation {
        return request.sameAs(candidate);
    }

+    private void validateRestoredState() {
+        boolean tokenRequired = state == ReservationState.GRANTED
+                || state == ReservationState.OCCUPIED
+                || state == ReservationState.RELEASED;
+        if (tokenRequired && fencingToken <= 0L) {
+            throw new IllegalArgumentException(
+                    "active or completed reservation requires a fencing token");
+        }
+        if ((commandUnknown || recoveryUnresolved)
+                && state != ReservationState.OCCUPIED) {
+            throw new IllegalArgumentException(
+                    "reservation protection flags require OCCUPIED state");
+        }
+        if (releaseEvidenceId != null && state != ReservationState.RELEASED) {
+            throw new IllegalArgumentException(
+                    "release evidence requires RELEASED state");
+        }
+    }
+
    private void requireState(ReservationState required) {
        if (state != required) {
            throw new DomainRuleViolation("reservation state must be " + required
--- a/ruoyi-wcs/src/main/java/com/ruoyi/wcs/domain/traffic/ResourceReservationState.java
+++ b/ruoyi-wcs/src/main/java/com/ruoyi/wcs/domain/traffic/ResourceReservationState.java
@@ -0,0 +1,69 @@
+package com.ruoyi.wcs.domain.traffic;
+
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+import java.util.Objects;
+
+public final class ResourceReservationState {
+    private final String reservationId;
+    private final String orderId;
+    private final String stepId;
+    private final String vehicleId;
+    private final List<String> resourceKeys;
+    private final ReservationState state;
+    private final long fencingToken;
+    private final String terminalReason;
+    private final String releaseEvidenceId;
+    private final boolean commandUnknown;
+    private final boolean recoveryUnresolved;
+
+    public ResourceReservationState(
+            String reservationId, String orderId, String stepId,
+            String vehicleId, List<String> resourceKeys,
+            ReservationState state, long fencingToken,
+            String terminalReason, String releaseEvidenceId,
+            boolean commandUnknown, boolean recoveryUnresolved) {
+        this.reservationId = required(reservationId, "reservationId");
+        this.orderId = required(orderId, "orderId");
+        this.stepId = required(stepId, "stepId");
+        this.vehicleId = required(vehicleId, "vehicleId");
+        this.resourceKeys = Collections.unmodifiableList(
+                new ArrayList<String>(Objects.requireNonNull(resourceKeys,
+                        "resourceKeys")));
+        this.state = Objects.requireNonNull(state, "state");
+        if (fencingToken < 0L) {
+            throw new IllegalArgumentException(
+                    "fencingToken must not be negative");
+        }
+        this.fencingToken = fencingToken;
+        this.terminalReason = normalized(terminalReason);
+        this.releaseEvidenceId = normalized(releaseEvidenceId);
+        this.commandUnknown = commandUnknown;
+        this.recoveryUnresolved = recoveryUnresolved;
+    }
+
+    public String getReservationId() { return reservationId; }
+    public String getOrderId() { return orderId; }
+    public String getStepId() { return stepId; }
+    public String getVehicleId() { return vehicleId; }
+    public List<String> getResourceKeys() { return resourceKeys; }
+    public ReservationState getState() { return state; }
+    public long getFencingToken() { return fencingToken; }
+    public String getTerminalReason() { return terminalReason; }
+    public String getReleaseEvidenceId() { return releaseEvidenceId; }
+    public boolean isCommandUnknown() { return commandUnknown; }
+    public boolean isRecoveryUnresolved() { return recoveryUnresolved; }
+
+    private static String required(String value, String field) {
+        String normalized = normalized(value);
+        if (normalized == null) {
+            throw new IllegalArgumentException(field + " must not be blank");
+        }
+        return normalized;
+    }
+
+    private static String normalized(String value) {
+        return value == null || value.trim().isEmpty() ? null : value.trim();
+    }
+}
--- a/ruoyi-wcs/src/main/java/com/ruoyi/wcs/domain/traffic/WaitForGraph.java
+++ b/ruoyi-wcs/src/main/java/com/ruoyi/wcs/domain/traffic/WaitForGraph.java
@@ -65,6 +65,27 @@ public final class WaitForGraph {
        return copy;
    }

+    public synchronized Map<String, Set<String>> snapshotEdges() {
+        Map<String, Set<String>> snapshot =
+                new LinkedHashMap<String, Set<String>>();
+        for (Map.Entry<String, Set<String>> entry : edges.entrySet()) {
+            snapshot.put(entry.getKey(), Collections.unmodifiableSet(
+                    new LinkedHashSet<String>(entry.getValue())));
+        }
+        return Collections.unmodifiableMap(snapshot);
+    }
+
+    public static WaitForGraph restore(Map<String, Set<String>> persisted) {
+        if (persisted == null) {
+            throw new IllegalArgumentException("persisted edges are required");
+        }
+        WaitForGraph graph = new WaitForGraph();
+        for (Map.Entry<String, Set<String>> entry : persisted.entrySet()) {
+            graph.replaceWaits(entry.getKey(), entry.getValue());
+        }
+        return graph;
+    }
+
    private Map<String, Set<String>> copyEdges() {
        Map<String, Set<String>> copy = new LinkedHashMap<String, Set<String>>();
        for (Map.Entry<String, Set<String>> entry : edges.entrySet()) {
--- a/ruoyi-wcs/src/test/java/com/ruoyi/wcs/application/ReadOnlyDispatchCycleTest.java
+++ b/ruoyi-wcs/src/test/java/com/ruoyi/wcs/application/ReadOnlyDispatchCycleTest.java
@@ -0,0 +1,91 @@
+package com.ruoyi.wcs.application;
+
+import com.ruoyi.wcs.application.dispatch.DispatchCycleResult;
+import com.ruoyi.wcs.application.dispatch.DispatchSnapshot;
+import com.ruoyi.wcs.application.dispatch.GrantedDispatch;
+import com.ruoyi.wcs.domain.topology.EffectiveTopology;
+import com.ruoyi.wcs.domain.topology.TopologyCompiler;
+import com.ruoyi.wcs.domain.topology.TopologyDefinition;
+import com.ruoyi.wcs.domain.topology.TopologyNode;
+import com.ruoyi.wcs.domain.traffic.ResourceLedger;
+import com.ruoyi.wcs.domain.traffic.RoutePlan;
+import org.junit.jupiter.api.Test;
+
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.List;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+
+class ReadOnlyDispatchCycleTest {
+
+    @Test
+    void usesSharedDispatchResultAndCreatesOnlyInMemoryIntents() {
+        List<String> calls = new ArrayList<String>();
+        ReadOnlyDispatchCycle<String> cycle = new ReadOnlyDispatchCycle<String>(
+                snapshot -> {
+                    calls.add("dispatch");
+                    return committedGrant();
+                },
+                grant -> {
+                    calls.add("intent:" + grant.getVehicleId());
+                    return "INTENT:" + grant.getVehicleId();
+                });
+
+        ShadowDispatchResult<String> result = cycle.run(snapshot());
+
+        assertEquals(Arrays.asList("dispatch", "intent:VEH-1"), calls);
+        assertEquals(Collections.singletonList("INTENT:VEH-1"),
+                result.getIntents());
+        assertEquals("COMMITTED", result.getDecisions().getStatusCode());
+        assertThrows(UnsupportedOperationException.class,
+                () -> result.getIntents().add("SIDE_EFFECT"));
+    }
+
+    @Test
+    void retryResultNeverCreatesAnIntent() {
+        List<String> calls = new ArrayList<String>();
+        ReadOnlyDispatchCycle<String> cycle = new ReadOnlyDispatchCycle<String>(
+                snapshot -> DispatchCycleResult.retrySnapshotChanged(),
+                grant -> {
+                    calls.add("intent");
+                    return "UNEXPECTED";
+                });
+
+        ShadowDispatchResult<String> result = cycle.run(snapshot());
+
+        assertEquals(Collections.emptyList(), calls);
+        assertEquals(Collections.emptyList(), result.getIntents());
+        assertEquals("RETRY_SNAPSHOT_CHANGED",
+                result.getDecisions().getStatusCode());
+    }
+
+    private static DispatchCycleResult committedGrant() {
+        RoutePlan route = new RoutePlan(Arrays.asList("A", "B"),
+                Arrays.asList("NODE:A", "EDGE:A:B", "NODE:B"),
+                1, 0, 7L);
+        GrantedDispatch grant = new GrantedDispatch("ORDER-1", "STEP-1",
+                "VEH-1", route, Collections.singletonList("RES-1"),
+                7L, 11L);
+        return DispatchCycleResult.committed(Collections.singletonList(grant),
+                Collections.emptyList(), Collections.emptyList(),
+                Collections.emptyList());
+    }
+
+    private static DispatchSnapshot snapshot() {
+        EffectiveTopology topology = new TopologyCompiler().compile(
+                new TopologyDefinition("FT546", 7L, true,
+                        Arrays.asList(
+                                new TopologyNode("A", "F1", 0, 0, 1,
+                                        false, false),
+                                new TopologyNode("B", "F1", 1, 0, 1,
+                                        false, false)),
+                        Collections.emptyList(), Collections.emptyList(),
+                        Collections.emptyList(), Collections.emptyList()));
+        return new DispatchSnapshot(Collections.emptyList(),
+                Collections.emptyList(), topology,
+                new ResourceLedger(topology), 1L);
+    }
+}
--- a/ruoyi-wcs/src/test/java/com/ruoyi/wcs/application/WcsRuntimeCoordinatorTest.java
+++ b/ruoyi-wcs/src/test/java/com/ruoyi/wcs/application/WcsRuntimeCoordinatorTest.java
@@ -100,6 +100,40 @@ class WcsRuntimeCoordinatorTest {
        assertNull(result.getOutboxResult());
    }

+    @Test
+    void tickStillDispatchesProductionOutboxWhenThereAreNoNewGrants() {
+        List<String> calls = new ArrayList<String>();
+        WcsRuntimeCoordinator coordinator = new WcsRuntimeCoordinator(
+                () -> calls.add("snapshot"),
+                () -> {
+                    calls.add("recovery");
+                    return true;
+                },
+                () -> {
+                    calls.add("load");
+                    return snapshot();
+                },
+                value -> {
+                    calls.add("dispatch");
+                    return DispatchCycleResult.committed(
+                            Collections.emptyList(), Collections.emptyList(),
+                            Collections.emptyList(), Collections.emptyList());
+                },
+                value -> calls.add("command"),
+                (limit, now) -> {
+                    calls.add("outbox:" + limit);
+                    return new OutboxDispatchResult();
+                },
+                () -> Instant.parse("2026-08-11T08:00:00Z"), 100);
+
+        RuntimeTickResult result = coordinator.tick();
+
+        assertEquals(Arrays.asList("snapshot", "recovery", "load",
+                "dispatch", "outbox:100"), calls);
+        assertEquals(RuntimeTickResult.Status.COMPLETED, result.getStatus());
+        assertEquals(0, result.getCreatedCommandCount());
+    }
+
    private static GrantedDispatch grant() {
        RoutePlan route = new RoutePlan(Arrays.asList("A", "B"),
                Arrays.asList("NODE:A", "EDGE:A:B", "NODE:B"),