fix: separate shadow refresh transaction timeout

This commit is contained in:
zgw
2026-09-15 10:58:59 +08:00
parent acc6994385
commit a0af05bfa2
7 changed files with 58 additions and 10 deletions

View File

@@ -9,6 +9,7 @@ public class ShadowObservationProperties {
private boolean enabled; private boolean enabled;
private String warehouseId = ""; private String warehouseId = "";
private int timeoutSeconds = 3; private int timeoutSeconds = 3;
private int refreshTimeoutSeconds = 10;
private int snapshotMaxAgeSeconds = 60; private int snapshotMaxAgeSeconds = 60;
private int quietIntervalMillis = 1000; private int quietIntervalMillis = 1000;
@@ -36,6 +37,14 @@ public class ShadowObservationProperties {
this.timeoutSeconds = timeoutSeconds; this.timeoutSeconds = timeoutSeconds;
} }
public int getRefreshTimeoutSeconds() {
return refreshTimeoutSeconds;
}
public void setRefreshTimeoutSeconds(int refreshTimeoutSeconds) {
this.refreshTimeoutSeconds = refreshTimeoutSeconds;
}
public int getSnapshotMaxAgeSeconds() { public int getSnapshotMaxAgeSeconds() {
return snapshotMaxAgeSeconds; return snapshotMaxAgeSeconds;
} }

View File

