diff --git a/ruoyi-wcs-shadow/src/main/java/com/ruoyi/shadow/config/ShadowRuntimeConfiguration.java b/ruoyi-wcs-shadow/src/main/java/com/ruoyi/shadow/config/ShadowRuntimeConfiguration.java index 4d346bf..981c574 100644 --- a/ruoyi-wcs-shadow/src/main/java/com/ruoyi/shadow/config/ShadowRuntimeConfiguration.java +++ b/ruoyi-wcs-shadow/src/main/java/com/ruoyi/shadow/config/ShadowRuntimeConfiguration.java @@ -6,6 +6,7 @@ import com.ruoyi.shadow.startup.ShadowDatabaseGate; import com.ruoyi.shadow.startup.SqlServerPermissionProbe; import com.ruoyi.shadow.startup.SqlServerShadowDatabaseGate; import com.ruoyi.shadow.web.ShadowStatusController; +import com.ruoyi.shadow.facts.SqlServerShadowFactReader; import org.springframework.beans.factory.annotation.Qualifier; import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingBean; import org.springframework.boot.context.properties.EnableConfigurationProperties; @@ -14,8 +15,10 @@ import org.springframework.context.annotation.Bean; import org.springframework.context.annotation.Configuration; import org.springframework.core.env.Environment; import org.springframework.jdbc.core.JdbcTemplate; +import org.springframework.transaction.support.TransactionTemplate; import javax.sql.DataSource; +import java.time.Clock; @Configuration(proxyBeanMethods = false) @EnableConfigurationProperties(ShadowProperties.class) @@ -54,4 +57,17 @@ public class ShadowRuntimeConfiguration { ShadowStartupReport report) { return new ShadowStatusController(report); } + + @Bean + public SqlServerShadowFactReader shadowFactReader( + @Qualifier("shadowSourceJdbc") JdbcTemplate sourceJdbc, + @Qualifier("shadowSourceTx") + org.springframework.transaction.PlatformTransactionManager sourceTx) { + Clock clock = Clock.systemUTC(); + TransactionTemplate transaction = new TransactionTemplate(sourceTx); + transaction.setReadOnly(true); + return new SqlServerShadowFactReader( + new SqlServerShadowFactReader.JdbcFactSource(sourceJdbc, clock), + transaction, clock); + } } diff --git a/ruoyi-wcs-shadow/src/main/java/com/ruoyi/shadow/facts/FactHash.java b/ruoyi-wcs-shadow/src/main/java/com/ruoyi/shadow/facts/FactHash.java new file mode 100644 index 0000000..d29f4c4 --- /dev/null +++ b/ruoyi-wcs-shadow/src/main/java/com/ruoyi/shadow/facts/FactHash.java @@ -0,0 +1,41 @@ +package com.ruoyi.shadow.facts; + +import java.nio.charset.StandardCharsets; +import java.security.MessageDigest; +import java.security.NoSuchAlgorithmException; + +final class FactHash { + private FactHash() { + } + + static String sha256(Object... values) { + try { + MessageDigest digest = MessageDigest.getInstance("SHA-256"); + for (Object value : values) { + if (value == null) { + digest.update((byte) 0); + continue; + } + digest.update((byte) 1); + byte[] bytes = value.toString().getBytes(StandardCharsets.UTF_8); + digest.update((byte) (bytes.length >>> 24)); + digest.update((byte) (bytes.length >>> 16)); + digest.update((byte) (bytes.length >>> 8)); + digest.update((byte) bytes.length); + digest.update(bytes); + } + return hex(digest.digest()); + } catch (NoSuchAlgorithmException impossible) { + throw new IllegalStateException("SHA-256 is unavailable", impossible); + } + } + + private static String hex(byte[] bytes) { + StringBuilder result = new StringBuilder(bytes.length * 2); + for (byte value : bytes) { + result.append(Character.forDigit((value >>> 4) & 0xf, 16)); + result.append(Character.forDigit(value & 0xf, 16)); + } + return result.toString(); + } +} diff --git a/ruoyi-wcs-shadow/src/main/java/com/ruoyi/shadow/facts/LegacyQueueProtocolProjector.java b/ruoyi-wcs-shadow/src/main/java/com/ruoyi/shadow/facts/LegacyQueueProtocolProjector.java new file mode 100644 index 0000000..3a5a401 --- /dev/null +++ b/ruoyi-wcs-shadow/src/main/java/com/ruoyi/shadow/facts/LegacyQueueProtocolProjector.java @@ -0,0 +1,202 @@ +package com.ruoyi.shadow.facts; + +import java.util.ArrayList; +import java.util.Arrays; +import java.util.Collections; +import java.util.List; + +public final class LegacyQueueProtocolProjector { + public ShadowLegacyQueueResult project(ShadowQueueFact queue) { + if (queue == null) { + return unavailable("MISSING_QUEUE"); + } + if (!notBlank(queue.getTaskNo())) { + return unavailable("MISSING_TASK_NO"); + } + if (!notBlank(queue.getDefault02())) { + return unavailable("MISSING_DEFAULT02"); + } + if (!notBlank(queue.getAgvId())) { + return unavailable("MISSING_AGV_ID"); + } + if (!notBlank(queue.getPathInfo())) { + return unavailable("MISSING_PATH_INFO"); + } + Integer taskNo = integer(queue.getTaskNo()); + Integer taskType = integer(queue.getDefault02()); + if (taskNo == null) { + return unavailable("INVALID_TASK_NO"); + } + if (taskType == null) { + return unavailable("INVALID_DEFAULT02"); + } + String stage = trim(queue.getDefault08()); + if (stage.isEmpty()) { + return main(queue, taskNo, taskType); + } + if ("1".equals(stage)) { + return stageOne(queue, taskNo, taskType); + } + if ("2".equals(stage)) { + return stageTwo(queue, taskNo); + } + return unavailable("INVALID_DEFAULT08"); + } + + private static ShadowLegacyQueueResult main(ShadowQueueFact queue, + int taskNo, int taskType) { + List path = ordinaryPath(queue); + if (path == null) { + return unavailable("INVALID_MAIN_PATH"); + } + List writes = + new ArrayList(); + add(writes, "4004", ShadowLegacyQueueResult.DataType.INT32, false, path); + add(writes, "4000", ShadowLegacyQueueResult.DataType.INT32, false, + Collections.singletonList(taskNo)); + add(writes, "4002", ShadowLegacyQueueResult.DataType.INT16, false, + Collections.singletonList(taskType)); + add(writes, "4021", ShadowLegacyQueueResult.DataType.INT16, true, + Collections.singletonList(1)); + if (taskType == 8) { + Integer target = integer(queue.getDefault07()); + if (target == null) { + return unavailable("MISSING_DEFAULT07"); + } + add(writes, "4020", ShadowLegacyQueueResult.DataType.INT16, false, + Collections.singletonList(1)); + add(writes, "4022", ShadowLegacyQueueResult.DataType.INT16, false, + Collections.singletonList(target)); + } + return ShadowLegacyQueueResult.available(writes); + } + + private static ShadowLegacyQueueResult stageOne(ShadowQueueFact queue, + int taskNo, int taskType) { + List path = ordinaryPath(queue); + if (path == null) { + return unavailable("INVALID_STAGE_ONE_PATH"); + } + List writes = + new ArrayList(); + add(writes, "4404", ShadowLegacyQueueResult.DataType.INT32, false, path); + add(writes, "4422", ShadowLegacyQueueResult.DataType.INT16, true, + Collections.singletonList(1)); + add(writes, "4402", ShadowLegacyQueueResult.DataType.INT16, false, + Collections.singletonList(taskType)); + add(writes, "4400", ShadowLegacyQueueResult.DataType.INT32, false, + Collections.singletonList(taskNo)); + return ShadowLegacyQueueResult.available(writes); + } + + private static ShadowLegacyQueueResult stageTwo(ShadowQueueFact queue, + int taskNo) { + Integer start = site(queue.getEndSite()); + Integer end = site(queue.getDefault06()); + if (start == null) { + return unavailable("MISSING_STAGE_TWO_START"); + } + if (end == null) { + return unavailable("MISSING_DEFAULT06"); + } + List path = new ArrayList(Arrays.asList(start, end)); + while (path.size() < 8) { + path.add(0); + } + List writes = + new ArrayList(); + add(writes, "4404", ShadowLegacyQueueResult.DataType.INT32, false, path); + add(writes, "4422", ShadowLegacyQueueResult.DataType.INT16, true, + Collections.singletonList(2)); + add(writes, "4423", ShadowLegacyQueueResult.DataType.INT16, false, + Collections.singletonList(99)); + add(writes, "4402", ShadowLegacyQueueResult.DataType.INT16, false, + Collections.singletonList(9)); + add(writes, "4400", ShadowLegacyQueueResult.DataType.INT32, false, + Collections.singletonList(taskNo)); + return ShadowLegacyQueueResult.available(writes); + } + + private static List ordinaryPath(ShadowQueueFact queue) { + List executedPath = executedPath(queue.getPathInfo()); + if (executedPath == null || executedPath.size() < 2) { + return null; + } + List result = new ArrayList(); + result.add(executedPath.get(0)); + result.add(executedPath.get(executedPath.size() - 1)); + if (notBlank(queue.getTurnPath())) { + for (String token : queue.getTurnPath().split(",")) { + if (!notBlank(token)) { + continue; + } + Integer point = integer(token); + if (point != null && point != 0) { + point = site(token); + } + if (point == null || result.size() == 8) { + return null; + } + result.add(point); + } + } + while (result.size() < 8) { + result.add(0); + } + return result; + } + + private static List executedPath(String value) { + if (!notBlank(value)) { + return null; + } + List result = new ArrayList(); + for (String token : value.split(",")) { + if (!notBlank(token)) { + continue; + } + Integer point = site(token); + if (point == null) { + return null; + } + result.add(point); + } + return result; + } + + private static void add( + List writes, + String address, ShadowLegacyQueueResult.DataType type, + boolean trigger, List values) { + writes.add(new ShadowLegacyQueueResult.LegacyRegisterWrite(address, + type, values, writes.size() + 1, trigger)); + } + + private static ShadowLegacyQueueResult unavailable(String reason) { + return ShadowLegacyQueueResult.unavailable(reason); + } + + private static Integer site(String value) { + Integer parsed = integer(value); + return parsed == null || parsed <= 0 || parsed > 999_999 ? null : parsed; + } + + private static Integer integer(String value) { + if (!notBlank(value)) { + return null; + } + try { + return Integer.valueOf(value.trim()); + } catch (NumberFormatException invalid) { + return null; + } + } + + private static String trim(String value) { + return value == null ? "" : value.trim(); + } + + private static boolean notBlank(String value) { + return value != null && !value.trim().isEmpty(); + } +} diff --git a/ruoyi-wcs-shadow/src/main/java/com/ruoyi/shadow/facts/ShadowFactBatch.java b/ruoyi-wcs-shadow/src/main/java/com/ruoyi/shadow/facts/ShadowFactBatch.java new file mode 100644 index 0000000..1961fb9 --- /dev/null +++ b/ruoyi-wcs-shadow/src/main/java/com/ruoyi/shadow/facts/ShadowFactBatch.java @@ -0,0 +1,120 @@ +package com.ruoyi.shadow.facts; + +import java.time.Instant; +import java.util.ArrayList; +import java.util.Collections; +import java.util.Comparator; +import java.util.LinkedHashMap; +import java.util.List; +import java.util.Map; + +public final class ShadowFactBatch { + private final Instant databaseCaptureStartedAt; + private final Instant databaseCapturedAt; + private final List sourceQueues; + private final List queues; + private final List vehicles; + private final List maps; + + private ShadowFactBatch(Instant databaseCaptureStartedAt, + Instant databaseCapturedAt, + List sourceQueues, + List vehicles, + List maps) { + this.databaseCaptureStartedAt = databaseCaptureStartedAt; + this.databaseCapturedAt = databaseCapturedAt; + this.sourceQueues = immutable(sourceQueues); + this.queues = immutable(deduplicateAndSort(sourceQueues)); + this.vehicles = immutable(sortVehicles(vehicles)); + this.maps = immutable(sortMaps(maps)); + } + + public static ShadowFactBatch create(Instant startedAt, Instant capturedAt, + List queues, + List vehicles, + List maps) { + return new ShadowFactBatch(startedAt, capturedAt, + requiredList(queues, "queues"), + requiredList(vehicles, "vehicles"), + requiredList(maps, "maps")); + } + + public Instant getDatabaseCaptureStartedAt() { return databaseCaptureStartedAt; } + public Instant getDatabaseCapturedAt() { return databaseCapturedAt; } + public List getSourceQueues() { return sourceQueues; } + public List getQueues() { return queues; } + public List getVehicles() { return vehicles; } + public List getMaps() { return maps; } + + public ShadowQueueFact queue(String guid) { + for (ShadowQueueFact fact : queues) { + if (fact.getGuid().equals(guid)) { + return fact; + } + } + return null; + } + + public ShadowVehicleFact vehicle(String deviceId) { + for (ShadowVehicleFact fact : vehicles) { + if (fact.getDeviceId().equals(deviceId)) { + return fact; + } + } + return null; + } + + private static List deduplicateAndSort( + List source) { + Map unique = + new LinkedHashMap(); + for (ShadowQueueFact fact : source) { + String key = fact.getTableName() + '\u0000' + fact.getGuid() + + '\u0000' + fact.getContentHash(); + unique.put(key, fact); + } + List result = + new ArrayList(unique.values()); + Collections.sort(result, Comparator + .comparing(ShadowQueueFact::getCreatedAt, + Comparator.nullsFirst(Comparator.naturalOrder())) + .thenComparing(ShadowQueueFact::getGuid) + .thenComparing(ShadowQueueFact::getContentHash)); + return result; + } + + private static List sortVehicles( + List source) { + List result = + new ArrayList(source); + Collections.sort(result, Comparator + .comparing(ShadowVehicleFact::getDeviceId) + .thenComparing(ShadowVehicleFact::getGuid) + .thenComparing(ShadowVehicleFact::getContentHash)); + return result; + } + + private static List sortMaps(List source) { + List result = new ArrayList(source); + Collections.sort(result, Comparator + .comparing(ShadowMapFact::getSiteId) + .thenComparing(ShadowMapFact::getGuid)); + return result; + } + + private static List immutable(List source) { + return Collections.unmodifiableList(new ArrayList(source)); + } + + private static List requiredList(List source, String name) { + if (source == null) { + throw new IllegalArgumentException(name + " must not be null"); + } + for (T value : source) { + if (value == null) { + throw new IllegalArgumentException(name + " must not contain null"); + } + } + return source; + } +} 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 new file mode 100644 index 0000000..40cb34c --- /dev/null +++ b/ruoyi-wcs-shadow/src/main/java/com/ruoyi/shadow/facts/ShadowFactBatchValidator.java @@ -0,0 +1,183 @@ +package com.ruoyi.shadow.facts; + +import java.time.Duration; +import java.time.Instant; +import java.util.HashMap; +import java.util.HashSet; +import java.util.Map; +import java.util.Set; + +public final class ShadowFactBatchValidator { + private final Duration maximumCaptureSpan; + private final Duration maximumAge; + + public ShadowFactBatchValidator(Duration maximumCaptureSpan, + Duration maximumAge) { + if (maximumCaptureSpan == null || maximumCaptureSpan.isNegative() + || maximumAge == null || maximumAge.isNegative()) { + throw new IllegalArgumentException("durations must not be negative"); + } + this.maximumCaptureSpan = maximumCaptureSpan; + this.maximumAge = maximumAge; + } + + public ShadowFactValidation validate(ShadowFactBatch batch, Instant now) { + if (batch == null || now == null) { + return ShadowFactValidation.insufficient("MISSING_BATCH_CLOCK"); + } + Instant started = batch.getDatabaseCaptureStartedAt(); + Instant completed = batch.getDatabaseCapturedAt(); + if (started == null || completed == null) { + return ShadowFactValidation.insufficient("MISSING_DATABASE_CLOCK"); + } + if (completed.isBefore(started)) { + return ShadowFactValidation.insufficient( + "NON_MONOTONIC_DATABASE_CLOCK"); + } + if (Duration.between(started, completed).compareTo(maximumCaptureSpan) > 0) { + return ShadowFactValidation.insufficient( + "DATABASE_CAPTURE_SPAN_EXCEEDED"); + } + if (completed.isAfter(now)) { + return ShadowFactValidation.insufficient("DATABASE_CLOCK_IN_FUTURE"); + } + if (Duration.between(completed, now).compareTo(maximumAge) > 0) { + return ShadowFactValidation.insufficient("STALE_DATABASE_CLOCK"); + } + + ShadowFactValidation mutation = queueMutations(batch); + if (mutation != null) { + return mutation; + } + Map vehicleSites = new HashMap(); + Map vehicleHashes = new HashMap(); + Map vehicles = + new HashMap(); + for (ShadowVehicleFact vehicle : batch.getVehicles()) { + if (!notBlank(vehicle.getCurrentSite())) { + return ShadowFactValidation.insufficient( + "MISSING_VEHICLE_CURRENT_SITE:" + vehicle.getDeviceId()); + } + if (vehicle.getWorkMode() == null) { + return ShadowFactValidation.insufficient( + "MISSING_VEHICLE_WORK_MODE:" + vehicle.getDeviceId()); + } + if (vehicle.getBattery() == null) { + return ShadowFactValidation.insufficient( + "MISSING_VEHICLE_BATTERY:" + vehicle.getDeviceId()); + } + if (vehicle.getBattery() < 0 || vehicle.getBattery() > 100) { + return ShadowFactValidation.insufficient( + "INVALID_VEHICLE_BATTERY:" + vehicle.getDeviceId()); + } + String existing = vehicleSites.putIfAbsent(vehicle.getDeviceId(), + vehicle.getCurrentSite()); + if (existing != null && !same(existing, vehicle.getCurrentSite())) { + 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 + && !existingHash.equals(vehicle.getContentHash())) { + return ShadowFactValidation.insufficient( + "CONTRADICTORY_VEHICLE_FACT:" + vehicle.getDeviceId()); + } + vehicles.putIfAbsent(vehicle.getDeviceId(), vehicle); + } + if (vehicles.isEmpty()) { + return ShadowFactValidation.insufficient("MISSING_VEHICLE_FACTS"); + } + + Set sites = new HashSet(); + Map maps = + new HashMap(); + for (ShadowMapFact map : batch.getMaps()) { + ShadowMapFact existing = maps.putIfAbsent(map.getSiteId(), map); + if (existing != null && !same(existing, map)) { + return ShadowFactValidation.insufficient( + "CONTRADICTORY_MAP_SITE:" + map.getSiteId()); + } + sites.add(map.getSiteId()); + } + if (maps.isEmpty()) { + return ShadowFactValidation.insufficient("MISSING_MAP_FACTS"); + } + for (ShadowQueueFact queue : batch.getQueues()) { + if (queue.getCreatedAt() == null) { + return ShadowFactValidation.insufficient( + "MISSING_QUEUE_CREATED_AT:" + queue.getTableName() + + ':' + queue.getGuid()); + } + String missing = firstMissingSite(sites, queue.getStartSite(), + queue.getEndSite(), queue.getCargoPickSite(), + queue.getCargoPlaceSite()); + if (missing != null) { + return ShadowFactValidation.insufficient("MISSING_MAP_SITE:" + missing); + } + if (notBlank(queue.getAgvId()) + && !vehicleSites.containsKey(queue.getAgvId())) { + return ShadowFactValidation.insufficient( + "MISSING_LEGACY_ASSIGNED_VEHICLE:" + queue.getAgvId()); + } + if ("2".equals(trimmed(queue.getDefault08())) + && notBlank(queue.getAgvId())) { + ShadowVehicleFact vehicle = vehicles.get(queue.getAgvId()); + if (vehicle != null && vehicle.getSpare5() == null) { + return ShadowFactValidation.insufficient( + "MISSING_STAGE_TWO_GATE:" + queue.getAgvId()); + } + } + } + return ShadowFactValidation.comparable(); + } + + private static ShadowFactValidation queueMutations(ShadowFactBatch batch) { + Map hashes = new HashMap(); + for (ShadowQueueFact queue : batch.getSourceQueues()) { + String key = queue.getTableName() + ':' + queue.getGuid(); + String existing = hashes.putIfAbsent(key, queue.getContentHash()); + if (existing != null && !existing.equals(queue.getContentHash())) { + return ShadowFactValidation.insufficient( + "QUEUE_MUTATED_DURING_OVERLAP:" + key); + } + } + return null; + } + + private static String firstMissingSite(Set sites, String... required) { + for (String site : required) { + if (notBlank(site) && !sites.contains(site.trim())) { + return site.trim(); + } + } + return null; + } + + private static boolean notBlank(String value) { + return value != null && !value.trim().isEmpty(); + } + + private static String trimmed(String value) { + return value == null ? null : value.trim(); + } + + private static boolean same(String left, String right) { + return left == null ? right == null : left.equals(right); + } + + private static boolean same(ShadowMapFact left, ShadowMapFact right) { + return same(left.getGuid(), right.getGuid()) + && left.getX() == right.getX() + && left.getY() == right.getY() + && left.getZ() == right.getZ() + && left.getValue() == right.getValue(); + } +} diff --git a/ruoyi-wcs-shadow/src/main/java/com/ruoyi/shadow/facts/ShadowFactReadException.java b/ruoyi-wcs-shadow/src/main/java/com/ruoyi/shadow/facts/ShadowFactReadException.java new file mode 100644 index 0000000..45c22da --- /dev/null +++ b/ruoyi-wcs-shadow/src/main/java/com/ruoyi/shadow/facts/ShadowFactReadException.java @@ -0,0 +1,25 @@ +package com.ruoyi.shadow.facts; + +import java.time.Duration; + +public final class ShadowFactReadException extends RuntimeException { + private final String reasonCode; + private final Duration retryAfter; + + public ShadowFactReadException(String reasonCode, Duration retryAfter) { + this(reasonCode, retryAfter, null); + } + + public ShadowFactReadException(String reasonCode, Duration retryAfter, + Throwable cause) { + super(reasonCode, cause); + if (reasonCode == null || reasonCode.trim().isEmpty()) { + throw new IllegalArgumentException("reasonCode must not be blank"); + } + this.reasonCode = reasonCode.trim(); + this.retryAfter = retryAfter == null ? Duration.ZERO : retryAfter; + } + + public String getReasonCode() { return reasonCode; } + public Duration getRetryAfter() { return retryAfter; } +} diff --git a/ruoyi-wcs-shadow/src/main/java/com/ruoyi/shadow/facts/ShadowFactValidation.java b/ruoyi-wcs-shadow/src/main/java/com/ruoyi/shadow/facts/ShadowFactValidation.java new file mode 100644 index 0000000..9c55a41 --- /dev/null +++ b/ruoyi-wcs-shadow/src/main/java/com/ruoyi/shadow/facts/ShadowFactValidation.java @@ -0,0 +1,24 @@ +package com.ruoyi.shadow.facts; + +public final class ShadowFactValidation { + public enum Status { COMPARABLE, INSUFFICIENT_FACTS } + + private final Status status; + private final String reasonCode; + + private ShadowFactValidation(Status status, String reasonCode) { + this.status = status; + this.reasonCode = reasonCode; + } + + public static ShadowFactValidation comparable() { + return new ShadowFactValidation(Status.COMPARABLE, "FACTS_COMPLETE"); + } + + public static ShadowFactValidation insufficient(String reasonCode) { + return new ShadowFactValidation(Status.INSUFFICIENT_FACTS, reasonCode); + } + + public Status getStatus() { return status; } + public String getReasonCode() { return reasonCode; } +} diff --git a/ruoyi-wcs-shadow/src/main/java/com/ruoyi/shadow/facts/ShadowLegacyQueueResult.java b/ruoyi-wcs-shadow/src/main/java/com/ruoyi/shadow/facts/ShadowLegacyQueueResult.java new file mode 100644 index 0000000..a9d3898 --- /dev/null +++ b/ruoyi-wcs-shadow/src/main/java/com/ruoyi/shadow/facts/ShadowLegacyQueueResult.java @@ -0,0 +1,83 @@ +package com.ruoyi.shadow.facts; + +import java.util.ArrayList; +import java.util.Collections; +import java.util.List; + +public final class ShadowLegacyQueueResult { + public enum Status { AVAILABLE, LEGACY_RESULT_UNAVAILABLE } + public enum DataType { INT16, INT32 } + + private final Status status; + private final String reasonCode; + private final List writes; + + private ShadowLegacyQueueResult(Status status, String reasonCode, + List writes) { + this.status = status; + this.reasonCode = reasonCode; + this.writes = Collections.unmodifiableList( + new ArrayList(writes)); + } + + public static ShadowLegacyQueueResult available( + List writes) { + return new ShadowLegacyQueueResult(Status.AVAILABLE, + "LEGACY_RESULT_AVAILABLE", writes); + } + + public static ShadowLegacyQueueResult unavailable(String missing) { + return new ShadowLegacyQueueResult(Status.LEGACY_RESULT_UNAVAILABLE, + "LEGACY_RESULT_UNAVAILABLE:" + missing, + Collections.emptyList()); + } + + public Status getStatus() { return status; } + public String getReasonCode() { return reasonCode; } + public List getWrites() { return writes; } + + public String getWriteAddresses() { + StringBuilder addresses = new StringBuilder(); + for (LegacyRegisterWrite write : writes) { + if (addresses.length() > 0) { + addresses.append(','); + } + addresses.append(write.getAddress()); + } + return addresses.toString(); + } + + public LegacyRegisterWrite write(String address) { + for (LegacyRegisterWrite write : writes) { + if (write.getAddress().equals(address)) { + return write; + } + } + return null; + } + + public static final class LegacyRegisterWrite { + private final String address; + private final DataType dataType; + private final List values; + private final int writeSequence; + private final boolean trigger; + + LegacyRegisterWrite(String address, DataType dataType, + List values, int writeSequence, + boolean trigger) { + this.address = address; + this.dataType = dataType; + this.values = Collections.unmodifiableList( + new ArrayList(values)); + this.writeSequence = writeSequence; + this.trigger = trigger; + } + + public String getAddress() { return address; } + public DataType getDataType() { return dataType; } + public List getValues() { return values; } + public int getWriteSequence() { return writeSequence; } + public boolean isTrigger() { return trigger; } + } +} diff --git a/ruoyi-wcs-shadow/src/main/java/com/ruoyi/shadow/facts/ShadowMapFact.java b/ruoyi-wcs-shadow/src/main/java/com/ruoyi/shadow/facts/ShadowMapFact.java new file mode 100644 index 0000000..c9c8352 --- /dev/null +++ b/ruoyi-wcs-shadow/src/main/java/com/ruoyi/shadow/facts/ShadowMapFact.java @@ -0,0 +1,34 @@ +package com.ruoyi.shadow.facts; + +public final class ShadowMapFact { + private final String guid; + private final String siteId; + private final int x; + private final int y; + private final int z; + private final int value; + + public ShadowMapFact(String guid, String siteId, int x, int y, int z, + int value) { + this.guid = required(guid, "guid"); + this.siteId = required(siteId, "siteId"); + this.x = x; + this.y = y; + this.z = z; + this.value = value; + } + + public String getGuid() { return guid; } + public String getSiteId() { return siteId; } + public int getX() { return x; } + public int getY() { return y; } + public int getZ() { return z; } + public int getValue() { return value; } + + private static String required(String value, String name) { + if (value == null || value.trim().isEmpty()) { + throw new IllegalArgumentException(name + " must not be blank"); + } + return value.trim(); + } +} diff --git a/ruoyi-wcs-shadow/src/main/java/com/ruoyi/shadow/facts/ShadowQueueFact.java b/ruoyi-wcs-shadow/src/main/java/com/ruoyi/shadow/facts/ShadowQueueFact.java new file mode 100644 index 0000000..6f1b007 --- /dev/null +++ b/ruoyi-wcs-shadow/src/main/java/com/ruoyi/shadow/facts/ShadowQueueFact.java @@ -0,0 +1,135 @@ +package com.ruoyi.shadow.facts; + +import java.time.Instant; + +public final class ShadowQueueFact { + private final String tableName; + private final String guid; + private final String taskNo; + private final String taskType; + private final Long priority; + private final String cargoPickSite; + private final String cargoPlaceSite; + private final String startSite; + private final String endSite; + private final Instant createdAt; + private final String agvId; + private final String pathInfo; + private final String turnPath; + private final String default02; + private final String default03; + private final String default04; + private final String default05; + private final String default06; + private final String default07; + private final String default08; + private final String contentHash; + + private ShadowQueueFact(Builder builder) { + this.tableName = required(builder.tableName, "tableName"); + this.guid = required(builder.guid, "guid"); + this.taskNo = builder.taskNo; + this.taskType = builder.taskType; + this.priority = builder.priority; + this.cargoPickSite = builder.cargoPickSite; + this.cargoPlaceSite = builder.cargoPlaceSite; + this.startSite = builder.startSite; + this.endSite = builder.endSite; + this.createdAt = builder.createdAt; + this.agvId = builder.agvId; + this.pathInfo = builder.pathInfo; + this.turnPath = builder.turnPath; + this.default02 = builder.default02; + this.default03 = builder.default03; + this.default04 = builder.default04; + this.default05 = builder.default05; + this.default06 = builder.default06; + this.default07 = builder.default07; + this.default08 = builder.default08; + this.contentHash = FactHash.sha256(tableName, guid, taskNo, taskType, + priority, cargoPickSite, cargoPlaceSite, startSite, endSite, + createdAt, agvId, pathInfo, turnPath, default02, default03, + default04, default05, default06, default07, default08); + } + + public static Builder builder(String tableName, String guid) { + return new Builder(tableName, guid); + } + + public String getTableName() { return tableName; } + public String getGuid() { return guid; } + public String getTaskNo() { return taskNo; } + public String getTaskType() { return taskType; } + public Long getPriority() { return priority; } + public String getCargoPickSite() { return cargoPickSite; } + public String getCargoPlaceSite() { return cargoPlaceSite; } + public String getStartSite() { return startSite; } + public String getEndSite() { return endSite; } + public Instant getCreatedAt() { return createdAt; } + public String getAgvId() { return agvId; } + public String getPathInfo() { return pathInfo; } + public String getTurnPath() { return turnPath; } + public String getDefault02() { return default02; } + public String getDefault03() { return default03; } + public String getDefault04() { return default04; } + public String getDefault05() { return default05; } + public String getDefault06() { return default06; } + public String getDefault07() { return default07; } + public String getDefault08() { return default08; } + public String getContentHash() { return contentHash; } + + private static String required(String value, String name) { + if (value == null || value.trim().isEmpty()) { + throw new IllegalArgumentException(name + " must not be blank"); + } + return value.trim(); + } + + public static final class Builder { + private final String tableName; + private final String guid; + private String taskNo; + private String taskType; + private Long priority; + private String cargoPickSite; + private String cargoPlaceSite; + private String startSite; + private String endSite; + private Instant createdAt; + private String agvId; + private String pathInfo; + private String turnPath; + private String default02; + private String default03; + private String default04; + private String default05; + private String default06; + private String default07; + private String default08; + + private Builder(String tableName, String guid) { + this.tableName = tableName; + this.guid = guid; + } + + public Builder taskNo(String value) { taskNo = value; return this; } + public Builder taskType(String value) { taskType = value; return this; } + public Builder priority(Long value) { priority = value; return this; } + public Builder cargoPickSite(String value) { cargoPickSite = value; return this; } + public Builder cargoPlaceSite(String value) { cargoPlaceSite = value; return this; } + public Builder startSite(String value) { startSite = value; return this; } + public Builder endSite(String value) { endSite = value; return this; } + public Builder createdAt(Instant value) { createdAt = value; return this; } + public Builder agvId(String value) { agvId = value; return this; } + public Builder pathInfo(String value) { pathInfo = value; return this; } + public Builder turnPath(String value) { turnPath = value; return this; } + public Builder default02(String value) { default02 = value; return this; } + public Builder default03(String value) { default03 = value; return this; } + public Builder default04(String value) { default04 = value; return this; } + public Builder default05(String value) { default05 = value; return this; } + public Builder default06(String value) { default06 = value; return this; } + public Builder default07(String value) { default07 = value; return this; } + public Builder default08(String value) { default08 = value; return this; } + public ShadowQueueFact build() { return new ShadowQueueFact(this); } + } +} diff --git a/ruoyi-wcs-shadow/src/main/java/com/ruoyi/shadow/facts/ShadowReadCursor.java b/ruoyi-wcs-shadow/src/main/java/com/ruoyi/shadow/facts/ShadowReadCursor.java new file mode 100644 index 0000000..a7dbc71 --- /dev/null +++ b/ruoyi-wcs-shadow/src/main/java/com/ruoyi/shadow/facts/ShadowReadCursor.java @@ -0,0 +1,38 @@ +package com.ruoyi.shadow.facts; + +import java.time.Duration; +import java.time.Instant; + +public final class ShadowReadCursor { + private static final Instant BEGINNING = Instant.parse("1970-01-01T00:00:00Z"); + + private final Instant lastCreateDate; + private final String lastGuid; + + public ShadowReadCursor(Instant lastCreateDate, String lastGuid) { + if (lastCreateDate == null) { + throw new IllegalArgumentException("lastCreateDate must not be null"); + } + this.lastCreateDate = lastCreateDate; + this.lastGuid = lastGuid == null ? "" : lastGuid; + } + + public static ShadowReadCursor initial() { + return new ShadowReadCursor(BEGINNING, ""); + } + + public ShadowReadCursor withOverlap(Duration overlap) { + if (overlap == null || overlap.isNegative()) { + throw new IllegalArgumentException("overlap must not be negative"); + } + Instant start = lastCreateDate.minus(overlap); + if (start.isBefore(BEGINNING)) { + start = BEGINNING; + } + return new ShadowReadCursor(start, ""); + } + + public Instant getLastCreateDate() { return lastCreateDate; } + public Instant getOverlapStart() { return lastCreateDate; } + public String getLastGuid() { return lastGuid; } +} diff --git a/ruoyi-wcs-shadow/src/main/java/com/ruoyi/shadow/facts/ShadowVehicleFact.java b/ruoyi-wcs-shadow/src/main/java/com/ruoyi/shadow/facts/ShadowVehicleFact.java new file mode 100644 index 0000000..c4179aa --- /dev/null +++ b/ruoyi-wcs-shadow/src/main/java/com/ruoyi/shadow/facts/ShadowVehicleFact.java @@ -0,0 +1,161 @@ +package com.ruoyi.shadow.facts; + +import java.time.Instant; + +public final class ShadowVehicleFact { + private final String guid; + private final String deviceId; + private final String agvStatus; + private final Integer taskNo; + private final String sourceSite; + private final String targetSite; + private final String currentSite; + private final Integer taskType; + private final Integer battery; + private final Integer workMode; + private final Integer alarm; + private final Integer status; + private final Integer finish; + private final Integer chargeFlag; + private final Integer workPhase; + private final Integer errorCode; + private final Integer changeLayer; + private final Integer cargoFlag; + private final String path; + private final Integer allowSend; + private final Integer currentFloor; + private final Integer locked; + private final String spare2; + private final String spare5; + private final Instant capturedAt; + private final String contentHash; + + private ShadowVehicleFact(Builder builder) { + this.guid = required(builder.guid, "guid"); + this.deviceId = required(builder.deviceId, "deviceId"); + this.agvStatus = builder.agvStatus; + this.taskNo = builder.taskNo; + this.sourceSite = builder.sourceSite; + this.targetSite = builder.targetSite; + this.currentSite = builder.currentSite; + this.taskType = builder.taskType; + this.battery = builder.battery; + this.workMode = builder.workMode; + this.alarm = builder.alarm; + this.status = builder.status; + this.finish = builder.finish; + this.chargeFlag = builder.chargeFlag; + this.workPhase = builder.workPhase; + this.errorCode = builder.errorCode; + this.changeLayer = builder.changeLayer; + this.cargoFlag = builder.cargoFlag; + this.path = builder.path; + this.allowSend = builder.allowSend; + this.currentFloor = builder.currentFloor; + this.locked = builder.locked; + this.spare2 = builder.spare2; + this.spare5 = builder.spare5; + this.capturedAt = builder.capturedAt; + this.contentHash = FactHash.sha256(guid, deviceId, agvStatus, + taskNo, sourceSite, targetSite, currentSite, taskType, + battery, workMode, alarm, status, finish, chargeFlag, + workPhase, errorCode, changeLayer, cargoFlag, path, + allowSend, currentFloor, locked, spare2, spare5, capturedAt); + } + + public static Builder builder(String guid, String deviceId) { + return new Builder(guid, deviceId); + } + + public String getGuid() { return guid; } + public String getDeviceId() { return deviceId; } + public String getAgvStatus() { return agvStatus; } + public Integer getTaskNo() { return taskNo; } + public String getSourceSite() { return sourceSite; } + public String getTargetSite() { return targetSite; } + public String getCurrentSite() { return currentSite; } + public Integer getTaskType() { return taskType; } + public Integer getBattery() { return battery; } + public Integer getWorkMode() { return workMode; } + public Integer getAlarm() { return alarm; } + public Integer getStatus() { return status; } + public Integer getFinish() { return finish; } + public Integer getChargeFlag() { return chargeFlag; } + public Integer getWorkPhase() { return workPhase; } + public Integer getErrorCode() { return errorCode; } + public Integer getChangeLayer() { return changeLayer; } + public Integer getCargoFlag() { return cargoFlag; } + public String getPath() { return path; } + public Integer getAllowSend() { return allowSend; } + public Integer getCurrentFloor() { return currentFloor; } + public Integer getLocked() { return locked; } + public String getSpare2() { return spare2; } + public String getSpare5() { return spare5; } + public Instant getCapturedAt() { return capturedAt; } + public String getContentHash() { return contentHash; } + + private static String required(String value, String name) { + if (value == null || value.trim().isEmpty()) { + throw new IllegalArgumentException(name + " must not be blank"); + } + return value.trim(); + } + + public static final class Builder { + private final String guid; + private final String deviceId; + private String agvStatus; + private Integer taskNo; + private String sourceSite; + private String targetSite; + private String currentSite; + private Integer taskType; + private Integer battery; + private Integer workMode; + private Integer alarm; + private Integer status; + private Integer finish; + private Integer chargeFlag; + private Integer workPhase; + private Integer errorCode; + private Integer changeLayer; + private Integer cargoFlag; + private String path; + private Integer allowSend; + private Integer currentFloor; + private Integer locked; + private String spare2; + private String spare5; + private Instant capturedAt; + + private Builder(String guid, String deviceId) { + this.guid = guid; + this.deviceId = deviceId; + } + + public Builder agvStatus(String value) { agvStatus = value; return this; } + public Builder taskNo(Integer value) { taskNo = value; return this; } + public Builder sourceSite(String value) { sourceSite = value; return this; } + public Builder targetSite(String value) { targetSite = value; return this; } + public Builder currentSite(String value) { currentSite = value; return this; } + public Builder taskType(Integer value) { taskType = value; return this; } + public Builder battery(Integer value) { battery = value; return this; } + public Builder workMode(Integer value) { workMode = value; return this; } + public Builder alarm(Integer value) { alarm = value; return this; } + public Builder status(Integer value) { status = value; return this; } + public Builder finish(Integer value) { finish = value; return this; } + public Builder chargeFlag(Integer value) { chargeFlag = value; return this; } + public Builder workPhase(Integer value) { workPhase = value; return this; } + public Builder errorCode(Integer value) { errorCode = value; return this; } + public Builder changeLayer(Integer value) { changeLayer = value; return this; } + public Builder cargoFlag(Integer value) { cargoFlag = value; return this; } + public Builder path(String value) { path = value; return this; } + public Builder allowSend(Integer value) { allowSend = value; return this; } + public Builder currentFloor(Integer value) { currentFloor = value; return this; } + public Builder locked(Integer value) { locked = value; return this; } + public Builder spare2(String value) { spare2 = value; return this; } + public Builder spare5(String value) { spare5 = value; return this; } + public Builder capturedAt(Instant value) { capturedAt = value; return this; } + public ShadowVehicleFact build() { return new ShadowVehicleFact(this); } + } +} 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 new file mode 100644 index 0000000..7eece7f --- /dev/null +++ b/ruoyi-wcs-shadow/src/main/java/com/ruoyi/shadow/facts/SqlServerShadowFactReader.java @@ -0,0 +1,388 @@ +package com.ruoyi.shadow.facts; + +import org.springframework.dao.DataAccessException; +import org.springframework.jdbc.core.ConnectionCallback; +import org.springframework.jdbc.core.JdbcTemplate; +import org.springframework.transaction.support.TransactionTemplate; + +import java.sql.PreparedStatement; +import java.sql.ResultSet; +import java.sql.SQLException; +import java.sql.Timestamp; +import java.time.Clock; +import java.time.Duration; +import java.time.Instant; +import java.util.ArrayList; +import java.util.Calendar; +import java.util.Collections; +import java.util.List; +import java.util.TimeZone; +import java.util.concurrent.atomic.AtomicBoolean; +import java.util.concurrent.atomic.AtomicInteger; + +public final class SqlServerShadowFactReader { + static final int QUEUE_LIMIT = 500; + static final int VEHICLE_LIMIT = 100; + static final int MAP_LIMIT = 10_000; + private static final Duration CURSOR_OVERLAP = Duration.ofSeconds(5); + private static final Duration MAXIMUM_BATCH_DURATION = Duration.ofSeconds(4); + private static final Duration[] BACKOFF = { + Duration.ofSeconds(1), Duration.ofSeconds(2), + Duration.ofSeconds(4), Duration.ofSeconds(8), + Duration.ofSeconds(30) + }; + + private final FactSource source; + private final TransactionTemplate sourceTransaction; + private final Clock clock; + private final AtomicBoolean inFlight = new AtomicBoolean(); + private final AtomicInteger consecutiveFailures = new AtomicInteger(); + private volatile List cachedMaps; + + public SqlServerShadowFactReader(FactSource source, + TransactionTemplate sourceTransaction, + Clock clock) { + if (source == null || sourceTransaction == null || clock == null) { + throw new IllegalArgumentException( + "source, sourceTransaction, and clock are required"); + } + this.source = source; + this.sourceTransaction = sourceTransaction; + this.sourceTransaction.setReadOnly(true); + this.clock = clock; + } + + public ShadowFactBatch read(ShadowReadCursor cursor, Instant deadline) { + if (cursor == null || deadline == null) { + throw new IllegalArgumentException("cursor and deadline are required"); + } + Instant startedAt = clock.instant(); + Duration requested = Duration.between(startedAt, deadline); + if (requested.isNegative() || requested.isZero()) { + throw immediate("SOURCE_BATCH_DEADLINE_EXCEEDED"); + } + if (requested.compareTo(MAXIMUM_BATCH_DURATION) > 0) { + throw immediate("SOURCE_BATCH_DEADLINE_MUST_NOT_EXCEED_4_SECONDS"); + } + if (!inFlight.compareAndSet(false, true)) { + throw immediate("SOURCE_CYCLE_ALREADY_IN_FLIGHT"); + } + try { + ShadowFactBatch batch = sourceTransaction.execute(status -> + collect(cursor.withOverlap(CURSOR_OVERLAP), deadline)); + if (batch == null) { + throw new ShadowFactReadException("SOURCE_EMPTY_BATCH", + Duration.ZERO); + } + consecutiveFailures.set(0); + return batch; + } catch (ShadowFactReadException failure) { + if (failure.getReasonCode().startsWith("SOURCE_CYCLE_") + || failure.getReasonCode().contains("DEADLINE_MUST_NOT")) { + throw failure; + } + throw failed(failure.getReasonCode(), failure); + } catch (DataAccessException failure) { + throw failed(classify(failure), failure); + } catch (RuntimeException failure) { + throw failed("SOURCE_READ_FAILED", failure); + } finally { + inFlight.set(false); + } + } + + private ShadowFactBatch collect(ShadowReadCursor cursor, Instant deadline) { + ensureTime(deadline); + Instant databaseStartedAt = source.readDatabaseClock(deadline); + List queues = immutable(source.readQueues(cursor, deadline)); + requireLimit("QUEUE", queues.size(), QUEUE_LIMIT); + List vehicles = + immutable(source.readVehicles(deadline)); + requireLimit("VEHICLE", vehicles.size(), VEHICLE_LIMIT); + List maps = cachedMaps; + if (maps == null) { + maps = immutable(source.readMaps(deadline)); + requireLimit("MAP", maps.size(), MAP_LIMIT); + } + ensureTime(deadline); + Instant databaseCompletedAt = source.readDatabaseClock(deadline); + ensureTime(deadline); + ShadowFactBatch batch = ShadowFactBatch.create(databaseStartedAt, + databaseCompletedAt, queues, vehicles, maps); + if (cachedMaps == null) { + cachedMaps = maps; + } + return batch; + } + + private void ensureTime(Instant deadline) { + queryTimeoutSeconds(clock.instant(), deadline); + } + + public static int queryTimeoutSeconds(Instant now, Instant deadline) { + if (now == null || deadline == null || !deadline.isAfter(now)) { + throw immediate("SOURCE_BATCH_DEADLINE_EXCEEDED"); + } + long remainingMillis = Duration.between(now, deadline).toMillis(); + long completeSeconds = remainingMillis / 1_000L; + if (completeSeconds < 1L) { + throw immediate("SOURCE_BATCH_DEADLINE_EXCEEDED"); + } + return (int) Math.min(3L, completeSeconds); + } + + private static void requireLimit(String kind, int actual, int limit) { + if (actual > limit) { + throw new ShadowFactReadException( + kind + "_ROW_LIMIT_EXCEEDED:" + limit, Duration.ZERO); + } + } + + private ShadowFactReadException failed(String reason, Throwable cause) { + int failure = consecutiveFailures.getAndIncrement(); + Duration delay = BACKOFF[Math.min(failure, BACKOFF.length - 1)]; + return new ShadowFactReadException(reason, delay, cause); + } + + private static ShadowFactReadException immediate(String reason) { + return new ShadowFactReadException(reason, Duration.ZERO); + } + + private static String classify(DataAccessException failure) { + String name = failure.getClass().getSimpleName().toUpperCase(); + if (name.contains("TIMEOUT") || name.contains("TRANSIENT")) { + return "SOURCE_QUERY_TIMEOUT"; + } + return "SOURCE_UNAVAILABLE"; + } + + private static List immutable(List values) { + if (values == null) { + throw new ShadowFactReadException("SOURCE_RETURNED_NULL_FACT_LIST", + Duration.ZERO); + } + return Collections.unmodifiableList(new ArrayList(values)); + } + + public interface FactSource { + Instant readDatabaseClock(Instant deadline); + List readQueues(ShadowReadCursor cursor, + Instant deadline); + List readVehicles(Instant deadline); + List readMaps(Instant deadline); + } + + public static final class JdbcFactSource implements FactSource { + static final String CLOCK_SQL = + "SELECT database_utc_time FROM shadow_ro.v_current_database_clock"; + static final String QUEUE_SQL_PREFIX = + "SELECT TOP (?) guid, task_no, task_type, priority, " + + "cargo_pick_site, cargo_place_site, start_site, end_site, " + + "create_date_utc, agv_id, path_info, trun_path, default02, " + + "default03, default04, default05, default06, default07, default08 "; + static final String QUEUE_SQL_SUFFIX = + " WHERE create_date_utc > ? " + + "OR (create_date_utc = ? AND guid > ?) " + + "ORDER BY create_date_utc, guid"; + static final String ACTIVE_QUEUE_SQL_SUFFIX = + " ORDER BY create_date_utc, guid"; + static final String VEHICLE_SQL = + "SELECT TOP (?) guid, device_id, agv_status, task_no, src_site, " + + "tar_site, curr_site, task_type, battery, work_mode, alarm, " + + "status, finish, charge_flag, work_phase, err_msg, " + + "change_layer, cargo_flag, path, allow_send, curr_floor, " + + "locked, spare2, spare5, last_time_utc " + + "FROM shadow_ro.v_device_srgv ORDER BY device_id, guid"; + static final String MAP_SQL = + "SELECT TOP (?) guid, site_id, x, y, z, value " + + "FROM shadow_ro.v_wcs_srgv_map ORDER BY site_id, guid"; + + private final JdbcTemplate jdbc; + private final Clock clock; + + public JdbcFactSource(JdbcTemplate jdbc, Clock clock) { + if (jdbc == null || clock == null) { + throw new IllegalArgumentException("jdbc and clock are required"); + } + this.jdbc = jdbc; + this.clock = clock; + } + + @Override + public Instant readDatabaseClock(Instant deadline) { + return jdbc.execute((ConnectionCallback) connection -> { + try (PreparedStatement statement = connection.prepareStatement( + CLOCK_SQL)) { + timeout(statement, deadline); + try (ResultSet result = statement.executeQuery()) { + if (!result.next()) { + throw new ShadowFactReadException( + "SOURCE_DATABASE_CLOCK_MISSING", Duration.ZERO); + } + return instant(result, "database_utc_time"); + } + } + }); + } + + @Override + public List readQueues(ShadowReadCursor cursor, + Instant deadline) { + List result = queryQueues( + "shadow_ro.v_wcs_srgv_queue", null, QUEUE_LIMIT + 1, + deadline, ACTIVE_QUEUE_SQL_SUFFIX); + if (result.size() > QUEUE_LIMIT) { + return result; + } + int remainingWithOverflow = QUEUE_LIMIT - result.size() + 1; + result.addAll(queryQueues("shadow_ro.v_wcs_srgv_queue_finish", + cursor, remainingWithOverflow, deadline, QUEUE_SQL_SUFFIX)); + return result; + } + + @Override + public List readVehicles(Instant deadline) { + return jdbc.execute((ConnectionCallback>) + connection -> { + try (PreparedStatement statement = connection.prepareStatement( + VEHICLE_SQL)) { + statement.setInt(1, VEHICLE_LIMIT + 1); + timeout(statement, deadline); + try (ResultSet result = statement.executeQuery()) { + List facts = + new ArrayList(); + while (result.next()) { + facts.add(ShadowVehicleFact.builder( + result.getString("guid"), + result.getString("device_id")) + .agvStatus(result.getString("agv_status")) + .taskNo(intValue(result, "task_no")) + .sourceSite(result.getString("src_site")) + .targetSite(result.getString("tar_site")) + .currentSite(result.getString("curr_site")) + .taskType(intValue(result, "task_type")) + .battery(intValue(result, "battery")) + .workMode(intValue(result, "work_mode")) + .alarm(intValue(result, "alarm")) + .status(intValue(result, "status")) + .finish(intValue(result, "finish")) + .chargeFlag(intValue(result, "charge_flag")) + .workPhase(intValue(result, "work_phase")) + .errorCode(intValue(result, "err_msg")) + .changeLayer(intValue(result, "change_layer")) + .cargoFlag(intValue(result, "cargo_flag")) + .path(result.getString("path")) + .allowSend(intValue(result, "allow_send")) + .currentFloor(intValue(result, "curr_floor")) + .locked(intValue(result, "locked")) + .spare2(result.getString("spare2")) + .spare5(result.getString("spare5")) + .capturedAt(instant(result, "last_time_utc")) + .build()); + } + return facts; + } + } + }); + } + + @Override + public List readMaps(Instant deadline) { + return jdbc.execute((ConnectionCallback>) + connection -> { + try (PreparedStatement statement = connection.prepareStatement( + MAP_SQL)) { + statement.setInt(1, MAP_LIMIT + 1); + timeout(statement, deadline); + try (ResultSet result = statement.executeQuery()) { + List facts = new ArrayList(); + while (result.next()) { + facts.add(new ShadowMapFact(result.getString("guid"), + result.getString("site_id"), + result.getInt("x"), result.getInt("y"), + result.getInt("z"), result.getInt("value"))); + } + return facts; + } + } + }); + } + + private List queryQueues(String table, + ShadowReadCursor cursor, + int limit, + Instant deadline, + String suffix) { + String sql = QUEUE_SQL_PREFIX + "FROM " + table + suffix; + return jdbc.execute((ConnectionCallback>) + connection -> { + try (PreparedStatement statement = connection.prepareStatement(sql)) { + statement.setInt(1, limit); + if (cursor != null) { + Timestamp timestamp = Timestamp.from( + cursor.getLastCreateDate()); + statement.setTimestamp(2, timestamp, utcCalendar()); + statement.setTimestamp(3, timestamp, utcCalendar()); + statement.setString(4, cursor.getLastGuid()); + } + timeout(statement, deadline); + try (ResultSet result = statement.executeQuery()) { + List facts = + new ArrayList(); + while (result.next()) { + facts.add(ShadowQueueFact.builder(table, + result.getString("guid")) + .taskNo(result.getString("task_no")) + .taskType(result.getString("task_type")) + .priority(longValue(result, "priority")) + .cargoPickSite(result.getString("cargo_pick_site")) + .cargoPlaceSite(result.getString("cargo_place_site")) + .startSite(result.getString("start_site")) + .endSite(result.getString("end_site")) + .createdAt(instant(result, "create_date_utc")) + .agvId(result.getString("agv_id")) + .pathInfo(result.getString("path_info")) + .turnPath(result.getString("trun_path")) + .default02(result.getString("default02")) + .default03(result.getString("default03")) + .default04(result.getString("default04")) + .default05(result.getString("default05")) + .default06(result.getString("default06")) + .default07(result.getString("default07")) + .default08(result.getString("default08")) + .build()); + } + return facts; + } + } + }); + } + + private void timeout(PreparedStatement statement, Instant deadline) + throws SQLException { + statement.setQueryTimeout(queryTimeoutSeconds(clock.instant(), deadline)); + } + + private static Instant instant(ResultSet result, String column) + throws SQLException { + Timestamp timestamp = result.getTimestamp(column, utcCalendar()); + return timestamp == null ? null : timestamp.toInstant(); + } + + private static Calendar utcCalendar() { + return Calendar.getInstance(TimeZone.getTimeZone("UTC")); + } + + private static Long longValue(ResultSet result, String column) + throws SQLException { + long value = result.getLong(column); + return result.wasNull() ? null : value; + } + + private static Integer intValue(ResultSet result, String column) + throws SQLException { + int value = result.getInt(column); + return result.wasNull() ? null : value; + } + } +} 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 new file mode 100644 index 0000000..0fae7f3 --- /dev/null +++ b/ruoyi-wcs-shadow/src/test/java/com/ruoyi/shadow/facts/ShadowFactBatchValidatorTest.java @@ -0,0 +1,277 @@ +package com.ruoyi.shadow.facts; + +import org.junit.jupiter.api.Test; + +import java.time.Duration; +import java.time.Instant; +import java.util.Arrays; +import java.util.Collections; + +import static org.junit.jupiter.api.Assertions.assertEquals; + +class ShadowFactBatchValidatorTest { + private static final Instant NOW = Instant.parse("2026-08-13T02:00:00Z"); + private final ShadowFactBatchValidator validator = + new ShadowFactBatchValidator(Duration.ofSeconds(2), + Duration.ofSeconds(5)); + + @Test + void completeFreshFactsAreComparable() { + 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.minusMillis(500)), NOW); + + assertEquals(ShadowFactValidation.Status.COMPARABLE, + result.getStatus()); + assertEquals("FACTS_COMPLETE", result.getReasonCode()); + } + + @Test + void contradictoryVehiclePositionsAreInsufficient() { + ShadowFactValidation result = validator.validate(batch( + Collections.singletonList(queue("Q-1", "070501", "070502", "1")), + Arrays.asList(vehicle("1", "070501"), + vehicle("1", "070502")), + Arrays.asList(map("070501"), map("070502")), + NOW.minusSeconds(1), NOW.minusMillis(500)), NOW); + + assertInsufficient(result, "CONTRADICTORY_VEHICLE_POSITION:1"); + } + + @Test + void contradictoryVehicleStateIsInsufficientEvenAtTheSameSite() { + ShadowVehicleFact changedBattery = completeVehicleBuilder( + "1", "070501").battery(20).build(); + ShadowFactValidation result = validator.validate(batch( + Collections.singletonList(queue("Q-1", "070501", "070502", "1")), + Arrays.asList(vehicle("1", "070501"), changedBattery), + Arrays.asList(map("070501"), map("070502")), + NOW.minusSeconds(1), NOW.minusMillis(500)), NOW); + + assertInsufficient(result, "CONTRADICTORY_VEHICLE_FACT:1"); + } + + @Test + void missingRequiredVehicleOrMapFactsAreInsufficient() { + ShadowFactValidation missingVehicles = validator.validate(batch( + Collections.emptyList(), Collections.emptyList(), + Collections.singletonList(map("070501")), + NOW.minusSeconds(1), NOW.minusMillis(500)), NOW); + assertInsufficient(missingVehicles, "MISSING_VEHICLE_FACTS"); + + ShadowFactValidation missingMaps = validator.validate(batch( + Collections.emptyList(), + Collections.singletonList(vehicle("1", "070501")), + Collections.emptyList(), NOW.minusSeconds(1), + NOW.minusMillis(500)), NOW); + assertInsufficient(missingMaps, "MISSING_MAP_FACTS"); + } + + @Test + void contradictoryMapFactsForOneSiteAreInsufficient() { + ShadowFactValidation result = validator.validate(batch( + Collections.emptyList(), + Collections.singletonList(vehicle("1", "070501")), + Arrays.asList(map("070501"), + new ShadowMapFact("M-OTHER", "070501", 2, 1, 1, 1)), + NOW.minusSeconds(1), NOW.minusMillis(500)), NOW); + + assertInsufficient(result, "CONTRADICTORY_MAP_SITE:070501"); + } + + @Test + void queueMutationInsideOverlapIsInsufficient() { + ShadowQueueFact first = queue("Q-1", "070501", "070502", "1"); + ShadowQueueFact changed = ShadowQueueFact.builder( + "shadow_ro.v_wcs_srgv_queue", "Q-1") + .taskNo("30001").taskType("2").priority(1L) + .startSite("070501").endSite("070503") + .createdAt(NOW.minusSeconds(2)).agvId("1") + .pathInfo("70501,70503,").turnPath("") + .default02("1").build(); + ShadowFactValidation result = validator.validate(batch( + Arrays.asList(first, changed), + Collections.singletonList(vehicle("1", "070501")), + Arrays.asList(map("070501"), map("070502"), map("070503")), + NOW.minusSeconds(1), NOW.minusMillis(500)), NOW); + + assertInsufficient(result, + "QUEUE_MUTATED_DURING_OVERLAP:shadow_ro.v_wcs_srgv_queue:Q-1"); + } + + @Test + void nonMonotonicOrPartialDatabaseClockIsInsufficient() { + ShadowFactBatch nonMonotonic = batch(Collections.emptyList(), + Collections.emptyList(), Collections.emptyList(), + NOW, NOW.minusSeconds(1)); + assertInsufficient(validator.validate(nonMonotonic, NOW), + "NON_MONOTONIC_DATABASE_CLOCK"); + + ShadowFactBatch partial = batch(Collections.emptyList(), + Collections.emptyList(), Collections.emptyList(), + NOW.minusSeconds(4), NOW.minusSeconds(1)); + assertInsufficient(validator.validate(partial, NOW), + "DATABASE_CAPTURE_SPAN_EXCEEDED"); + } + + @Test + void staleOrFutureDatabaseClockIsInsufficient() { + assertInsufficient(validator.validate(batch(Collections.emptyList(), + Collections.emptyList(), Collections.emptyList(), + NOW.minusSeconds(8), NOW.minusSeconds(7)), NOW), + "STALE_DATABASE_CLOCK"); + assertInsufficient(validator.validate(batch(Collections.emptyList(), + Collections.emptyList(), Collections.emptyList(), + NOW.plusSeconds(1), NOW.plusSeconds(2)), NOW), + "DATABASE_CLOCK_IN_FUTURE"); + } + + @Test + void missingMapSiteIsInsufficient() { + ShadowFactValidation result = validator.validate(batch( + Collections.singletonList(queue("Q-1", "070501", "070502", "1")), + Collections.singletonList(vehicle("1", "070501")), + Collections.singletonList(map("070501")), + NOW.minusSeconds(1), NOW.minusMillis(500)), NOW); + + assertInsufficient(result, "MISSING_MAP_SITE:070502"); + } + + @Test + void missingLegacyAssignedVehicleIsInsufficient() { + ShadowFactValidation result = validator.validate(batch( + Collections.singletonList(queue("Q-1", "070501", "070502", "9")), + Collections.singletonList(vehicle("1", "070501")), + Arrays.asList(map("070501"), map("070502")), + NOW.minusSeconds(1), NOW.minusMillis(500)), NOW); + + assertInsufficient(result, "MISSING_LEGACY_ASSIGNED_VEHICLE:9"); + } + + @Test + void missingVehicleKernelFieldsAreInsufficient() { + ShadowVehicleFact missingSite = completeVehicleBuilder("1", null).build(); + ShadowFactValidation siteResult = validator.validate(batch( + Collections.emptyList(), Collections.singletonList(missingSite), + Collections.emptyList(), NOW.minusSeconds(1), + NOW.minusMillis(500)), NOW); + assertInsufficient(siteResult, "MISSING_VEHICLE_CURRENT_SITE:1"); + + ShadowVehicleFact missingMode = completeVehicleBuilder("1", "070501") + .workMode(null).build(); + ShadowFactValidation modeResult = validator.validate(batch( + Collections.emptyList(), Collections.singletonList(missingMode), + Collections.emptyList(), NOW.minusSeconds(1), + NOW.minusMillis(500)), NOW); + assertInsufficient(modeResult, "MISSING_VEHICLE_WORK_MODE:1"); + + ShadowVehicleFact invalidBattery = completeVehicleBuilder("1", "070501") + .battery(101).build(); + ShadowFactValidation batteryResult = validator.validate(batch( + Collections.emptyList(), Collections.singletonList(invalidBattery), + Collections.emptyList(), NOW.minusSeconds(1), + NOW.minusMillis(500)), NOW); + assertInsufficient(batteryResult, "INVALID_VEHICLE_BATTERY:1"); + } + + @Test + void missingQueueClockIsInsufficient() { + ShadowQueueFact queue = ShadowQueueFact.builder( + "shadow_ro.v_wcs_srgv_queue", "Q-1") + .taskNo("30001").taskType("2").priority(1L) + .startSite("070501").endSite("070502") + .cargoPickSite("070501").cargoPlaceSite("070502") + .createdAt(null).agvId("1") + .pathInfo("70501,70502,").turnPath("") + .default02("1").build(); + + ShadowFactValidation result = validator.validate(batch( + Collections.singletonList(queue), + Collections.singletonList(vehicle("1", "070501")), + Arrays.asList(map("070501"), map("070502")), + NOW.minusSeconds(1), NOW.minusMillis(500)), NOW); + + assertInsufficient(result, + "MISSING_QUEUE_CREATED_AT:shadow_ro.v_wcs_srgv_queue:Q-1"); + } + + @Test + void missingStageTwoVehicleGateIsInsufficientButKnownClosedGateIsComparable() { + ShadowQueueFact stageTwo = ShadowQueueFact.builder( + "shadow_ro.v_wcs_srgv_queue", "Q-2") + .taskNo("30002").taskType("2").priority(1L) + .startSite("070501").endSite("070502") + .cargoPickSite("070501").cargoPlaceSite("070502") + .createdAt(NOW.minusSeconds(2)).agvId("1") + .pathInfo("70501,70502,").turnPath("") + .default02("1").default06("070503").default08("2") + .build(); + ShadowVehicleFact unknownGate = completeVehicleBuilder("1", "070501") + .spare5(null).build(); + + ShadowFactValidation unknown = validator.validate(batch( + Collections.singletonList(stageTwo), + Collections.singletonList(unknownGate), + Arrays.asList(map("070501"), map("070502")), + NOW.minusSeconds(1), NOW.minusMillis(500)), NOW); + assertInsufficient(unknown, "MISSING_STAGE_TWO_GATE:1"); + + ShadowVehicleFact closedGate = completeVehicleBuilder("1", "070501") + .spare5("0").build(); + ShadowFactValidation knownClosed = validator.validate(batch( + Collections.singletonList(stageTwo), + Collections.singletonList(closedGate), + Arrays.asList(map("070501"), map("070502")), + NOW.minusSeconds(1), NOW.minusMillis(500)), NOW); + assertEquals(ShadowFactValidation.Status.COMPARABLE, + knownClosed.getStatus()); + } + + private static void assertInsufficient(ShadowFactValidation validation, + String reason) { + assertEquals(ShadowFactValidation.Status.INSUFFICIENT_FACTS, + validation.getStatus()); + assertEquals(reason, validation.getReasonCode()); + } + + private static ShadowFactBatch batch( + java.util.List queues, + java.util.List vehicles, + java.util.List maps, + Instant started, Instant completed) { + return ShadowFactBatch.create(started, completed, queues, vehicles, maps); + } + + private static ShadowQueueFact queue(String guid, String start, + String end, String agvId) { + return ShadowQueueFact.builder("shadow_ro.v_wcs_srgv_queue", guid) + .taskNo("30001").taskType("2").priority(1L) + .startSite(start).endSite(end) + .cargoPickSite(start).cargoPlaceSite(end) + .createdAt(NOW.minusSeconds(2)).agvId(agvId) + .pathInfo("70501,70502,").turnPath("") + .default02("1").build(); + } + + private static ShadowVehicleFact vehicle(String deviceId, String site) { + return completeVehicleBuilder(deviceId, site).build(); + } + + private static ShadowVehicleFact.Builder completeVehicleBuilder( + String deviceId, String site) { + return ShadowVehicleFact.builder("V-" + deviceId, deviceId) + .agvStatus("1").taskNo(0) + .sourceSite(site).targetSite(site).currentSite(site) + .taskType(0).battery(76).workMode(5).alarm(0).status(1) + .finish(0).chargeFlag(0).workPhase(0).errorCode(0) + .changeLayer(0).cargoFlag(0).path(site).allowSend(1) + .currentFloor(1).locked(0).spare2("0").spare5("0") + .capturedAt(NOW.minusSeconds(1)); + } + + private static ShadowMapFact map(String site) { + return new ShadowMapFact("M-" + site, site, 1, 1, 1, 1); + } +} diff --git a/ruoyi-wcs-shadow/src/test/java/com/ruoyi/shadow/facts/SqlServerShadowFactReaderTest.java b/ruoyi-wcs-shadow/src/test/java/com/ruoyi/shadow/facts/SqlServerShadowFactReaderTest.java new file mode 100644 index 0000000..787ea07 --- /dev/null +++ b/ruoyi-wcs-shadow/src/test/java/com/ruoyi/shadow/facts/SqlServerShadowFactReaderTest.java @@ -0,0 +1,668 @@ +package com.ruoyi.shadow.facts; + +import org.junit.jupiter.api.Test; +import org.springframework.transaction.PlatformTransactionManager; +import org.springframework.transaction.TransactionDefinition; +import org.springframework.transaction.support.SimpleTransactionStatus; +import org.springframework.transaction.support.TransactionTemplate; + +import java.time.Clock; +import java.time.Duration; +import java.time.Instant; +import java.time.ZoneId; +import java.time.ZoneOffset; +import java.util.ArrayList; +import java.util.Arrays; +import java.util.Collections; +import java.util.List; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicReference; +import javax.sql.DataSource; +import java.sql.Connection; +import java.sql.PreparedStatement; +import java.sql.ResultSet; +import java.sql.Timestamp; + +import org.springframework.jdbc.core.JdbcTemplate; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertFalse; +import static org.junit.jupiter.api.Assertions.assertThrows; +import static org.junit.jupiter.api.Assertions.assertTrue; +import static org.mockito.ArgumentMatchers.any; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.never; +import static org.mockito.Mockito.verify; +import static org.mockito.Mockito.when; + +class SqlServerShadowFactReaderTest { + private static final Instant NOW = Instant.parse("2026-08-13T02:00:00Z"); + + @Test + void readsOneImmutableOrderedBatchInOneReadOnlyTransaction() { + FakeFactSource source = completeSource(); + source.queues = Arrays.asList(queue("Q-2", NOW.minusSeconds(1)), + queue("Q-1", NOW.minusSeconds(2)), + queue("Q-1", NOW.minusSeconds(2))); + source.vehicles = Arrays.asList(vehicle("2", "070503"), + vehicle("1", "070502")); + source.maps = Arrays.asList(map("070503"), map("070502"), + map("070501")); + RecordingTransactions transactions = new RecordingTransactions(); + SqlServerShadowFactReader reader = reader(source, transactions); + + ShadowFactBatch batch = reader.read(ShadowReadCursor.initial(), + NOW.plusSeconds(4)); + + assertEquals(Arrays.asList("Q-1", "Q-2"), guids(batch.getQueues())); + assertEquals(Arrays.asList("1", "2"), deviceIds(batch.getVehicles())); + assertEquals(Arrays.asList("070501", "070502", "070503"), + siteIds(batch.getMaps())); + assertEquals(3, batch.getSourceQueues().size()); + assertEquals("070502", batch.vehicle("1").getCurrentSite()); + assertTrue(batch.getDatabaseCapturedAt().isBefore(NOW.plusSeconds(4))); + assertEquals(1, transactions.calls); + assertTrue(transactions.readOnly); + assertThrows(UnsupportedOperationException.class, + () -> batch.getQueues().clear()); + assertThrows(UnsupportedOperationException.class, + () -> batch.getVehicles().add(vehicle("2", "070503"))); + } + + @Test + void cursorReplaysFiveSecondOverlapAndRetainsMutatedEvidence() { + FakeFactSource source = completeSource(); + ShadowQueueFact old = queue("Q-1", NOW.minusSeconds(10)); + ShadowQueueFact changed = queueBuilder("Q-1", NOW.minusSeconds(10)) + .endSite("070503").pathInfo("70501,70503,").build(); + source.queues = Arrays.asList(old, changed); + SqlServerShadowFactReader reader = reader(source, + new RecordingTransactions()); + + ShadowFactBatch batch = reader.read( + new ShadowReadCursor(NOW.minusSeconds(6), "Q-9"), + NOW.plusSeconds(4)); + + assertEquals(NOW.minusSeconds(11), source.cursor.getOverlapStart()); + assertEquals(2, batch.getSourceQueues().size()); + assertEquals(2, batch.getQueues().size()); + assertFalse(batch.getQueues().get(0).getContentHash().equals( + batch.getQueues().get(1).getContentHash())); + } + + @Test + void hardRowLimitsDiscardPartialBatchAndAdvanceBackoff() { + FakeFactSource source = completeSource(); + source.queues = repeatedQueues(501); + SqlServerShadowFactReader reader = reader(source, + new RecordingTransactions()); + + ShadowFactReadException first = assertThrows( + ShadowFactReadException.class, + () -> reader.read(ShadowReadCursor.initial(), + NOW.plusSeconds(4))); + assertEquals("QUEUE_ROW_LIMIT_EXCEEDED:500", first.getReasonCode()); + assertEquals(Duration.ofSeconds(1), first.getRetryAfter()); + + source.queues = Collections.singletonList(queue("Q-1", NOW)); + source.vehicles = repeatedVehicles(101); + ShadowFactReadException second = assertThrows( + ShadowFactReadException.class, + () -> reader.read(ShadowReadCursor.initial(), + NOW.plusSeconds(4))); + assertEquals("VEHICLE_ROW_LIMIT_EXCEEDED:100", second.getReasonCode()); + assertEquals(Duration.ofSeconds(2), second.getRetryAfter()); + + source.vehicles = Collections.singletonList(vehicle("1", "070502")); + source.maps = repeatedMaps(10_001); + ShadowFactReadException third = assertThrows( + ShadowFactReadException.class, + () -> reader.read(ShadowReadCursor.initial(), + NOW.plusSeconds(4))); + assertEquals("MAP_ROW_LIMIT_EXCEEDED:10000", third.getReasonCode()); + assertEquals(Duration.ofSeconds(4), third.getRetryAfter()); + } + + @Test + void sourceOutageBacksOffWithoutRetryAndSuccessResetsIt() { + FakeFactSource source = completeSource(); + source.failureReason = "SOURCE_QUERY_TIMEOUT"; + SqlServerShadowFactReader reader = reader(source, + new RecordingTransactions()); + + List observed = new ArrayList(); + for (int i = 0; i < 6; i++) { + ShadowFactReadException failure = assertThrows( + ShadowFactReadException.class, + () -> reader.read(ShadowReadCursor.initial(), + NOW.plusSeconds(4))); + observed.add(failure.getRetryAfter()); + } + assertEquals(Arrays.asList(Duration.ofSeconds(1), + Duration.ofSeconds(2), Duration.ofSeconds(4), + Duration.ofSeconds(8), Duration.ofSeconds(30), + Duration.ofSeconds(30)), observed); + assertEquals(6, source.queueCalls); + + source.failureReason = null; + reader.read(ShadowReadCursor.initial(), NOW.plusSeconds(4)); + source.failureReason = "SOURCE_UNAVAILABLE"; + ShadowFactReadException reset = assertThrows( + ShadowFactReadException.class, + () -> reader.read(ShadowReadCursor.initial(), + NOW.plusSeconds(4))); + assertEquals(Duration.ofSeconds(1), reset.getRetryAfter()); + } + + @Test + void wholeBatchDeadlineAndQueryTimeoutAreFailClosed() { + assertEquals(3, SqlServerShadowFactReader.queryTimeoutSeconds( + NOW, NOW.plusMillis(3_500))); + assertEquals(1, SqlServerShadowFactReader.queryTimeoutSeconds( + NOW, NOW.plusMillis(1_001))); + ShadowFactReadException subSecond = assertThrows( + ShadowFactReadException.class, + () -> SqlServerShadowFactReader.queryTimeoutSeconds( + NOW, NOW.plusMillis(999))); + assertEquals("SOURCE_BATCH_DEADLINE_EXCEEDED", + subSecond.getReasonCode()); + ShadowFactReadException expired = assertThrows( + ShadowFactReadException.class, + () -> SqlServerShadowFactReader.queryTimeoutSeconds( + NOW, NOW)); + assertEquals("SOURCE_BATCH_DEADLINE_EXCEEDED", + expired.getReasonCode()); + + FakeFactSource source = completeSource(); + SqlServerShadowFactReader reader = reader(source, + new RecordingTransactions()); + ShadowFactReadException tooLong = assertThrows( + ShadowFactReadException.class, + () -> reader.read(ShadowReadCursor.initial(), + NOW.plusSeconds(5))); + assertEquals("SOURCE_BATCH_DEADLINE_MUST_NOT_EXCEED_4_SECONDS", + tooLong.getReasonCode()); + } + + @Test + void batchThatFinishesAfterItsDeadlineIsDiscardedAndBacksOff() { + FakeFactSource source = completeSource(); + RecordingTransactions transactions = new RecordingTransactions(); + SqlServerShadowFactReader reader = new SqlServerShadowFactReader( + source, transactions.template, + new SequenceClock(NOW, NOW, NOW, NOW.plusSeconds(5))); + + ShadowFactReadException failure = assertThrows( + ShadowFactReadException.class, + () -> reader.read(ShadowReadCursor.initial(), + NOW.plusSeconds(4))); + + assertEquals("SOURCE_BATCH_DEADLINE_EXCEEDED", + failure.getReasonCode()); + assertEquals(Duration.ofSeconds(1), failure.getRetryAfter()); + } + + @Test + void contentHashDistinguishesMissingValuesFromLiteralNullText() { + assertFalse(FactHash.sha256((Object) null) + .equals(FactHash.sha256("null"))); + } + + @Test + void jdbcSourceUsesOnlyApprovedViewsAndNormalizedClockColumns() + throws Exception { + assertTrue(SqlServerShadowFactReader.JdbcFactSource.QUEUE_SQL_PREFIX + .contains("create_date_utc")); + assertTrue(SqlServerShadowFactReader.JdbcFactSource.QUEUE_SQL_SUFFIX + .contains("create_date_utc")); + assertTrue(SqlServerShadowFactReader.JdbcFactSource.VEHICLE_SQL + .contains("last_time_utc")); + String allSql = SqlServerShadowFactReader.JdbcFactSource.CLOCK_SQL + + SqlServerShadowFactReader.JdbcFactSource.QUEUE_SQL_PREFIX + + SqlServerShadowFactReader.JdbcFactSource.QUEUE_SQL_SUFFIX + + SqlServerShadowFactReader.JdbcFactSource.VEHICLE_SQL + + SqlServerShadowFactReader.JdbcFactSource.MAP_SQL; + assertFalse(allSql.toLowerCase().contains(" dbo.")); + assertFalse(allSql.toLowerCase().contains(" wcs_srgv_queue")); + assertFalse(allSql.toLowerCase().contains(" device_srgv")); + + DataSource dataSource = mock(DataSource.class); + Connection connection = mock(Connection.class); + PreparedStatement statement = mock(PreparedStatement.class); + ResultSet result = mock(ResultSet.class); + when(dataSource.getConnection()).thenReturn(connection); + when(connection.prepareStatement( + SqlServerShadowFactReader.JdbcFactSource.CLOCK_SQL)) + .thenReturn(statement); + when(statement.executeQuery()).thenReturn(result); + when(result.next()).thenReturn(true); + when(result.getObject("database_utc_time")).thenReturn(null); + when(result.getTimestamp( + org.mockito.ArgumentMatchers.eq("database_utc_time"), + org.mockito.ArgumentMatchers.any(java.util.Calendar.class))) + .thenReturn(Timestamp.from(NOW.minusMillis(500))); + JdbcTemplate jdbc = new JdbcTemplate(dataSource); + SqlServerShadowFactReader.JdbcFactSource source = + new SqlServerShadowFactReader.JdbcFactSource(jdbc, + Clock.fixed(NOW, ZoneOffset.UTC)); + + assertEquals(NOW.minusMillis(500), + source.readDatabaseClock(NOW.plusMillis(1_001))); + verify(statement).setQueryTimeout(1); + verify(result).getTimestamp( + org.mockito.ArgumentMatchers.eq("database_utc_time"), + org.mockito.ArgumentMatchers.argThat(calendar -> + "UTC".equals(calendar.getTimeZone().getID()))); + verify(result, never()).getTimestamp("database_utc_time"); + verify(connection, never()).prepareStatement( + org.mockito.ArgumentMatchers.contains("dbo.")); + } + + @Test + void jdbcVehicleProjectionCapturesDispatchAndLegacyGateFacts() + throws Exception { + DataSource dataSource = mock(DataSource.class); + Connection connection = mock(Connection.class); + PreparedStatement statement = mock(PreparedStatement.class); + ResultSet result = mock(ResultSet.class); + when(dataSource.getConnection()).thenReturn(connection); + when(connection.prepareStatement( + SqlServerShadowFactReader.JdbcFactSource.VEHICLE_SQL)) + .thenReturn(statement); + when(statement.executeQuery()).thenReturn(result); + when(result.next()).thenReturn(true, false); + when(result.wasNull()).thenReturn(false); + when(result.getString("guid")).thenReturn("V-1"); + when(result.getString("device_id")).thenReturn("1"); + when(result.getString("agv_status")).thenReturn("1"); + when(result.getString("src_site")).thenReturn("070501"); + when(result.getString("tar_site")).thenReturn("070504"); + when(result.getString("curr_site")).thenReturn("070502"); + when(result.getString("path")).thenReturn("070502,070503,070504"); + when(result.getString("spare2")).thenReturn("1"); + when(result.getString("spare5")).thenReturn("1"); + when(result.getInt("task_no")).thenReturn(30001); + when(result.getInt("task_type")).thenReturn(3); + when(result.getInt("battery")).thenReturn(76); + when(result.getInt("work_mode")).thenReturn(5); + when(result.getInt("alarm")).thenReturn(0); + when(result.getInt("status")).thenReturn(1); + when(result.getInt("finish")).thenReturn(0); + when(result.getInt("charge_flag")).thenReturn(0); + when(result.getInt("work_phase")).thenReturn(2); + when(result.getInt("err_msg")).thenReturn(0); + when(result.getInt("change_layer")).thenReturn(1); + when(result.getInt("cargo_flag")).thenReturn(1); + when(result.getInt("allow_send")).thenReturn(1); + when(result.getInt("curr_floor")).thenReturn(2); + when(result.getInt("locked")).thenReturn(0); + when(result.getTimestamp( + org.mockito.ArgumentMatchers.eq("last_time_utc"), + org.mockito.ArgumentMatchers.any(java.util.Calendar.class))) + .thenReturn(Timestamp.from(NOW.minusSeconds(1))); + SqlServerShadowFactReader.JdbcFactSource source = + new SqlServerShadowFactReader.JdbcFactSource( + new JdbcTemplate(dataSource), + Clock.fixed(NOW, ZoneOffset.UTC)); + + ShadowVehicleFact vehicle = source.readVehicles( + NOW.plusMillis(1_001)).get(0); + + assertEquals("1", vehicle.getAgvStatus()); + assertEquals(Integer.valueOf(30001), vehicle.getTaskNo()); + assertEquals("070501", vehicle.getSourceSite()); + assertEquals("070504", vehicle.getTargetSite()); + assertEquals(Integer.valueOf(3), vehicle.getTaskType()); + assertEquals(Integer.valueOf(76), vehicle.getBattery()); + assertEquals(Integer.valueOf(5), vehicle.getWorkMode()); + assertEquals(Integer.valueOf(0), vehicle.getAlarm()); + assertEquals(Integer.valueOf(1), vehicle.getStatus()); + assertEquals(Integer.valueOf(0), vehicle.getFinish()); + assertEquals(Integer.valueOf(0), vehicle.getChargeFlag()); + assertEquals(Integer.valueOf(2), vehicle.getWorkPhase()); + assertEquals(Integer.valueOf(0), vehicle.getErrorCode()); + assertEquals(Integer.valueOf(1), vehicle.getChangeLayer()); + assertEquals(Integer.valueOf(1), vehicle.getCargoFlag()); + assertEquals("070502,070503,070504", vehicle.getPath()); + assertEquals(Integer.valueOf(1), vehicle.getAllowSend()); + assertEquals(Integer.valueOf(2), vehicle.getCurrentFloor()); + assertEquals(Integer.valueOf(0), vehicle.getLocked()); + assertEquals("1", vehicle.getSpare2()); + assertEquals("1", vehicle.getSpare5()); + assertEquals(NOW.minusSeconds(1), vehicle.getCapturedAt()); + verify(statement).setInt(1, 101); + verify(statement).setQueryTimeout(1); + } + + @Test + void concurrentCycleIsRejectedWithoutStartingAnotherTransaction() + throws Exception { + FakeFactSource source = completeSource(); + source.block = new CountDownLatch(1); + source.entered = new CountDownLatch(1); + RecordingTransactions transactions = new RecordingTransactions(); + SqlServerShadowFactReader reader = reader(source, transactions); + AtomicReference backgroundFailure = + new AtomicReference(); + Thread first = new Thread(() -> { + try { + reader.read(ShadowReadCursor.initial(), NOW.plusSeconds(4)); + } catch (Throwable failure) { + backgroundFailure.set(failure); + } + }); + first.start(); + assertTrue(source.entered.await(2, TimeUnit.SECONDS)); + + ShadowFactReadException overlap = assertThrows( + ShadowFactReadException.class, + () -> reader.read(ShadowReadCursor.initial(), + NOW.plusSeconds(4))); + assertEquals("SOURCE_CYCLE_ALREADY_IN_FLIGHT", + overlap.getReasonCode()); + assertEquals(Duration.ZERO, overlap.getRetryAfter()); + + source.block.countDown(); + first.join(2_000L); + assertFalse(first.isAlive()); + assertEquals(null, backgroundFailure.get()); + assertEquals(1, transactions.calls); + } + + @Test + void projectsOrdinaryCrossFloorAndBothTwoStageContractsIndependently() { + LegacyQueueProtocolProjector projector = + new LegacyQueueProtocolProjector(); + + ShadowLegacyQueueResult ordinary = projector.project( + queueBuilder("ORD", NOW).taskNo("30001") + .default02("1").pathInfo("70501,70502,") + .turnPath("").build()); + assertEquals("4004,4000,4002,4021", ordinary.getWriteAddresses()); + assertEquals(Arrays.asList(70501, 70502, 0, 0, 0, 0, 0, 0), + ordinary.getWrites().get(0).getValues()); + + ShadowLegacyQueueResult crossFloor = projector.project( + queueBuilder("CROSS", NOW).taskNo("30002") + .default02("8").default07("8") + .pathInfo("70501,70507,80507,80501,") + .turnPath("70507,80507").build()); + assertEquals("4004,4000,4002,4021,4020,4022", + crossFloor.getWriteAddresses()); + assertEquals(Collections.singletonList(8), + crossFloor.write("4022").getValues()); + + ShadowLegacyQueueResult stageOne = projector.project( + queueBuilder("STAGE-1", NOW).taskNo("30003") + .default02("1").default08("1") + .pathInfo("70501,70503,").turnPath("").build()); + assertEquals("4404,4422,4402,4400", stageOne.getWriteAddresses()); + + ShadowLegacyQueueResult stageTwo = projector.project( + queueBuilder("STAGE-2", NOW).taskNo("30004") + .default02("1").default08("2") + .endSite("070503").default06("070504") + .pathInfo("70501,70503,").build()); + assertEquals("4404,4422,4423,4402,4400", + stageTwo.getWriteAddresses()); + assertEquals(Arrays.asList(70503, 70504, 0, 0, 0, 0, 0, 0), + stageTwo.getWrites().get(0).getValues()); + assertEquals(Collections.singletonList(9), + stageTwo.write("4402").getValues()); + } + + @Test + void missingLegacyFieldsNeverProduceAComparableRegisterSequence() { + LegacyQueueProtocolProjector projector = + new LegacyQueueProtocolProjector(); + ShadowLegacyQueueResult result = projector.project( + queueBuilder("BROKEN", NOW).taskNo(null) + .pathInfo(null).default02("8") + .default07(null).build()); + + assertEquals(ShadowLegacyQueueResult.Status.LEGACY_RESULT_UNAVAILABLE, + result.getStatus()); + assertEquals("LEGACY_RESULT_UNAVAILABLE:MISSING_TASK_NO", + result.getReasonCode()); + assertTrue(result.getWrites().isEmpty()); + } + + @Test + void legacyProjectionUsesTheExecutedPathEndpointsInsteadOfQueueOrigins() { + ShadowQueueFact queue = queueBuilder("REPOSITIONED", NOW) + .startSite("070501").endSite("070504") + .pathInfo("070502,070503,070504,") + .turnPath("070503").build(); + + ShadowLegacyQueueResult result = + new LegacyQueueProtocolProjector().project(queue); + + assertEquals(Arrays.asList(70502, 70504, 70503, 0, 0, 0, 0, 0), + result.write("4004").getValues()); + } + + @Test + void legacyProjectionAcceptsPersistedZeroPaddedTurnPoints() { + ShadowQueueFact queue = queueBuilder("PADDED-TURNS", NOW) + .pathInfo("070501,070502,070504,") + .turnPath("070503,0,0,0,0,0").build(); + + ShadowLegacyQueueResult result = + new LegacyQueueProtocolProjector().project(queue); + + assertEquals(ShadowLegacyQueueResult.Status.AVAILABLE, + result.getStatus()); + assertEquals(Arrays.asList(70501, 70504, 70503, 0, 0, 0, 0, 0), + result.write("4004").getValues()); + } + + private static SqlServerShadowFactReader reader(FakeFactSource source, + RecordingTransactions tx) { + return new SqlServerShadowFactReader(source, tx.template, + Clock.fixed(NOW, ZoneOffset.UTC)); + } + + private static FakeFactSource completeSource() { + FakeFactSource source = new FakeFactSource(); + source.clockValues.add(NOW.minusMillis(500)); + source.clockValues.add(NOW.minusMillis(100)); + source.queues = Collections.singletonList(queue("Q-1", NOW.minusSeconds(2))); + source.vehicles = Collections.singletonList(vehicle("1", "070502")); + source.maps = Arrays.asList(map("070501"), map("070502")); + return source; + } + + private static ShadowQueueFact queue(String guid, Instant createdAt) { + return queueBuilder(guid, createdAt).build(); + } + + private static ShadowQueueFact.Builder queueBuilder(String guid, + Instant createdAt) { + return ShadowQueueFact.builder("shadow_ro.v_wcs_srgv_queue", guid) + .taskNo("30001").taskType("2").priority(1L) + .cargoPickSite("070501").cargoPlaceSite("070502") + .startSite("070501").endSite("070502") + .createdAt(createdAt).agvId("1") + .pathInfo("70501,70502,").turnPath("") + .default02("1").default03("").default04("") + .default05("").default06("").default07("") + .default08(""); + } + + private static ShadowVehicleFact vehicle(String deviceId, + String currentSite) { + return completeVehicleBuilder(deviceId, currentSite).build(); + } + + private static ShadowVehicleFact.Builder completeVehicleBuilder( + String deviceId, String currentSite) { + return ShadowVehicleFact.builder("V-" + deviceId, deviceId) + .agvStatus("1").taskNo(0) + .sourceSite(currentSite).targetSite(currentSite) + .currentSite(currentSite).taskType(0) + .battery(76).workMode(5).alarm(0).status(1) + .finish(0).chargeFlag(0).workPhase(0).errorCode(0) + .changeLayer(0).cargoFlag(0).path(currentSite) + .allowSend(1).currentFloor(2).locked(0) + .spare2("0").spare5("0") + .capturedAt(NOW.minusSeconds(1)); + } + + private static ShadowMapFact map(String siteId) { + return new ShadowMapFact("M-" + siteId, siteId, 1, 1, 1, 1); + } + + private static List repeatedQueues(int count) { + List facts = new ArrayList(); + for (int i = 0; i < count; i++) { + facts.add(queue("Q-" + i, NOW.minusSeconds(1))); + } + return facts; + } + + private static List repeatedVehicles(int count) { + List facts = new ArrayList(); + for (int i = 0; i < count; i++) { + facts.add(vehicle(Integer.toString(i), "070502")); + } + return facts; + } + + private static List repeatedMaps(int count) { + List facts = new ArrayList(); + for (int i = 0; i < count; i++) { + facts.add(map(Integer.toString(70_000 + i))); + } + return facts; + } + + private static List guids(List facts) { + List result = new ArrayList(); + for (ShadowQueueFact fact : facts) { + result.add(fact.getGuid()); + } + return result; + } + + private static List deviceIds(List facts) { + List result = new ArrayList(); + for (ShadowVehicleFact fact : facts) { + result.add(fact.getDeviceId()); + } + return result; + } + + private static List siteIds(List facts) { + List result = new ArrayList(); + for (ShadowMapFact fact : facts) { + result.add(fact.getSiteId()); + } + return result; + } + + private static final class SequenceClock extends Clock { + private final Instant[] values; + private int index; + + private SequenceClock(Instant... values) { + this.values = values; + } + + @Override + public ZoneId getZone() { + return ZoneOffset.UTC; + } + + @Override + public Clock withZone(ZoneId zone) { + if (!ZoneOffset.UTC.equals(zone)) { + throw new IllegalArgumentException("only UTC is supported"); + } + return this; + } + + @Override + public Instant instant() { + int current = Math.min(index, values.length - 1); + index++; + return values[current]; + } + } + + private static final class FakeFactSource + implements SqlServerShadowFactReader.FactSource { + private final List clockValues = new ArrayList(); + private List queues = Collections.emptyList(); + private List vehicles = Collections.emptyList(); + private List maps = Collections.emptyList(); + private ShadowReadCursor cursor; + private String failureReason; + private int queueCalls; + private CountDownLatch entered; + private CountDownLatch block; + + @Override + public Instant readDatabaseClock(Instant deadline) { + if (clockValues.size() > 1) { + return clockValues.remove(0); + } + return clockValues.isEmpty() ? NOW : clockValues.get(0); + } + + @Override + public List readQueues(ShadowReadCursor cursor, + Instant deadline) { + this.cursor = cursor; + queueCalls++; + if (entered != null) { + entered.countDown(); + } + if (block != null) { + try { + if (!block.await(2, TimeUnit.SECONDS)) { + throw new IllegalStateException("test latch timeout"); + } + } catch (InterruptedException interrupted) { + Thread.currentThread().interrupt(); + throw new IllegalStateException(interrupted); + } + } + if (failureReason != null) { + throw new ShadowFactReadException(failureReason, + Duration.ZERO); + } + return queues; + } + + @Override + public List readVehicles(Instant deadline) { + return vehicles; + } + + @Override + public List readMaps(Instant deadline) { + return maps; + } + } + + private static final class RecordingTransactions { + private final PlatformTransactionManager manager = + mock(PlatformTransactionManager.class); + private final TransactionTemplate template = + new TransactionTemplate(manager); + private int calls; + private boolean readOnly; + + private RecordingTransactions() { + when(manager.getTransaction(any(TransactionDefinition.class))) + .thenAnswer(invocation -> { + TransactionDefinition definition = invocation.getArgument(0); + calls++; + readOnly = definition.isReadOnly(); + return new SimpleTransactionStatus(); + }); + template.setReadOnly(true); + } + } +} diff --git a/sql/shadow/V001__ft546_shadow_read_views.sql b/sql/shadow/V001__ft546_shadow_read_views.sql index aa2761e..d6745ac 100644 --- a/sql/shadow/V001__ft546_shadow_read_views.sql +++ b/sql/shadow/V001__ft546_shadow_read_views.sql @@ -1,4 +1,5 @@ :setvar ShadowSourceDatabase "FT546-wcs" +:setvar ShadowSourceUtcOffset "+08:00" SET NOCOUNT ON; SET XACT_ABORT ON; @@ -18,7 +19,11 @@ SELECT guid, task_no, task_type, priority, cargo, issued, path_info, execution_status, cargo_pick_site, cargo_place_site, start_site, end_site, the_floor, trun_path, up_down_site, create_date, start_date, finish_date, agv_id, default02, default03, default04, - default05, default06, default07, default08 + default05, default06, default07, default08, + CONVERT(datetime2(7), SWITCHOFFSET(TODATETIMEOFFSET( + TRY_CONVERT(datetime2(7), create_date, 120), + N'$(ShadowSourceUtcOffset)'), '+00:00')) + AS create_date_utc FROM dbo.wcs_srgv_queue; GO @@ -28,7 +33,11 @@ SELECT guid, task_no, task_type, priority, cargo, issued, path_info, execution_status, cargo_pick_site, cargo_place_site, start_site, end_site, the_floor, trun_path, up_down_site, create_date, start_date, finish_date, agv_id, default02, default03, default04, - default05, default06, default07, default08 + default05, default06, default07, default08, + CONVERT(datetime2(7), SWITCHOFFSET(TODATETIMEOFFSET( + TRY_CONVERT(datetime2(7), create_date, 120), + N'$(ShadowSourceUtcOffset)'), '+00:00')) + AS create_date_utc FROM dbo.wcs_srgv_queue_finish; GO @@ -39,7 +48,11 @@ SELECT guid, device_id, device_name, type, set_floor, power_time, task_type, battery, work_mode, alarm, status, finish, charge_flag, work_phase, err_msg, ssj_request, change_layer, cargo_flag, manual_charge, path, allow_send, curr_floor, offline_site, locked, - spare1, spare2, spare3, spare4, spare5 + spare1, spare2, spare3, spare4, spare5, + CONVERT(datetime2(7), SWITCHOFFSET(TODATETIMEOFFSET( + TRY_CONVERT(datetime2(7), last_time, 120), + N'$(ShadowSourceUtcOffset)'), '+00:00')) + AS last_time_utc FROM dbo.device_srgv; GO @@ -71,7 +84,7 @@ GO CREATE OR ALTER VIEW shadow_ro.v_current_database_clock AS -SELECT SYSUTCDATETIME() AS database_utc_time, +SELECT CONVERT(datetime2(7), SYSUTCDATETIME()) AS database_utc_time, DB_NAME() AS database_name, @@SERVERNAME AS server_name; GO