zyr/FT546MWS
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix: enforce stable shadow snapshot boundaries

Browse Source
This commit is contained in:
zgw
2026-09-03 15:07:37 +08:00
parent 935d7af43d
commit d77fac1d56
12 changed files with 256 additions and 88 deletions
Download Patch File Download Diff File Expand all files Collapse all files

25
ruoyi-admin/src/main/java/com/ruoyi/algorithm/shadow/InboundShadowObservation.java
View File

@@ -1,26 +1,43 @@
package com.ruoyi.algorithm.shadow;
import java.time.Instant;
public record InboundShadowObservation(
boolean observed,
long snapshotGeneration,
String warehouseId,
Instant snapshotSynchronizedAt,
String sourceSignature,
String requestId,
String resultCode,
String proposedSpaceId,
String reason) {
public InboundShadowObservation(boolean observed, long snapshotGeneration, String warehouseId,
String requestId, String resultCode, String proposedSpaceId, String reason) {
this(observed, snapshotGeneration, warehouseId, null, null,
requestId, resultCode, proposedSpaceId, reason);
}
public InboundShadowObservation(boolean observed, String warehouseId, String requestId,
String resultCode, String proposedSpaceId, String reason) {
this(observed, 0L, warehouseId, requestId, resultCode, proposedSpaceId, reason);
this(observed, 0L, warehouseId, null, null, requestId, resultCode, proposedSpaceId, reason);
}
public static InboundShadowObservation empty() {
return new InboundShadowObservation(false, 0L, null, null, "NOT_OBSERVED", null, null);
return new InboundShadowObservation(false, 0L, null, null, null,
null, "NOT_OBSERVED", null, null);
}
public static InboundShadowObservation unavailable(long snapshotGeneration, String warehouseId,
String requestId, String resultCode, String reason) {
return new InboundShadowObservation(true, snapshotGeneration, warehouseId, requestId,
resultCode, null, reason);
return new InboundShadowObservation(true, snapshotGeneration, warehouseId, null, null,
requestId, resultCode, null, reason);
}
public static InboundShadowObservation unavailable(PublishedShadowSnapshot snapshot,
String requestId, String resultCode, String reason) {
return new InboundShadowObservation(true, snapshot.generation(), snapshot.warehouseId(),
snapshot.synchronizedAt(), snapshot.sourceSignature(), requestId, resultCode, null, reason);
}
}

28
ruoyi-admin/src/main/java/com/ruoyi/algorithm/shadow/OutboundShadowObservation.java
View File

@@ -1,27 +1,47 @@
package com.ruoyi.algorithm.shadow;
import java.time.Instant;
public record OutboundShadowObservation(
boolean observed,
long snapshotGeneration,
String warehouseId,
Instant snapshotSynchronizedAt,
String sourceSignature,
String requestId,
String resultCode,
String proposedStockId,
String proposedSpaceId,
String reason) {
public OutboundShadowObservation(boolean observed, long snapshotGeneration, String warehouseId,
String requestId, String resultCode, String proposedStockId, String proposedSpaceId,
String reason) {
this(observed, snapshotGeneration, warehouseId, null, null,
requestId, resultCode, proposedStockId, proposedSpaceId, reason);
}
public OutboundShadowObservation(boolean observed, String warehouseId, String requestId,
String resultCode, String proposedStockId, String proposedSpaceId, String reason) {
this(observed, 0L, warehouseId, requestId, resultCode, proposedStockId, proposedSpaceId, reason);
this(observed, 0L, warehouseId, null, null,
requestId, resultCode, proposedStockId, proposedSpaceId, reason);
}
public static OutboundShadowObservation empty() {
return new OutboundShadowObservation(false, 0L, null, null, "NOT_OBSERVED", null, null, null);
return new OutboundShadowObservation(false, 0L, null, null, null,
null, "NOT_OBSERVED", null, null, null);
}
public static OutboundShadowObservation unavailable(long snapshotGeneration, String warehouseId,
String requestId, String resultCode, String reason) {
return new OutboundShadowObservation(true, snapshotGeneration, warehouseId, requestId,
resultCode, null, null, reason);
return new OutboundShadowObservation(true, snapshotGeneration, warehouseId, null, null,
requestId, resultCode, null, null, reason);
}
public static OutboundShadowObservation unavailable(PublishedShadowSnapshot snapshot,
String requestId, String resultCode, String reason) {
return new OutboundShadowObservation(true, snapshot.generation(), snapshot.warehouseId(),
snapshot.synchronizedAt(), snapshot.sourceSignature(),
requestId, resultCode, null, null, reason);
}
}

