From 63071fab6ff7d7fec743e2e7f582dfc38d5647d5 Mon Sep 17 00:00:00 2001 From: zgw <13623675548@163.com> Date: Thu, 30 Jul 2026 13:46:33 +0800 Subject: [PATCH] feat: reconcile feedback fail closed --- .../shadow/feedback/ExecutionFeedback.java | 41 +++++++++ .../shadow/feedback/FeedbackReconciler.java | 60 ++++++++++++++ .../wcs/shadow/feedback/FeedbackResult.java | 33 ++++++++ .../wcs/shadow/FeedbackReconcilerTest.java | 83 +++++++++++++++++++ 4 files changed, 217 insertions(+) create mode 100644 javaTest/src/main/java/com/ft546/wcs/shadow/feedback/ExecutionFeedback.java create mode 100644 javaTest/src/main/java/com/ft546/wcs/shadow/feedback/FeedbackReconciler.java create mode 100644 javaTest/src/main/java/com/ft546/wcs/shadow/feedback/FeedbackResult.java create mode 100644 javaTest/src/test/java/com/ft546/wcs/shadow/FeedbackReconcilerTest.java diff --git a/javaTest/src/main/java/com/ft546/wcs/shadow/feedback/ExecutionFeedback.java b/javaTest/src/main/java/com/ft546/wcs/shadow/feedback/ExecutionFeedback.java new file mode 100644 index 0000000..48da417 --- /dev/null +++ b/javaTest/src/main/java/com/ft546/wcs/shadow/feedback/ExecutionFeedback.java @@ -0,0 +1,41 @@ +package com.ft546.wcs.shadow.feedback; + +import java.util.ArrayList; +import java.util.Collections; +import java.util.List; + +public final class ExecutionFeedback { + private final String plcBootId; + private final long plannerEpoch; + private final long planVersion; + private final long segmentSeq; + private final long snapshotSeq; + private final String currentSite; + private final List remainingPath; + private final boolean connected; + private final boolean commitAckKnown; + + public ExecutionFeedback(String plcBootId, long plannerEpoch, long planVersion, long segmentSeq, + long snapshotSeq, String currentSite, List remainingPath, + boolean connected, boolean commitAckKnown) { + this.plcBootId = plcBootId; + this.plannerEpoch = plannerEpoch; + this.planVersion = planVersion; + this.segmentSeq = segmentSeq; + this.snapshotSeq = snapshotSeq; + this.currentSite = currentSite; + this.remainingPath = Collections.unmodifiableList(new ArrayList(remainingPath)); + this.connected = connected; + this.commitAckKnown = commitAckKnown; + } + + public String getPlcBootId() { return plcBootId; } + public long getPlannerEpoch() { return plannerEpoch; } + public long getPlanVersion() { return planVersion; } + public long getSegmentSeq() { return segmentSeq; } + public long getSnapshotSeq() { return snapshotSeq; } + public String getCurrentSite() { return currentSite; } + public List getRemainingPath() { return remainingPath; } + public boolean isConnected() { return connected; } + public boolean isCommitAckKnown() { return commitAckKnown; } +} diff --git a/javaTest/src/main/java/com/ft546/wcs/shadow/feedback/FeedbackReconciler.java b/javaTest/src/main/java/com/ft546/wcs/shadow/feedback/FeedbackReconciler.java new file mode 100644 index 0000000..500aa67 --- /dev/null +++ b/javaTest/src/main/java/com/ft546/wcs/shadow/feedback/FeedbackReconciler.java @@ -0,0 +1,60 @@ +package com.ft546.wcs.shadow.feedback; + +import com.ft546.wcs.shadow.resource.ResourceKey; +import com.ft546.wcs.shadow.segment.SegmentEvent; +import com.ft546.wcs.shadow.segment.TrafficSegment; + +import java.util.Collections; +import java.util.LinkedHashSet; +import java.util.List; +import java.util.Set; + +public final class FeedbackReconciler { + public FeedbackResult reconcile(TrafficSegment segment, ExecutionFeedback feedback) { + if (!feedback.isConnected()) return quarantine(segment, "COMMUNICATION_LOST"); + if (!feedback.isCommitAckKnown()) return quarantine(segment, "COMMIT_ACK_UNKNOWN"); + if (!segment.getPlcBootId().equals(feedback.getPlcBootId())) { + return quarantine(segment, "PLC_RESTARTED"); + } + if (segment.getPlannerEpoch() != feedback.getPlannerEpoch() + || segment.getPlanVersion() != feedback.getPlanVersion() + || segment.getSegmentSeq() != feedback.getSegmentSeq()) { + return quarantine(segment, "VERSION_MISMATCH"); + } + if (feedback.getSnapshotSeq() <= segment.getLastSnapshotSeq()) { + return new FeedbackResult(false, "DUPLICATE_SNAPSHOT", + Collections.emptySet(), allResources(segment.getPath()), segment); + } + int currentIndex = segment.getPath().indexOf(feedback.getCurrentSite()); + if (currentIndex < 0 + || !segment.getPath().subList(currentIndex, segment.getPath().size()) + .equals(feedback.getRemainingPath())) { + return quarantine(segment, "PATH_MISMATCH"); + } + Set released = new LinkedHashSet(); + Set retained = new LinkedHashSet(); + List path = segment.getPath(); + for (int i = 0; i < path.size(); i++) { + (i < currentIndex ? released : retained).add(ResourceKey.node(path.get(i))); + if (i < path.size() - 1) { + (i < currentIndex ? released : retained).add(ResourceKey.edge(path.get(i), path.get(i + 1))); + } + } + return new FeedbackResult(false, "MATCHED_PROGRESS", released, retained, + segment.withLastSnapshotSeq(feedback.getSnapshotSeq())); + } + + private FeedbackResult quarantine(TrafficSegment segment, String reason) { + return new FeedbackResult(true, reason, Collections.emptySet(), + allResources(segment.getPath()), segment.transition(SegmentEvent.QUARANTINE)); + } + + private Set allResources(List path) { + Set resources = new LinkedHashSet(); + for (int i = 0; i < path.size(); i++) { + resources.add(ResourceKey.node(path.get(i))); + if (i < path.size() - 1) resources.add(ResourceKey.edge(path.get(i), path.get(i + 1))); + } + return resources; + } +} diff --git a/javaTest/src/main/java/com/ft546/wcs/shadow/feedback/FeedbackResult.java b/javaTest/src/main/java/com/ft546/wcs/shadow/feedback/FeedbackResult.java new file mode 100644 index 0000000..7ce9b95 --- /dev/null +++ b/javaTest/src/main/java/com/ft546/wcs/shadow/feedback/FeedbackResult.java @@ -0,0 +1,33 @@ +package com.ft546.wcs.shadow.feedback; + +import com.ft546.wcs.shadow.resource.ResourceKey; +import com.ft546.wcs.shadow.segment.TrafficSegment; + +import java.util.Collections; +import java.util.LinkedHashSet; +import java.util.Set; + +public final class FeedbackResult { + private final boolean quarantined; + private final String reasonCode; + private final Set releasedResources; + private final Set retainedResources; + private final TrafficSegment updatedSegment; + + FeedbackResult(boolean quarantined, String reasonCode, Set releasedResources, + Set retainedResources, TrafficSegment updatedSegment) { + this.quarantined = quarantined; + this.reasonCode = reasonCode; + this.releasedResources = Collections.unmodifiableSet( + new LinkedHashSet(releasedResources)); + this.retainedResources = Collections.unmodifiableSet( + new LinkedHashSet(retainedResources)); + this.updatedSegment = updatedSegment; + } + + public boolean isQuarantined() { return quarantined; } + public String getReasonCode() { return reasonCode; } + public Set getReleasedResources() { return releasedResources; } + public Set getRetainedResources() { return retainedResources; } + public TrafficSegment getUpdatedSegment() { return updatedSegment; } +} diff --git a/javaTest/src/test/java/com/ft546/wcs/shadow/FeedbackReconcilerTest.java b/javaTest/src/test/java/com/ft546/wcs/shadow/FeedbackReconcilerTest.java new file mode 100644 index 0000000..78f19ab --- /dev/null +++ b/javaTest/src/test/java/com/ft546/wcs/shadow/FeedbackReconcilerTest.java @@ -0,0 +1,83 @@ +package com.ft546.wcs.shadow; + +import com.ft546.wcs.shadow.feedback.ExecutionFeedback; +import com.ft546.wcs.shadow.feedback.FeedbackReconciler; +import com.ft546.wcs.shadow.feedback.FeedbackResult; +import com.ft546.wcs.shadow.resource.ResourceKey; +import com.ft546.wcs.shadow.segment.ReservationState; +import com.ft546.wcs.shadow.segment.SegmentEvent; +import com.ft546.wcs.shadow.segment.TrafficSegment; +import org.junit.jupiter.api.Test; + +import java.util.Arrays; +import java.util.Collections; + +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 FeedbackReconcilerTest { + private final FeedbackReconciler reconciler = new FeedbackReconciler(); + + @Test + void releasesOnlyPassedNodeAndEdgeAndKeepsDestinationProtected() { + FeedbackResult result = reconciler.reconcile(committed(), feedback( + "BOOT-1", 7L, 2L, 1L, 10L, "B", Arrays.asList("B", "C"), true, true)); + + assertFalse(result.isQuarantined()); + assertEquals("MATCHED_PROGRESS", result.getReasonCode()); + assertTrue(result.getReleasedResources().contains(ResourceKey.node("A"))); + assertTrue(result.getReleasedResources().contains(ResourceKey.edge("A", "B"))); + assertTrue(result.getRetainedResources().contains(ResourceKey.node("B"))); + assertTrue(result.getRetainedResources().contains(ResourceKey.node("C"))); + } + + @Test + void duplicateSnapshotIsIdempotentAndEtaNeverReleases() { + TrafficSegment processed = committed().withLastSnapshotSeq(10L); + FeedbackResult duplicate = reconciler.reconcile(processed, feedback( + "BOOT-1", 7L, 2L, 1L, 10L, "C", Collections.singletonList("C"), true, true)); + + assertEquals("DUPLICATE_SNAPSHOT", duplicate.getReasonCode()); + assertTrue(duplicate.getReleasedResources().isEmpty()); + assertEquals(ReservationState.HARD, duplicate.getUpdatedSegment().getReservationState()); + } + + @Test + void mismatchRestartDisconnectAndAckLossFailClosed() { + assertQuarantined("VERSION_MISMATCH", feedback( + "BOOT-1", 7L, 3L, 1L, 10L, "B", Arrays.asList("B", "C"), true, true)); + assertQuarantined("PLC_RESTARTED", feedback( + "BOOT-2", 7L, 2L, 1L, 10L, "B", Arrays.asList("B", "C"), true, true)); + assertQuarantined("COMMUNICATION_LOST", feedback( + "BOOT-1", 7L, 2L, 1L, 10L, "B", Arrays.asList("B", "C"), false, true)); + assertQuarantined("COMMIT_ACK_UNKNOWN", feedback( + "BOOT-1", 7L, 2L, 1L, 10L, "B", Arrays.asList("B", "C"), true, false)); + assertQuarantined("PATH_MISMATCH", feedback( + "BOOT-1", 7L, 2L, 1L, 10L, "C", Arrays.asList("C", "B"), true, true)); + } + + private void assertQuarantined(String reason, ExecutionFeedback feedback) { + FeedbackResult result = reconciler.reconcile(committed(), feedback); + assertTrue(result.isQuarantined()); + assertEquals(reason, result.getReasonCode()); + assertTrue(result.getReleasedResources().isEmpty()); + assertEquals(ReservationState.UNKNOWN_HARD, result.getUpdatedSegment().getReservationState()); + } + + private TrafficSegment committed() { + return TrafficSegment.planned("SEG-1", "V1", "BOOT-1", 7L, 2L, 1L, + Arrays.asList("A", "B", "C")) + .transition(SegmentEvent.PREPARE) + .transition(SegmentEvent.PREPARED) + .transition(SegmentEvent.BEGIN_COMMIT) + .transition(SegmentEvent.COMMITTED); + } + + private ExecutionFeedback feedback(String bootId, long epoch, long version, long segmentSeq, + long snapshotSeq, String current, java.util.List remaining, + boolean connected, boolean ackKnown) { + return new ExecutionFeedback(bootId, epoch, version, segmentSeq, snapshotSeq, + current, remaining, connected, ackKnown); + } +}