@@ -35,17 +35,19 @@ public class ShadowPlanningEngine {
private final SchedulingFacade schedulingFacade; private final SchedulingFacade schedulingFacade;
private final ShadowInputMapper inputMapper; private final ShadowInputMapper inputMapper;
private final ShadowObservationProperties properties; private final ShadowObservationProperties properties;
private final TransactionTemplate transactions; private final TransactionTemplate planningTransactions;
private final TransactionTemplate refreshTransactions;
public ShadowPlanningEngine(SchedulingFacade schedulingFacade, ShadowInputMapper inputMapper, public ShadowPlanningEngine(SchedulingFacade schedulingFacade, ShadowInputMapper inputMapper,
ShadowObservationProperties properties, PlatformTransactionManager transactionManager) { ShadowObservationProperties properties, PlatformTransactionManager transactionManager) {
this.schedulingFacade = Objects.requireNonNull(schedulingFacade, "schedulingFacade"); this.schedulingFacade = Objects.requireNonNull(schedulingFacade, "schedulingFacade");
this.inputMapper = Objects.requireNonNull(inputMapper, "inputMapper"); this.inputMapper = Objects.requireNonNull(inputMapper, "inputMapper");
this.properties = Objects.requireNonNull(properties, "properties"); this.properties = Objects.requireNonNull(properties, "properties");
this.transactions = new TransactionTemplate(Objects.requireNonNull(transactionManager, "transactionManager")); PlatformTransactionManager requiredTransactionManager =
this.transactions.setPropagationBehavior(TransactionDefinition.PROPAGATION_REQUIRES_NEW); Objects.requireNonNull(transactionManager, "transactionManager");
this.transactions.setIsolationLevel(TransactionDefinition.ISOLATION_READ_UNCOMMITTED); this.planningTransactions = transactionTemplate(requiredTransactionManager, properties.getTimeoutSeconds());
this.transactions.setTimeout(properties.getTimeoutSeconds()); this.refreshTransactions = transactionTemplate(
requiredTransactionManager, properties.getRefreshTimeoutSeconds());
} }
public InboundShadowObservation planInbound(String taskNo, String trayNo) { public InboundShadowObservation planInbound(String taskNo, String trayNo) {
@@ -68,7 +70,7 @@ public class ShadowPlanningEngine {
throw new IllegalArgumentException("shadow snapshot instanceId is required"); throw new IllegalArgumentException("shadow snapshot instanceId is required");
} }
String warehouseId = configuredWarehouseId(); String warehouseId = configuredWarehouseId();
return transactions.execute(status -> { return refreshTransactions.execute(status -> {
if (!inputMapper.claimInstance(warehouseId, instanceId, INSTANCE_STALE_SECONDS)) { if (!inputMapper.claimInstance(warehouseId, instanceId, INSTANCE_STALE_SECONDS)) {
throw new IllegalStateException("could not claim shadow snapshot refresh for warehouse " + warehouseId); throw new IllegalStateException("could not claim shadow snapshot refresh for warehouse " + warehouseId);
} }
@@ -215,7 +217,7 @@ public class ShadowPlanningEngine {
public void compareInbound(InboundShadowObservation observation, String legacyResultCode, public void compareInbound(InboundShadowObservation observation, String legacyResultCode,
String legacySpaceId) { String legacySpaceId) {
Objects.requireNonNull(observation, "observation"); Objects.requireNonNull(observation, "observation");
transactions.executeWithoutResult(status -> { planningTransactions.executeWithoutResult(status -> {
validateComparisonSnapshot(observation.snapshotGeneration(), observation.warehouseId(), validateComparisonSnapshot(observation.snapshotGeneration(), observation.warehouseId(),
observation.snapshotSynchronizedAt(), observation.sourceSignature()); observation.snapshotSynchronizedAt(), observation.sourceSignature());
int inserted = inputMapper.insertComparison( int inserted = inputMapper.insertComparison(
@@ -230,7 +232,7 @@ public class ShadowPlanningEngine {
public void compareOutbound(OutboundShadowObservation observation, String legacyResultCode) { public void compareOutbound(OutboundShadowObservation observation, String legacyResultCode) {
Objects.requireNonNull(observation, "observation"); Objects.requireNonNull(observation, "observation");
transactions.executeWithoutResult(status -> { planningTransactions.executeWithoutResult(status -> {
validateComparisonSnapshot(observation.snapshotGeneration(), observation.warehouseId(), validateComparisonSnapshot(observation.snapshotGeneration(), observation.warehouseId(),
observation.snapshotSynchronizedAt(), observation.sourceSignature()); observation.snapshotSynchronizedAt(), observation.sourceSignature());
int inserted = inputMapper.insertComparison( int inserted = inputMapper.insertComparison(
@@ -349,7 +351,7 @@ public class ShadowPlanningEngine {
private <T> T planWithStableSnapshot(PublishedShadowSnapshot snapshot, Supplier<T> planner) { private <T> T planWithStableSnapshot(PublishedShadowSnapshot snapshot, Supplier<T> planner) {
validateLeasedSnapshot(snapshot); validateLeasedSnapshot(snapshot);
return transactions.execute(status -> { return planningTransactions.execute(status -> {
requirePublishedSnapshot(snapshot, "published snapshot changed during planning"); requirePublishedSnapshot(snapshot, "published snapshot changed during planning");
T result = planner.get(); T result = planner.get();
requirePublishedSnapshot(snapshot, "published snapshot changed during planning"); requirePublishedSnapshot(snapshot, "published snapshot changed during planning");
@@ -357,6 +359,15 @@ public class ShadowPlanningEngine {
}); });
} }
private TransactionTemplate transactionTemplate(
PlatformTransactionManager transactionManager, int timeoutSeconds) {
TransactionTemplate template = new TransactionTemplate(transactionManager);
template.setPropagationBehavior(TransactionDefinition.PROPAGATION_REQUIRES_NEW);
template.setIsolationLevel(TransactionDefinition.ISOLATION_READ_UNCOMMITTED);
template.setTimeout(timeoutSeconds);
return template;
}
private void validateLeasedSnapshot(PublishedShadowSnapshot snapshot) { private void validateLeasedSnapshot(PublishedShadowSnapshot snapshot) {
Objects.requireNonNull(snapshot, "snapshot"); Objects.requireNonNull(snapshot, "snapshot");
if (snapshot.generation() <= 0) { if (snapshot.generation() <= 0) {

View File

@@ -36,6 +36,10 @@ public class ShadowSafetyGuard implements InitializingBean {
if (shadowProperties.getTimeoutSeconds() < 1 || shadowProperties.getTimeoutSeconds() > 30) { if (shadowProperties.getTimeoutSeconds() < 1 || shadowProperties.getTimeoutSeconds() > 30) {
throw new IllegalStateException("Shadow observation timeout must be between 1 and 30 seconds"); throw new IllegalStateException("Shadow observation timeout must be between 1 and 30 seconds");
} }
if (shadowProperties.getRefreshTimeoutSeconds() < 5
|| shadowProperties.getRefreshTimeoutSeconds() > 60) {
throw new IllegalStateException("Shadow refresh timeout must be between 5 and 60 seconds");
}
if (shadowProperties.getSnapshotMaxAgeSeconds() < 5 if (shadowProperties.getSnapshotMaxAgeSeconds() < 5
|| shadowProperties.getSnapshotMaxAgeSeconds() > 300) { || shadowProperties.getSnapshotMaxAgeSeconds() > 300) {
throw new IllegalStateException("Shadow snapshot maximum age must be between 5 and 300 seconds"); throw new IllegalStateException("Shadow snapshot maximum age must be between 5 and 300 seconds");

View File

@@ -189,6 +189,7 @@ fortucky:
enabled: ${FORTUCKY_SHADOW_ENABLED:false} enabled: ${FORTUCKY_SHADOW_ENABLED:false}
warehouse-id: ${FORTUCKY_SHADOW_WAREHOUSE_ID:} warehouse-id: ${FORTUCKY_SHADOW_WAREHOUSE_ID:}
timeout-seconds: ${FORTUCKY_SHADOW_TIMEOUT_SECONDS:3} timeout-seconds: ${FORTUCKY_SHADOW_TIMEOUT_SECONDS:3}
refresh-timeout-seconds: ${FORTUCKY_SHADOW_REFRESH_TIMEOUT_SECONDS:10}
snapshot-max-age-seconds: ${FORTUCKY_SHADOW_SNAPSHOT_MAX_AGE_SECONDS:60} snapshot-max-age-seconds: ${FORTUCKY_SHADOW_SNAPSHOT_MAX_AGE_SECONDS:60}
quiet-interval-millis: ${FORTUCKY_SHADOW_QUIET_INTERVAL_MILLIS:1000} quiet-interval-millis: ${FORTUCKY_SHADOW_QUIET_INTERVAL_MILLIS:1000}
algorithm: algorithm:

View File

@@ -33,6 +33,7 @@ class ShadowConfigurationTest {
assertThat(shadow.isEnabled()).isFalse(); assertThat(shadow.isEnabled()).isFalse();
assertThat(shadow.getWarehouseId()).isEmpty(); assertThat(shadow.getWarehouseId()).isEmpty();
assertThat(shadow.getTimeoutSeconds()).isEqualTo(3); assertThat(shadow.getTimeoutSeconds()).isEqualTo(3);
assertThat(shadow.getRefreshTimeoutSeconds()).isEqualTo(10);
assertThat(shadow.getSnapshotMaxAgeSeconds()).isEqualTo(60); assertThat(shadow.getSnapshotMaxAgeSeconds()).isEqualTo(60);
assertThat(shadow.getQuietIntervalMillis()).isEqualTo(1000); assertThat(shadow.getQuietIntervalMillis()).isEqualTo(1000);
assertThat(algorithm.getMode()).isEqualTo(SchedulingMode.SHADOW); assertThat(algorithm.getMode()).isEqualTo(SchedulingMode.SHADOW);

View File

@@ -315,17 +315,23 @@ class ShadowPlanningEngineTest {
@Test @Test
void refreshClaimsInstanceAndReturnsValidatedPublishedSnapshot() { void refreshClaimsInstanceAndReturnsValidatedPublishedSnapshot() {
ShadowObservationProperties properties = properties();
properties.setTimeoutSeconds(3);
properties.setRefreshTimeoutSeconds(11);
CapturingTransactionManager transactions = new CapturingTransactionManager();
ShadowInputMapper mapper = mock(ShadowInputMapper.class); ShadowInputMapper mapper = mock(ShadowInputMapper.class);
when(mapper.claimInstance("01", "INSTANCE-1", 120)).thenReturn(true); when(mapper.claimInstance("01", "INSTANCE-1", 120)).thenReturn(true);
PublishedShadowSnapshot refreshedSnapshot = snapshot(43L, "01"); PublishedShadowSnapshot refreshedSnapshot = snapshot(43L, "01");
when(mapper.selectPublishedSnapshot("01")).thenReturn(descriptor(refreshedSnapshot)); when(mapper.selectPublishedSnapshot("01")).thenReturn(descriptor(refreshedSnapshot));
ShadowPlanningEngine engine = engine(mock(SchedulingFacade.class), mapper); ShadowPlanningEngine engine = new ShadowPlanningEngine(
mock(SchedulingFacade.class), mapper, properties, transactions);
PublishedShadowSnapshot refreshed = engine.refreshSnapshot(43L, "INSTANCE-1"); PublishedShadowSnapshot refreshed = engine.refreshSnapshot(43L, "INSTANCE-1");
verify(mapper).claimInstance("01", "INSTANCE-1", 120); verify(mapper).claimInstance("01", "INSTANCE-1", 120);
verify(mapper).refreshSnapshot("01", 43L, "INSTANCE-1"); verify(mapper).refreshSnapshot("01", 43L, "INSTANCE-1");
assertThat(refreshed).isEqualTo(refreshedSnapshot); assertThat(refreshed).isEqualTo(refreshedSnapshot);
assertThat(transactions.definition.getTimeout()).isEqualTo(11);
} }
@Test @Test

View File

@@ -76,6 +76,22 @@ class ShadowSafetyGuardTest {
.hasMessageContaining("between 100 and 10000"); .hasMessageContaining("between 100 and 10000");
} }
@Test
void rejectsRefreshTimeoutOutsideValidatedRange() {
ShadowObservationProperties shadow = enabledShadow();
AlgorithmProperties algorithm = new AlgorithmProperties();
algorithm.setMode(SchedulingMode.SHADOW);
algorithm.setDatabaseWritesEnabled(true);
shadow.setRefreshTimeoutSeconds(3);
assertThatThrownBy(() -> new ShadowSafetyGuard(shadow, algorithm).validate())
.hasMessageContaining("refresh timeout");
shadow.setRefreshTimeoutSeconds(61);
assertThatThrownBy(() -> new ShadowSafetyGuard(shadow, algorithm).validate())
.hasMessageContaining("refresh timeout");
}
private ShadowObservationProperties enabledShadow() { private ShadowObservationProperties enabledShadow() {
ShadowObservationProperties properties = new ShadowObservationProperties(); ShadowObservationProperties properties = new ShadowObservationProperties();
properties.setEnabled(true); properties.setEnabled(true);