66
ruoyi-admin/src/main/java/com/ruoyi/algorithm/shadow/ShadowPlanningEngine.java
View File

@@ -21,6 +21,7 @@ import java.util.function.Supplier;
@Component
public class ShadowPlanningEngine {
private static final int INSTANCE_STALE_SECONDS = 120;
private static final int DECISION_REASON_MAX_LENGTH = 2000;
private static final String PUBLISHED = "PUBLISHED";
private final SchedulingFacade schedulingFacade;
@@ -81,8 +82,8 @@ public class ShadowPlanningEngine {
ShadowStockReference reference = inputMapper.selectProjectOutboundReference(
warehouseId, snapshot.generation(), stockoutDetailId);
if (reference == null || isBlank(reference.getProjectNo())) {
return OutboundShadowObservation.unavailable(snapshot.generation(), warehouseId,
stockoutDetailId, "SKIPPED_MISSING_PROJECT",
return OutboundShadowObservation.unavailable(snapshot, stockoutDetailId,
"SKIPPED_MISSING_PROJECT",
outboundDiagnostic(snapshot, stockoutDetailId, reference,
"No project number was available for shadow planning"));
}
@@ -100,8 +101,8 @@ public class ShadowPlanningEngine {
ShadowStockReference reference = inputMapper.selectStockReference(
warehouseId, snapshot.generation(), stockId);
if (reference == null || isBlank(reference.getTrayNo())) {
return OutboundShadowObservation.unavailable(snapshot.generation(), warehouseId,
stockoutDetailId, "SKIPPED_MISSING_STOCK",
return OutboundShadowObservation.unavailable(snapshot, stockoutDetailId,
"SKIPPED_MISSING_STOCK",
outboundDiagnostic(snapshot, stockoutDetailId, reference,
"Selected stock was unavailable to shadow planning"));
}
@@ -116,37 +117,41 @@ public class ShadowPlanningEngine {
String legacySpaceId) {
Objects.requireNonNull(observation, "observation");
transactions.executeWithoutResult(status -> {
requirePublishedGeneration(observation.snapshotGeneration(), observation.warehouseId(),
"published snapshot changed before comparison");
inputMapper.insertComparison(
validateComparisonSnapshot(observation.snapshotGeneration(), observation.warehouseId(),
observation.snapshotSynchronizedAt(), observation.sourceSignature());
int inserted = inputMapper.insertComparison(
UUID.randomUUID().toString(), observation.warehouseId(), "INBOUND_COMPARE",
observation.snapshotGeneration(),
observation.snapshotGeneration(), observation.snapshotSynchronizedAt(),
observation.sourceSignature(),
comparisonCode(observation.proposedSpaceId(), legacySpaceId),
"snapshotGeneration=" + observation.snapshotGeneration()
abbreviate("snapshotGeneration=" + observation.snapshotGeneration()
+ "; request=" + observation.requestId()
+ "; shadow=" + observation.resultCode()
+ "; proposedSpace=" + nullSafe(observation.proposedSpaceId())
+ "; shadowReason=" + nullSafe(observation.reason())
+ "; legacy=" + nullSafe(legacyResultCode)
+ "; legacySpace=" + nullSafe(legacySpaceId));
+ "; legacySpace=" + nullSafe(legacySpaceId), DECISION_REASON_MAX_LENGTH));
requireComparisonInserted(inserted);
});
}
public void compareOutbound(OutboundShadowObservation observation, String legacyResultCode) {
Objects.requireNonNull(observation, "observation");
transactions.executeWithoutResult(status -> {
requirePublishedGeneration(observation.snapshotGeneration(), observation.warehouseId(),
"published snapshot changed before comparison");
inputMapper.insertComparison(
validateComparisonSnapshot(observation.snapshotGeneration(), observation.warehouseId(),
observation.snapshotSynchronizedAt(), observation.sourceSignature());
int inserted = inputMapper.insertComparison(
UUID.randomUUID().toString(), observation.warehouseId(), "OUTBOUND_COMPARE",
observation.snapshotGeneration(), "LEGACY_" + nullSafe(legacyResultCode),
"snapshotGeneration=" + observation.snapshotGeneration()
observation.snapshotGeneration(), observation.snapshotSynchronizedAt(),
observation.sourceSignature(), "LEGACY_" + nullSafe(legacyResultCode),
abbreviate("snapshotGeneration=" + observation.snapshotGeneration()
+ "; request=" + observation.requestId()
+ "; shadow=" + observation.resultCode()
+ "; proposedStock=" + nullSafe(observation.proposedStockId())
+ "; proposedSpace=" + nullSafe(observation.proposedSpaceId())
+ "; shadowReason=" + nullSafe(observation.reason())
+ "; legacy=" + nullSafe(legacyResultCode));
+ "; legacy=" + nullSafe(legacyResultCode), DECISION_REASON_MAX_LENGTH));
requireComparisonInserted(inserted);
});
}
@@ -156,7 +161,7 @@ public class ShadowPlanningEngine {
ShadowInboundReference reference = inputMapper.selectInboundInput(
warehouseId, snapshot.generation(), taskNo, trayNo);
if (reference == null || isBlank(reference.getRequestId()) || isBlank(reference.getTrayNo())) {
return InboundShadowObservation.unavailable(snapshot.generation(), warehouseId, taskNo,
return InboundShadowObservation.unavailable(snapshot, taskNo,
"SKIPPED_MISSING_INPUT",
inboundDiagnostic(snapshot, taskNo, trayNo, reference,
"Inbound task details were unavailable to shadow planning"));
@@ -169,11 +174,12 @@ public class ShadowPlanningEngine {
if (plan.assignment().isPresent()) {
var assignment = plan.assignment().orElseThrow();
return new InboundShadowObservation(true, snapshot.generation(), warehouseId,
snapshot.synchronizedAt(), snapshot.sourceSignature(),
reference.getRequestId(), "READY", assignment.spaceId(),
inboundDiagnostic(snapshot, taskNo, trayNo, reference, assignment.reason()));
}
var rejection = plan.rejection().orElseThrow();
return InboundShadowObservation.unavailable(snapshot.generation(), warehouseId,
return InboundShadowObservation.unavailable(snapshot,
reference.getRequestId(), "REJECTED_" + rejection.code().name(),
inboundDiagnostic(snapshot, taskNo, trayNo, reference, rejection.reason()));
}
@@ -186,17 +192,18 @@ public class ShadowPlanningEngine {
if (plan.assignment().isPresent()) {
var assignment = plan.assignment().orElseThrow();
return new OutboundShadowObservation(true, snapshot.generation(), snapshot.warehouseId(),
snapshot.synchronizedAt(), snapshot.sourceSignature(),
requestId, "READY", reference.getStockId(), assignment.sourceSpaceId(),
outboundDiagnostic(snapshot, requestId, reference, assignment.reason()));
}
if (plan.waiting().isPresent()) {
var waiting = plan.waiting().orElseThrow();
return OutboundShadowObservation.unavailable(snapshot.generation(), snapshot.warehouseId(),
return OutboundShadowObservation.unavailable(snapshot,
requestId, "WAITING_" + waiting.code().name(),
outboundDiagnostic(snapshot, requestId, reference, waiting.reason()));
}
var error = plan.error().orElseThrow();
return OutboundShadowObservation.unavailable(snapshot.generation(), snapshot.warehouseId(),
return OutboundShadowObservation.unavailable(snapshot,
requestId, "ERROR_" + error.code().name(),
outboundDiagnostic(snapshot, requestId, reference, error.message()));
}
@@ -253,16 +260,17 @@ public class ShadowPlanningEngine {
}
}
private void requirePublishedGeneration(long generation, String warehouseId, String message) {
if (generation <= 0 || isBlank(warehouseId)) {
throw new IllegalStateException(message);
private void validateComparisonSnapshot(long generation, String warehouseId,
Instant synchronizedAt, String sourceSignature) {
if (generation <= 0 || isBlank(warehouseId) || synchronizedAt == null) {
throw new IllegalStateException("published snapshot changed before comparison");
}
ShadowSnapshotDescriptor descriptor = inputMapper.selectPublishedSnapshot(warehouseId);
if (descriptor == null
|| !PUBLISHED.equals(descriptor.getSnapshotStatus())
|| descriptor.getSnapshotGeneration() != generation
|| !WarehouseIdNormalizer.equivalent(descriptor.getWarehouseId(), warehouseId)) {
throw new IllegalStateException(message);
requireUppercaseHexSignature(sourceSignature, "comparison source signature");
}
private void requireComparisonInserted(int inserted) {
if (inserted != 1) {
throw new IllegalStateException("published snapshot changed before comparison");
}
}

4
ruoyi-admin/src/main/java/com/ruoyi/algorithm/shadow/mapper/ShadowInputMapper.java
View File

@@ -6,6 +6,8 @@ import com.ruoyi.algorithm.shadow.ShadowStockReference;
import org.apache.ibatis.annotations.Mapper;
import org.apache.ibatis.annotations.Param;
import java.time.Instant;
@Mapper
public interface ShadowInputMapper {
ShadowInboundReference selectInboundInput(@Param("warehouseId") String warehouseId,
@@ -31,6 +33,8 @@ public interface ShadowInputMapper {
@Param("warehouseId") String warehouseId,
@Param("decisionType") String decisionType,
@Param("inputVersion") long inputVersion,
@Param("snapshotSynchronizedAt") Instant snapshotSynchronizedAt,
@Param("sourceSignature") String sourceSignature,
@Param("resultCode") String resultCode,
@Param("reason") String reason);
}

13
ruoyi-admin/src/main/resources/mapper/algorithm/ShadowInputMapper.xml
View File

@@ -96,15 +96,20 @@ PUBLIC "-//mybatis.org//DTD Mapper 3.0//EN"
state.snapshot_status as snapshotStatus,
state.synchronized_at as synchronizedAt,
convert(varchar(64), state.source_signature, 2) as sourceSignature
from alg.shadow_snapshot_state state
from alg.shadow_snapshot_state state with (readcommittedlock)
where state.warehouse_id = #{warehouseId}
</select>
<insert id="insertComparison">
insert into alg.decision_log (
decision_id, warehouse_id, decision_type, input_version, result_code, reason)
values (
try_convert(uniqueidentifier, #{decisionId}), #{warehouseId}, #{decisionType},
convert(binary(8), #{inputVersion}), #{resultCode}, #{reason})
select try_convert(uniqueidentifier, #{decisionId}), state.warehouse_id, #{decisionType},
convert(binary(8), #{inputVersion}), #{resultCode}, #{reason}
from alg.shadow_snapshot_state state with (updlock, holdlock)
where state.warehouse_id = #{warehouseId}
and state.snapshot_status = 'PUBLISHED'
and state.snapshot_generation = #{inputVersion}
and state.synchronized_at = #{snapshotSynchronizedAt}
and convert(varchar(64), state.source_signature, 2) = #{sourceSignature}
</insert>
</mapper>

62
ruoyi-admin/src/test/java/com/ruoyi/algorithm/shadow/ShadowPlanningEngineTest.java
View File

@@ -141,7 +141,8 @@ class ShadowPlanningEngineTest {
assertThatThrownBy(() -> engine.planInbound(SNAPSHOT, "TASK-1", "TRAY-1"))
.isInstanceOf(IllegalStateException.class)
.hasMessageContaining("published snapshot changed during planning");
verify(mapper, never()).insertComparison(any(), any(), any(), any(Long.class), any(), any());
verify(mapper, never()).insertComparison(any(), any(), any(), any(Long.class),
any(Instant.class), any(), any(), any());
}
@Test
@@ -160,30 +161,66 @@ class ShadowPlanningEngineTest {
@Test
void comparisonPersistsLeasedGenerationAsInputVersion() {
ShadowInputMapper mapper = mock(ShadowInputMapper.class);
when(mapper.selectPublishedSnapshot("01")).thenReturn(descriptor(SNAPSHOT));
when(mapper.insertComparison(any(), any(), any(), any(Long.class),
any(Instant.class), any(), any(), any())).thenReturn(1);
ShadowPlanningEngine engine = engine(mock(SchedulingFacade.class), mapper);
InboundShadowObservation observation = new InboundShadowObservation(
true, 42L, "01", "JOB-1", "READY", "SPACE-9", "shadow diagnostic");
true, 42L, "01", SNAPSHOT.synchronizedAt(), SNAPSHOT.sourceSignature(),
"JOB-1", "READY", "SPACE-9", "shadow diagnostic");
engine.compareInbound(observation, "READY", "SPACE-9");
verify(mapper).insertComparison(any(), eq("01"), eq("INBOUND_COMPARE"), eq(42L),
eq("MATCH"), argThat(reason -> reason.contains("request=JOB-1")));
eq(SNAPSHOT.synchronizedAt()), eq("ABC123"), eq("MATCH"),
argThat(reason -> reason.contains("request=JOB-1")));
}
@Test
void comparisonDoesNotPersistAfterPublishedGenerationChanges() {
void comparisonFailsWhenAtomicDescriptorGuardRejectsTheInsert() {
ShadowInputMapper mapper = mock(ShadowInputMapper.class);
when(mapper.selectPublishedSnapshot("01"))
.thenReturn(descriptor(snapshot(43L, "01")));
ShadowPlanningEngine engine = engine(mock(SchedulingFacade.class), mapper);
InboundShadowObservation observation = new InboundShadowObservation(
true, 42L, "01", "JOB-1", "READY", "SPACE-9", "shadow diagnostic");
true, 42L, "01", SNAPSHOT.synchronizedAt(), SNAPSHOT.sourceSignature(),
"JOB-1", "READY", "SPACE-9", "shadow diagnostic");
assertThatThrownBy(() -> engine.compareInbound(observation, "READY", "SPACE-9"))
.isInstanceOf(IllegalStateException.class)
.hasMessageContaining("published snapshot changed before comparison");
verify(mapper, never()).insertComparison(any(), any(), any(), any(Long.class), any(), any());
verify(mapper).insertComparison(any(), eq("01"), eq("INBOUND_COMPARE"), eq(42L),
eq(SNAPSHOT.synchronizedAt()), eq("ABC123"), any(), any());
}
@Test
void outboundComparisonCarriesTheCompleteSnapshotDescriptor() {
ShadowInputMapper mapper = mock(ShadowInputMapper.class);
when(mapper.insertComparison(any(), any(), any(), any(Long.class),
any(Instant.class), any(), any(), any())).thenReturn(1);
ShadowPlanningEngine engine = engine(mock(SchedulingFacade.class), mapper);
OutboundShadowObservation observation = new OutboundShadowObservation(
true, 42L, "01", SNAPSHOT.synchronizedAt(), SNAPSHOT.sourceSignature(),
"DETAIL-1", "READY", "STOCK-1", "SPACE-9", "shadow diagnostic");
engine.compareOutbound(observation, "10");
verify(mapper).insertComparison(any(), eq("01"), eq("OUTBOUND_COMPARE"), eq(42L),
eq(SNAPSHOT.synchronizedAt()), eq("ABC123"), eq("LEGACY_10"), any());
}
@Test
void comparisonReasonIsBoundedToDecisionLogCapacity() {
ShadowInputMapper mapper = mock(ShadowInputMapper.class);
when(mapper.insertComparison(any(), any(), any(), any(Long.class),
any(Instant.class), any(), any(), any())).thenReturn(1);
ShadowPlanningEngine engine = engine(mock(SchedulingFacade.class), mapper);
InboundShadowObservation observation = new InboundShadowObservation(
true, 42L, "01", SNAPSHOT.synchronizedAt(), SNAPSHOT.sourceSignature(),
"JOB-1", "READY", "SPACE-9", "X".repeat(2500));
engine.compareInbound(observation, "READY", "SPACE-9");
verify(mapper).insertComparison(any(), any(), any(), any(Long.class),
any(Instant.class), any(), any(),
argThat(reason -> reason.length() <= 2000));
}
@Test
@@ -247,7 +284,12 @@ class ShadowPlanningEngineTest {
assertThat(statements.get("selectStockReference"))
.contains("alg.inventory_item_snapshot", "stock_id = #{stockid}");
assertThat(statements.get("insertComparison"))
.contains("convert(binary(8), #{inputversion})");
.contains("convert(binary(8), #{inputversion})",
"from alg.shadow_snapshot_state state with (updlock, holdlock)",
"state.snapshot_status = 'published'",
"state.snapshot_generation = #{inputversion}",
"state.synchronized_at = #{snapshotsynchronizedat}",
"convert(varchar(64), state.source_signature, 2) = #{sourcesignature}");
assertThat(statements.get("selectPublishedSnapshot"))
.contains("alg.shadow_snapshot_state", "convert(varchar(64),",
"source_signature", ", 2)");