diff --git a/ruoyi-wcs-shadow/src/main/java/com/ruoyi/shadow/config/ShadowKernelConfiguration.java b/ruoyi-wcs-shadow/src/main/java/com/ruoyi/shadow/config/ShadowKernelConfiguration.java index a2e7d71..7c9c44d 100644 --- a/ruoyi-wcs-shadow/src/main/java/com/ruoyi/shadow/config/ShadowKernelConfiguration.java +++ b/ruoyi-wcs-shadow/src/main/java/com/ruoyi/shadow/config/ShadowKernelConfiguration.java @@ -29,6 +29,7 @@ import com.ruoyi.wcs.domain.traffic.WaitForGraph; import com.ruoyi.wcs.port.out.DispatchPolicy; import com.ruoyi.wcs.port.out.TrafficStateRepository; import org.springframework.beans.factory.annotation.Qualifier; +import org.springframework.beans.factory.annotation.Value; import org.springframework.context.annotation.Bean; import org.springframework.context.annotation.Configuration; import org.springframework.jdbc.core.JdbcTemplate; @@ -184,8 +185,11 @@ public class ShadowKernelConfiguration { SqlServerShadowFactReader factReader, Ft546ShadowCycle shadowCycle, ShadowStartupReport shadowStartupReport, - @Qualifier("shadowClock") Clock clock) { + @Qualifier("shadowClock") Clock clock, + @Value("${shadow.cycle.deadline-ms:10000}") + long cycleDeadlineMs) { return new Ft546ShadowScheduler(factReader, shadowCycle, - shadowStartupReport, clock); + shadowStartupReport, clock, + Duration.ofMillis(cycleDeadlineMs)); } } diff --git a/ruoyi-wcs-shadow/src/main/java/com/ruoyi/shadow/facts/ShadowFactBatchValidator.java b/ruoyi-wcs-shadow/src/main/java/com/ruoyi/shadow/facts/ShadowFactBatchValidator.java index 40cb34c..f0aa8b9 100644 --- a/ruoyi-wcs-shadow/src/main/java/com/ruoyi/shadow/facts/ShadowFactBatchValidator.java +++ b/ruoyi-wcs-shadow/src/main/java/com/ruoyi/shadow/facts/ShadowFactBatchValidator.java @@ -8,6 +8,9 @@ import java.util.Map; import java.util.Set; public final class ShadowFactBatchValidator { + private static final Duration MAXIMUM_FUTURE_SKEW = + Duration.ofMillis(100); + private final Duration maximumCaptureSpan; private final Duration maximumAge; @@ -38,7 +41,9 @@ public final class ShadowFactBatchValidator { return ShadowFactValidation.insufficient( "DATABASE_CAPTURE_SPAN_EXCEEDED"); } - if (completed.isAfter(now)) { + if (completed.isAfter(now) + && Duration.between(now, completed) + .compareTo(MAXIMUM_FUTURE_SKEW) > 0) { return ShadowFactValidation.insufficient("DATABASE_CLOCK_IN_FUTURE"); } if (Duration.between(completed, now).compareTo(maximumAge) > 0) { @@ -53,6 +58,7 @@ public final class ShadowFactBatchValidator { Map vehicleHashes = new HashMap(); Map vehicles = new HashMap(); + // device_srgv.last_time is the last task completion time, not a heartbeat. for (ShadowVehicleFact vehicle : batch.getVehicles()) { if (!notBlank(vehicle.getCurrentSite())) { return ShadowFactValidation.insufficient( @@ -76,13 +82,6 @@ public final class ShadowFactBatchValidator { return ShadowFactValidation.insufficient( "CONTRADICTORY_VEHICLE_POSITION:" + vehicle.getDeviceId()); } - if (vehicle.getCapturedAt() == null - || vehicle.getCapturedAt().isAfter(now) - || Duration.between(vehicle.getCapturedAt(), now) - .compareTo(maximumAge) > 0) { - return ShadowFactValidation.insufficient( - "STALE_VEHICLE_FACT:" + vehicle.getDeviceId()); - } String existingHash = vehicleHashes.putIfAbsent( vehicle.getDeviceId(), vehicle.getContentHash()); if (existingHash != null diff --git a/ruoyi-wcs-shadow/src/main/java/com/ruoyi/shadow/facts/SqlServerShadowFactReader.java b/ruoyi-wcs-shadow/src/main/java/com/ruoyi/shadow/facts/SqlServerShadowFactReader.java index 0a4bab7..24b088c 100644 --- a/ruoyi-wcs-shadow/src/main/java/com/ruoyi/shadow/facts/SqlServerShadowFactReader.java +++ b/ruoyi-wcs-shadow/src/main/java/com/ruoyi/shadow/facts/SqlServerShadowFactReader.java @@ -154,7 +154,7 @@ public final class SqlServerShadowFactReader { private static String classify(DataAccessException failure) { String name = failure.getClass().getSimpleName().toUpperCase(); - if (name.contains("TIMEOUT") || name.contains("TRANSIENT")) { + if (name.contains("TIMEOUT")) { return "SOURCE_QUERY_TIMEOUT"; } return "SOURCE_UNAVAILABLE"; diff --git a/ruoyi-wcs-shadow/src/main/java/com/ruoyi/shadow/kernel/Ft546ShadowScheduler.java b/ruoyi-wcs-shadow/src/main/java/com/ruoyi/shadow/kernel/Ft546ShadowScheduler.java index 8d05c9c..bc36398 100644 --- a/ruoyi-wcs-shadow/src/main/java/com/ruoyi/shadow/kernel/Ft546ShadowScheduler.java +++ b/ruoyi-wcs-shadow/src/main/java/com/ruoyi/shadow/kernel/Ft546ShadowScheduler.java @@ -26,7 +26,8 @@ import java.util.concurrent.atomic.AtomicBoolean; public final class Ft546ShadowScheduler implements AutoCloseable { private static final Logger LOGGER = LoggerFactory.getLogger(Ft546ShadowScheduler.class); - private static final Duration CYCLE_DEADLINE = Duration.ofSeconds(4); + private static final Duration CYCLE_DEADLINE = Duration.ofSeconds(10); + private static final Duration SOURCE_READ_DEADLINE = Duration.ofSeconds(4); private static final Duration CRASH_BACKOFF = Duration.ofSeconds(1); private static final String FINISH_QUEUE_VIEW = "shadow_ro.v_wcs_srgv_queue_finish"; @@ -36,9 +37,11 @@ public final class Ft546ShadowScheduler implements AutoCloseable { private final ShadowStartupReport statusReport; private final Clock clock; private final Duration cycleDeadline; + private final Duration sourceReadDeadline; private final ExecutorService cycleExecutor; private final AtomicBoolean inFlight = new AtomicBoolean(); private final AtomicBoolean permanentlyStopped = new AtomicBoolean(); + private volatile CycleStage activeStage = CycleStage.IDLE; private volatile ShadowReadCursor cursor = ShadowReadCursor.initial(); private volatile Instant nextAttemptAt = Instant.EPOCH; @@ -54,12 +57,23 @@ public final class Ft546ShadowScheduler implements AutoCloseable { ShadowStartupReport statusReport, Clock clock, Duration cycleDeadline) { + this(factReader, shadowCycle, statusReport, clock, cycleDeadline, + shorter(cycleDeadline, SOURCE_READ_DEADLINE)); + } + + Ft546ShadowScheduler(SqlServerShadowFactReader factReader, + Ft546ShadowCycle shadowCycle, + ShadowStartupReport statusReport, + Clock clock, + Duration cycleDeadline, + Duration sourceReadDeadline) { this.factReader = Objects.requireNonNull(factReader, "factReader"); this.shadowCycle = Objects.requireNonNull(shadowCycle, "shadowCycle"); this.statusReport = Objects.requireNonNull(statusReport, "statusReport"); this.clock = Objects.requireNonNull(clock, "clock"); this.cycleDeadline = requirePositive(cycleDeadline); + this.sourceReadDeadline = requirePositive(sourceReadDeadline); this.cycleExecutor = Executors.newSingleThreadExecutor( shadowCycleThreadFactory()); } @@ -94,16 +108,21 @@ public final class Ft546ShadowScheduler implements AutoCloseable { nextAttemptAt = Instant.EPOCH; return true; } catch (TimeoutException timeout) { + CycleStage timedOutStage = activeStage; future.cancel(true); - return stopPermanently("SHADOW_CYCLE_TIMEOUT", timeout); + return stopPermanently("SHADOW_CYCLE_TIMEOUT", timeout, + timedOutStage); } catch (InterruptedException interrupted) { + CycleStage interruptedStage = activeStage; future.cancel(true); Thread.currentThread().interrupt(); - return stopPermanently("SHADOW_CYCLE_INTERRUPTED", interrupted); + return stopPermanently("SHADOW_CYCLE_INTERRUPTED", interrupted, + interruptedStage); } catch (ExecutionException wrapped) { return handleFailure(wrapped.getCause()); } finally { inFlight.set(false); + activeStage = CycleStage.IDLE; } } @@ -123,15 +142,23 @@ public final class Ft546ShadowScheduler implements AutoCloseable { } private CycleAttempt executeCycle(Instant startedAt) { + activeStage = CycleStage.SOURCE_READ; ShadowFactBatch batch = factReader.read(cursor, - startedAt.plus(cycleDeadline)); - return new CycleAttempt(batch, shadowCycle.run(batch, clock.instant())); + startedAt.plus(sourceReadDeadline)); + activeStage = CycleStage.SHADOW_CYCLE; + return new CycleAttempt(batch, + shadowCycle.run(batch, clock.instant())); } private boolean handleFailure(Throwable failure) { if (failure instanceof ShadowFactReadException) { ShadowFactReadException readFailure = (ShadowFactReadException) failure; + String timeoutReason = sourceTimeoutReason(readFailure); + if (timeoutReason != null) { + return stopPermanently(timeoutReason, readFailure, + CycleStage.SOURCE_READ); + } Instant failedAt = clock.instant(); statusReport.markDegraded(readFailure.getReasonCode(), failedAt); nextAttemptAt = failedAt.plus(readFailure.getRetryAfter()); @@ -157,15 +184,41 @@ public final class Ft546ShadowScheduler implements AutoCloseable { return false; } - private boolean stopPermanently(String reasonCode, Exception failure) { + private static String sourceTimeoutReason( + ShadowFactReadException failure) { + String reasonCode = failure.getReasonCode(); + if ("SOURCE_QUERY_TIMEOUT".equals(reasonCode) + || "SOURCE_BATCH_DEADLINE_EXCEEDED".equals(reasonCode)) { + return reasonCode; + } + Throwable cause = failure.getCause(); + while (cause != null) { + String type = cause.getClass().getSimpleName() + .toUpperCase(java.util.Locale.ROOT); + if (type.contains("TIMEOUT")) { + return "SOURCE_QUERY_TIMEOUT"; + } + cause = cause.getCause(); + } + return null; + } + + private boolean stopPermanently(String reasonCode, Exception failure, + CycleStage stage) { permanentlyStopped.set(true); nextAttemptAt = Instant.EPOCH; statusReport.markPermanentlyDegraded(reasonCode, clock.instant()); - LOGGER.error("FT546 shadow collection stopped: {}", reasonCode, - failure); + LOGGER.error("FT546 shadow collection stopped: {}; stage={}", + reasonCode, stage, failure); return false; } + private static Duration shorter(Duration left, Duration right) { + Duration first = requirePositive(left); + Duration second = requirePositive(right); + return first.compareTo(second) <= 0 ? first : second; + } + private static Duration requirePositive(Duration value) { Duration duration = Objects.requireNonNull(value, "cycleDeadline"); if (duration.isZero() || duration.isNegative() @@ -184,6 +237,12 @@ public final class Ft546ShadowScheduler implements AutoCloseable { }; } + private enum CycleStage { + IDLE, + SOURCE_READ, + SHADOW_CYCLE + } + private static ShadowReadCursor advanceFinishCursor( ShadowReadCursor current, ShadowFactBatch batch) { Instant latestDate = current.getLastCreateDate(); diff --git a/ruoyi-wcs-shadow/src/main/resources/application-shadow.yml b/ruoyi-wcs-shadow/src/main/resources/application-shadow.yml index 49f0f0c..2185f9a 100644 --- a/ruoyi-wcs-shadow/src/main/resources/application-shadow.yml +++ b/ruoyi-wcs-shadow/src/main/resources/application-shadow.yml @@ -6,6 +6,8 @@ spring: banner-mode: "off" shadow: enabled: false + cycle: + deadline-ms: 10000 poll: initial-delay-ms: 5000 fixed-delay-ms: 5000 diff --git a/ruoyi-wcs-shadow/src/test/java/com/ruoyi/shadow/Ft546ShadowFaultIsolationTest.java b/ruoyi-wcs-shadow/src/test/java/com/ruoyi/shadow/Ft546ShadowFaultIsolationTest.java index 9541997..586bedc 100644 --- a/ruoyi-wcs-shadow/src/test/java/com/ruoyi/shadow/Ft546ShadowFaultIsolationTest.java +++ b/ruoyi-wcs-shadow/src/test/java/com/ruoyi/shadow/Ft546ShadowFaultIsolationTest.java @@ -1,5 +1,8 @@ package com.ruoyi.shadow; +import ch.qos.logback.classic.Logger; +import ch.qos.logback.classic.spi.ILoggingEvent; +import ch.qos.logback.core.read.ListAppender; import com.fasterxml.jackson.databind.ObjectMapper; import com.ruoyi.ft546.protocol.Ft546DeviceProfile; import com.ruoyi.ft546.protocol.Ft546SrgvCommandEncoder; @@ -11,6 +14,7 @@ import com.ruoyi.shadow.evidence.ShadowEvidenceException; import com.ruoyi.shadow.evidence.ShadowEvidenceRepository; import com.ruoyi.shadow.evidence.ShadowEvidenceStorageHealth; import com.ruoyi.shadow.facts.ShadowFactBatchValidator; +import com.ruoyi.shadow.facts.ShadowFactReadException; import com.ruoyi.shadow.facts.ShadowMapFact; import com.ruoyi.shadow.facts.ShadowQueueFact; import com.ruoyi.shadow.facts.ShadowReadCursor; @@ -28,10 +32,13 @@ import org.springframework.core.io.ClassPathResource; import org.springframework.core.env.EnumerablePropertySource; import org.springframework.core.env.PropertySource; import org.springframework.boot.env.YamlPropertySourceLoader; +import org.springframework.dao.DataAccessResourceFailureException; import org.springframework.dao.QueryTimeoutException; +import org.springframework.dao.TransientDataAccessResourceException; import org.springframework.jdbc.datasource.DataSourceTransactionManager; import org.springframework.jdbc.datasource.DriverManagerDataSource; import org.springframework.transaction.support.TransactionTemplate; +import org.slf4j.LoggerFactory; import java.time.Clock; import java.time.Duration; @@ -51,6 +58,7 @@ import java.util.UUID; import java.util.concurrent.CountDownLatch; import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicInteger; +import java.util.concurrent.atomic.AtomicReference; import java.util.function.Supplier; import static org.junit.jupiter.api.Assertions.assertEquals; @@ -61,7 +69,7 @@ class Ft546ShadowFaultIsolationTest { private static final Instant NOW = Instant.parse("2026-08-14T06:00:00Z"); @Test - void sourceTimeoutNeverEntersCycleOrRecordsACommand() { + void sourceQueryTimeoutPermanentlyStopsCollectionBeforeCycle() { MutableClock clock = new MutableClock(NOW); AtomicInteger sourceCalls = new AtomicInteger(); SqlServerShadowFactReader reader = reader(timeoutSource(sourceCalls), clock); @@ -72,15 +80,172 @@ class Ft546ShadowFaultIsolationTest { evidenceCycle(report, evidence, gateway, () -> ShadowEvidenceStorageHealth.Check.available( Long.MAX_VALUE)), report, clock); + Logger schedulerLogger = (Logger) LoggerFactory.getLogger( + Ft546ShadowScheduler.class); + ListAppender logAppender = + new ListAppender(); + logAppender.start(); + schedulerLogger.addAppender(logAppender); - assertFalse(scheduler.runOnce()); + try { + assertFalse(scheduler.runOnce()); + clock.advance(Duration.ofMinutes(1)); - assertEquals(1, sourceCalls.get()); - assertEquals(0, evidence.transactionCalls.get()); - assertEquals(Collections.emptyList(), gateway.getRecorded()); - assertEquals(ShadowStartupReport.Status.DEGRADED, report.getStatus()); - assertEquals("SOURCE_QUERY_TIMEOUT", report.getReasonCode()); - assertEquals(Duration.ofSeconds(1), scheduler.getRemainingBackoff()); + assertFalse(scheduler.runOnce()); + assertEquals(1, sourceCalls.get()); + assertEquals(0, evidence.transactionCalls.get()); + assertEquals(Collections.emptyList(), gateway.getRecorded()); + assertEquals(ShadowStartupReport.Status.DEGRADED, + report.getStatus()); + assertEquals("SOURCE_QUERY_TIMEOUT", report.getReasonCode()); + assertTrue(scheduler.isPermanentlyStopped()); + assertTrue(logAppender.list.stream().anyMatch(event -> + event.getFormattedMessage() + .contains("stage=SOURCE_READ")), + "source timeout log must identify the blocked stage"); + } finally { + schedulerLogger.detachAppender(logAppender); + logAppender.stop(); + scheduler.close(); + } + } + + @Test + void sourceBatchDeadlinePermanentlyStopsCollection() { + MutableClock clock = new MutableClock(NOW); + AtomicInteger sourceCalls = new AtomicInteger(); + SqlServerShadowFactReader.FactSource source = new FactSourceAdapter() { + @Override + public Instant readDatabaseClock(Instant deadline) { + sourceCalls.incrementAndGet(); + throw new ShadowFactReadException( + "SOURCE_BATCH_DEADLINE_EXCEEDED", Duration.ZERO); + } + }; + ShadowStartupReport report = report(); + Ft546ShadowScheduler scheduler = new Ft546ShadowScheduler( + reader(source, clock), + evidenceCycle(report, new RecordingEvidenceRepository(), + gateway(), + () -> ShadowEvidenceStorageHealth.Check.available( + Long.MAX_VALUE)), + report, clock); + + try { + assertFalse(scheduler.runOnce()); + clock.advance(Duration.ofMinutes(1)); + + assertFalse(scheduler.runOnce()); + assertEquals(1, sourceCalls.get()); + assertEquals("SOURCE_BATCH_DEADLINE_EXCEEDED", + report.getReasonCode()); + assertTrue(scheduler.isPermanentlyStopped()); + } finally { + scheduler.close(); + } + } + + @Test + void wrappedSourceQueryTimeoutPermanentlyStopsCollection() { + MutableClock clock = new MutableClock(NOW); + AtomicInteger sourceCalls = new AtomicInteger(); + SqlServerShadowFactReader.FactSource source = new FactSourceAdapter() { + @Override + public Instant readDatabaseClock(Instant deadline) { + sourceCalls.incrementAndGet(); + throw new DataAccessResourceFailureException( + "wrapped source timeout", + new QueryTimeoutException("query timed out")); + } + }; + ShadowStartupReport report = report(); + Ft546ShadowScheduler scheduler = new Ft546ShadowScheduler( + reader(source, clock), + evidenceCycle(report, new RecordingEvidenceRepository(), + gateway(), + () -> ShadowEvidenceStorageHealth.Check.available( + Long.MAX_VALUE)), + report, clock); + + try { + assertFalse(scheduler.runOnce()); + clock.advance(Duration.ofMinutes(1)); + + assertFalse(scheduler.runOnce()); + assertEquals(1, sourceCalls.get()); + assertEquals("SOURCE_QUERY_TIMEOUT", report.getReasonCode()); + assertTrue(scheduler.isPermanentlyStopped()); + } finally { + scheduler.close(); + } + } + + @Test + void nonTimeoutSourceFailureStillUsesBackoff() { + MutableClock clock = new MutableClock(NOW); + AtomicInteger sourceCalls = new AtomicInteger(); + SqlServerShadowFactReader.FactSource source = new FactSourceAdapter() { + @Override + public Instant readDatabaseClock(Instant deadline) { + sourceCalls.incrementAndGet(); + throw new DataAccessResourceFailureException( + "simulated source outage"); + } + }; + ShadowStartupReport report = report(); + Ft546ShadowScheduler scheduler = new Ft546ShadowScheduler( + reader(source, clock), + evidenceCycle(report, new RecordingEvidenceRepository(), + gateway(), + () -> ShadowEvidenceStorageHealth.Check.available( + Long.MAX_VALUE)), + report, clock); + + try { + assertFalse(scheduler.runOnce()); + + assertEquals(1, sourceCalls.get()); + assertEquals("SOURCE_UNAVAILABLE", report.getReasonCode()); + assertFalse(scheduler.isPermanentlyStopped()); + assertEquals(Duration.ofSeconds(1), + scheduler.getRemainingBackoff()); + } finally { + scheduler.close(); + } + } + + @Test + void transientSourceFailureStillUsesBackoff() { + MutableClock clock = new MutableClock(NOW); + AtomicInteger sourceCalls = new AtomicInteger(); + SqlServerShadowFactReader.FactSource source = new FactSourceAdapter() { + @Override + public Instant readDatabaseClock(Instant deadline) { + sourceCalls.incrementAndGet(); + throw new TransientDataAccessResourceException( + "simulated transient source failure"); + } + }; + ShadowStartupReport report = report(); + Ft546ShadowScheduler scheduler = new Ft546ShadowScheduler( + reader(source, clock), + evidenceCycle(report, new RecordingEvidenceRepository(), + gateway(), + () -> ShadowEvidenceStorageHealth.Check.available( + Long.MAX_VALUE)), + report, clock); + + try { + assertFalse(scheduler.runOnce()); + + assertEquals(1, sourceCalls.get()); + assertEquals("SOURCE_UNAVAILABLE", report.getReasonCode()); + assertFalse(scheduler.isPermanentlyStopped()); + assertEquals(Duration.ofSeconds(1), + scheduler.getRemainingBackoff()); + } finally { + scheduler.close(); + } } @Test @@ -174,6 +339,12 @@ class Ft546ShadowFaultIsolationTest { }, "shadow-cycle-test-release"); delayedRelease.setDaemon(true); delayedRelease.start(); + Logger schedulerLogger = (Logger) LoggerFactory.getLogger( + Ft546ShadowScheduler.class); + ListAppender logAppender = + new ListAppender(); + logAppender.start(); + schedulerLogger.addAppender(logAppender); long startedAt = System.nanoTime(); try { @@ -189,6 +360,10 @@ class Ft546ShadowFaultIsolationTest { assertEquals(ShadowStartupReport.Status.DEGRADED, report.getStatus()); assertEquals("SHADOW_CYCLE_TIMEOUT", report.getReasonCode()); + assertTrue(logAppender.list.stream().anyMatch(event -> + event.getFormattedMessage() + .contains("stage=SHADOW_CYCLE")), + "timeout log must identify the blocked stage"); assertFalse(scheduler.runOnce()); assertEquals(callsAfterTimeout, sourceCalls.get()); assertEquals(Collections.emptyList(), gateway.getRecorded()); @@ -198,6 +373,8 @@ class Ft546ShadowFaultIsolationTest { report.getStatus()); assertEquals("SHADOW_CYCLE_TIMEOUT", report.getReasonCode()); } finally { + schedulerLogger.detachAppender(logAppender); + logAppender.stop(); releaseCycle.countDown(); scheduler.close(); } @@ -219,10 +396,40 @@ class Ft546ShadowFaultIsolationTest { assertTrue(configuredKeys.contains("shadow.source.jdbc-url")); assertTrue(configuredKeys.contains("shadow.evidence.jdbc-url")); + assertTrue(configuredKeys.contains("shadow.cycle.deadline-ms")); assertFalse(configuredKeys.stream().anyMatch( Ft546ShadowFaultIsolationTest::isOutboundEndpointKey)); } + @Test + void longCycleDeadlineKeepsSourceReadDeadlineCappedAtFourSeconds() { + MutableClock clock = new MutableClock(NOW); + AtomicReference observedDeadline = + new AtomicReference(); + SqlServerShadowFactReader.FactSource source = new FactSourceAdapter() { + @Override + public Instant readDatabaseClock(Instant deadline) { + observedDeadline.set(deadline); + return NOW; + } + }; + RecordingEvidenceRepository evidence = new RecordingEvidenceRepository(); + ShadowStartupReport report = report(); + Ft546ShadowScheduler scheduler = new Ft546ShadowScheduler( + reader(source, clock), + evidenceCycle(report, evidence, gateway(), + () -> ShadowEvidenceStorageHealth.Check.available( + Long.MAX_VALUE)), + report, clock, Duration.ofSeconds(10)); + + try { + assertTrue(scheduler.runOnce()); + assertEquals(NOW.plusSeconds(4), observedDeadline.get()); + } finally { + scheduler.close(); + } + } + @Test void activeQueueCannotAdvanceTheFinishHistoryCursor() { MutableClock clock = new MutableClock(NOW); diff --git a/ruoyi-wcs-shadow/src/test/java/com/ruoyi/shadow/facts/ShadowFactBatchValidatorTest.java b/ruoyi-wcs-shadow/src/test/java/com/ruoyi/shadow/facts/ShadowFactBatchValidatorTest.java index 0fae7f3..e14bd42 100644 --- a/ruoyi-wcs-shadow/src/test/java/com/ruoyi/shadow/facts/ShadowFactBatchValidatorTest.java +++ b/ruoyi-wcs-shadow/src/test/java/com/ruoyi/shadow/facts/ShadowFactBatchValidatorTest.java @@ -28,6 +28,33 @@ class ShadowFactBatchValidatorTest { assertEquals("FACTS_COMPLETE", result.getReasonCode()); } + @Test + void databaseClockOneMillisecondAheadIsComparable() { + ShadowFactValidation result = validator.validate(batch( + Collections.singletonList(queue("Q-1", "070501", "070502", "1")), + Collections.singletonList(vehicle("1", "070501")), + Arrays.asList(map("070501"), map("070502")), + NOW.minusSeconds(1), NOW.plusMillis(1)), NOW); + + assertEquals(ShadowFactValidation.Status.COMPARABLE, + result.getStatus()); + } + + @Test + void oldLastTaskTimeDoesNotMakeVehicleFactStale() { + ShadowVehicleFact idleVehicle = completeVehicleBuilder("1", "070501") + .capturedAt(NOW.minus(Duration.ofHours(2))) + .build(); + ShadowFactValidation result = validator.validate(batch( + Collections.singletonList(queue("Q-1", "070501", "070502", "1")), + Collections.singletonList(idleVehicle), + Arrays.asList(map("070501"), map("070502")), + NOW.minusSeconds(1), NOW.minusMillis(500)), NOW); + + assertEquals(ShadowFactValidation.Status.COMPARABLE, + result.getStatus()); + } + @Test void contradictoryVehiclePositionsAreInsufficient() { ShadowFactValidation result = validator.validate(batch(