lhq/FT546-WCS
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat: reconcile feedback fail closed

Browse Source
This commit is contained in:
zgw
2026-07-30 13:46:33 +08:00
parent cdbb59ca65
commit 63071fab6f
4 changed files with 217 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files

41
javaTest/src/main/java/com/ft546/wcs/shadow/feedback/ExecutionFeedback.java Normal file
View File

@@ -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<String> remainingPath;
private final boolean connected;
private final boolean commitAckKnown;
public ExecutionFeedback(String plcBootId, long plannerEpoch, long planVersion, long segmentSeq,
long snapshotSeq, String currentSite, List<String> 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<String>(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<String> getRemainingPath() { return remainingPath; }
public boolean isConnected() { return connected; }
public boolean isCommitAckKnown() { return commitAckKnown; }
}

60
javaTest/src/main/java/com/ft546/wcs/shadow/feedback/FeedbackReconciler.java Normal file
View File

@@ -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.<ResourceKey>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<ResourceKey> released = new LinkedHashSet<ResourceKey>();
Set<ResourceKey> retained = new LinkedHashSet<ResourceKey>();
List<String> 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.<ResourceKey>emptySet(),
allResources(segment.getPath()), segment.transition(SegmentEvent.QUARANTINE));
}
private Set<ResourceKey> allResources(List<String> path) {
Set<ResourceKey> resources = new LinkedHashSet<ResourceKey>();
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;
}
}

33
javaTest/src/main/java/com/ft546/wcs/shadow/feedback/FeedbackResult.java Normal file
View File

@@ -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<ResourceKey> releasedResources;
private final Set<ResourceKey> retainedResources;
private final TrafficSegment updatedSegment;
FeedbackResult(boolean quarantined, String reasonCode, Set<ResourceKey> releasedResources,
Set<ResourceKey> retainedResources, TrafficSegment updatedSegment) {
this.quarantined = quarantined;
this.reasonCode = reasonCode;
this.releasedResources = Collections.unmodifiableSet(
new LinkedHashSet<ResourceKey>(releasedResources));
this.retainedResources = Collections.unmodifiableSet(
new LinkedHashSet<ResourceKey>(retainedResources));
this.updatedSegment = updatedSegment;
}
public boolean isQuarantined() { return quarantined; }
public String getReasonCode() { return reasonCode; }
public Set<ResourceKey> getReleasedResources() { return releasedResources; }
public Set<ResourceKey> getRetainedResources() { return retainedResources; }
public TrafficSegment getUpdatedSegment() { return updatedSegment; }
}

83
javaTest/src/test/java/com/ft546/wcs/shadow/FeedbackReconcilerTest.java Normal file
View File

@@ -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<String> remaining,
boolean connected, boolean ackKnown) {
return new ExecutionFeedback(bootId, epoch, version, segmentSeq, snapshotSeq,
current, remaining, connected, ackKnown);
}
}