feat(wcs): coordinate and persist traffic decisions
This commit is contained in:
@@ -0,0 +1,140 @@
|
||||
package com.ruoyi.wcs.application.dispatch;
|
||||
|
||||
import com.ruoyi.wcs.domain.dispatch.DefaultVehicleAssignmentPolicy;
|
||||
import com.ruoyi.wcs.domain.dispatch.DispatchTask;
|
||||
import com.ruoyi.wcs.domain.equipment.DeviceCapability;
|
||||
import com.ruoyi.wcs.domain.equipment.DeviceDescriptor;
|
||||
import com.ruoyi.wcs.domain.equipment.DeviceType;
|
||||
import com.ruoyi.wcs.domain.equipment.VehicleSnapshot;
|
||||
import com.ruoyi.wcs.domain.shared.IdGenerator;
|
||||
import com.ruoyi.wcs.domain.topology.EffectiveTopology;
|
||||
import com.ruoyi.wcs.domain.topology.ResourceDefinition;
|
||||
import com.ruoyi.wcs.domain.topology.TopologyCompiler;
|
||||
import com.ruoyi.wcs.domain.topology.TopologyDefinition;
|
||||
import com.ruoyi.wcs.domain.topology.TopologyEdge;
|
||||
import com.ruoyi.wcs.domain.topology.TopologyNode;
|
||||
import com.ruoyi.wcs.domain.traffic.AStarRoutePlanner;
|
||||
import com.ruoyi.wcs.domain.traffic.DeadlockGuard;
|
||||
import com.ruoyi.wcs.domain.traffic.ResourceLedger;
|
||||
import com.ruoyi.wcs.domain.traffic.WaitForGraph;
|
||||
import com.ruoyi.wcs.port.out.DispatchPolicy;
|
||||
import com.ruoyi.wcs.port.out.TrafficStateRepository;
|
||||
import org.junit.jupiter.api.Test;
|
||||
|
||||
import java.time.Instant;
|
||||
import java.util.Arrays;
|
||||
import java.util.Collections;
|
||||
import java.util.EnumSet;
|
||||
import java.util.concurrent.atomic.AtomicInteger;
|
||||
|
||||
import static org.junit.jupiter.api.Assertions.assertEquals;
|
||||
import static org.junit.jupiter.api.Assertions.assertFalse;
|
||||
import static org.junit.jupiter.api.Assertions.assertTrue;
|
||||
|
||||
class DispatchCoordinatorTest {
|
||||
|
||||
@Test
|
||||
void grantsOneVehicleAndHoldsTheOtherForTheSameIntersection() {
|
||||
EffectiveTopology topology = topology();
|
||||
RecordingTrafficRepository repository = new RecordingTrafficRepository(true);
|
||||
DispatchCoordinator coordinator = coordinator(repository);
|
||||
|
||||
DispatchCycleResult result = coordinator.dispatch(new DispatchSnapshot(
|
||||
Arrays.asList(task("O1", "S1", "A", 2),
|
||||
task("O2", "S2", "B", 1)),
|
||||
Arrays.asList(vehicle("V1", "A"), vehicle("V2", "B")),
|
||||
topology, new ResourceLedger(topology), 11L));
|
||||
|
||||
assertEquals(1, result.getGranted().size());
|
||||
assertEquals("S1", result.getGranted().get(0).getStepId());
|
||||
assertEquals("V1", result.getGranted().get(0).getVehicleId());
|
||||
assertEquals(Collections.singletonList("R-1"),
|
||||
result.getGranted().get(0).getReservationIds());
|
||||
assertEquals(1, result.getHeld().size());
|
||||
assertEquals("NO_SAFE_YIELD_POINT", result.getHeld().get(0).getReasonCode());
|
||||
assertEquals(0, result.getInvariantViolations().size());
|
||||
assertEquals(1, repository.commitCount);
|
||||
assertEquals("FT546", repository.lastCommit.getWarehouseCode());
|
||||
assertEquals(2, repository.lastCommit.getDecisionEvents().size());
|
||||
assertFalse(result.isRetrySnapshotChanged());
|
||||
}
|
||||
|
||||
@Test
|
||||
void returnsRetryWithoutExposingPartialGrantWhenSnapshotChanged() {
|
||||
EffectiveTopology topology = topology();
|
||||
RecordingTrafficRepository repository = new RecordingTrafficRepository(false);
|
||||
DispatchCoordinator coordinator = coordinator(repository);
|
||||
|
||||
DispatchCycleResult result = coordinator.dispatch(new DispatchSnapshot(
|
||||
Collections.singletonList(task("O1", "S1", "A", 1)),
|
||||
Collections.singletonList(vehicle("V1", "A")),
|
||||
topology, new ResourceLedger(topology), 11L));
|
||||
|
||||
assertTrue(result.isRetrySnapshotChanged());
|
||||
assertEquals("RETRY_SNAPSHOT_CHANGED", result.getStatusCode());
|
||||
assertEquals(0, result.getGranted().size());
|
||||
assertEquals(0, result.getHeld().size());
|
||||
assertEquals(1, repository.commitCount);
|
||||
}
|
||||
|
||||
private static DispatchCoordinator coordinator(TrafficStateRepository repository) {
|
||||
AStarRoutePlanner routePlanner = new AStarRoutePlanner();
|
||||
AtomicInteger sequence = new AtomicInteger();
|
||||
IdGenerator ids = () -> "R-" + sequence.incrementAndGet();
|
||||
return new DispatchCoordinator(
|
||||
new DefaultVehicleAssignmentPolicy(routePlanner, DispatchPolicy.NONE),
|
||||
new DeadlockGuard(new WaitForGraph(), routePlanner),
|
||||
repository, ids);
|
||||
}
|
||||
|
||||
private static DispatchTask task(String orderId, String stepId,
|
||||
String source, int priority) {
|
||||
return new DispatchTask(orderId, stepId, source, "X", priority,
|
||||
60L, 30, Collections.<DeviceCapability>emptySet());
|
||||
}
|
||||
|
||||
private static VehicleSnapshot vehicle(String id, String location) {
|
||||
DeviceDescriptor descriptor = new DeviceDescriptor(id,
|
||||
DeviceType.FOUR_WAY_SHUTTLE, "F1",
|
||||
EnumSet.of(DeviceCapability.FOUR_WAY_MOVE));
|
||||
return new VehicleSnapshot(descriptor,
|
||||
VehicleSnapshot.ConnectionState.ONLINE,
|
||||
VehicleSnapshot.OperatingMode.AUTO,
|
||||
VehicleSnapshot.HealthState.NORMAL,
|
||||
VehicleSnapshot.OccupationState.IDLE,
|
||||
location, null, 80, null, "F1",
|
||||
Instant.parse("2026-08-04T00:00:00Z"));
|
||||
}
|
||||
|
||||
private static EffectiveTopology topology() {
|
||||
TopologyDefinition definition = new TopologyDefinition("FT546", 7L, true,
|
||||
Arrays.asList(
|
||||
new TopologyNode("A", "F1", 0, 0, 1, false, false),
|
||||
new TopologyNode("X", "F1", 1, 0, 1, false, false),
|
||||
new TopologyNode("B", "F1", 2, 0, 1, false, false)),
|
||||
Arrays.asList(
|
||||
new TopologyEdge("A", "X", 1, 1),
|
||||
new TopologyEdge("B", "X", 1, 1)),
|
||||
Collections.<ResourceDefinition>emptyList(),
|
||||
Collections.<String>emptyList(), Collections.<String>emptyList());
|
||||
return new TopologyCompiler().compile(definition);
|
||||
}
|
||||
|
||||
private static final class RecordingTrafficRepository
|
||||
implements TrafficStateRepository {
|
||||
private final boolean commitResult;
|
||||
private int commitCount;
|
||||
private DispatchCommit lastCommit;
|
||||
|
||||
private RecordingTrafficRepository(boolean commitResult) {
|
||||
this.commitResult = commitResult;
|
||||
}
|
||||
|
||||
@Override
|
||||
public boolean commit(DispatchCommit commit) {
|
||||
commitCount++;
|
||||
lastCommit = commit;
|
||||
return commitResult;
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,248 @@
|
||||
package com.ruoyi.wcs.infrastructure.persistence.traffic;
|
||||
|
||||
import org.junit.jupiter.api.AfterEach;
|
||||
import org.junit.jupiter.api.Assumptions;
|
||||
import org.junit.jupiter.api.Tag;
|
||||
import org.junit.jupiter.api.Test;
|
||||
|
||||
import java.nio.charset.StandardCharsets;
|
||||
import java.nio.file.Files;
|
||||
import java.nio.file.Path;
|
||||
import java.nio.file.Paths;
|
||||
import java.sql.Connection;
|
||||
import java.sql.DriverManager;
|
||||
import java.sql.PreparedStatement;
|
||||
import java.sql.ResultSet;
|
||||
import java.sql.SQLException;
|
||||
import java.sql.Statement;
|
||||
import java.sql.Timestamp;
|
||||
import java.time.Instant;
|
||||
import java.util.UUID;
|
||||
import java.util.concurrent.CountDownLatch;
|
||||
import java.util.concurrent.ExecutorService;
|
||||
import java.util.concurrent.Executors;
|
||||
import java.util.concurrent.Future;
|
||||
|
||||
import static org.junit.jupiter.api.Assertions.assertEquals;
|
||||
import static org.junit.jupiter.api.Assertions.assertFalse;
|
||||
|
||||
@Tag("sqlserver")
|
||||
class TrafficConcurrencyIT {
|
||||
private String jdbcUrl;
|
||||
private String username;
|
||||
private String password;
|
||||
private String warehouseCode;
|
||||
private String floorId;
|
||||
private String resourceKey;
|
||||
|
||||
@Test
|
||||
void packagesTheTrafficMigrationUsedByTheSqlServerVerification() throws Exception {
|
||||
String root = System.getProperty("maven.multiModuleProjectDirectory", ".");
|
||||
String migration = readMigration(root, "V002__wcs_topology_and_traffic.sql");
|
||||
|
||||
assertFalse(migration.trim().isEmpty());
|
||||
}
|
||||
|
||||
@AfterEach
|
||||
void cleanFixture() throws Exception {
|
||||
if (jdbcUrl == null || resourceKey == null) {
|
||||
return;
|
||||
}
|
||||
try (Connection connection = connect()) {
|
||||
delete(connection, "DELETE FROM dbo.wcs_resource_reservation WHERE resource_key = ?", resourceKey);
|
||||
delete(connection, "DELETE FROM dbo.wcs_resource_state WHERE resource_key = ?", resourceKey);
|
||||
delete(connection, "DELETE FROM dbo.wcs_resource_definition WHERE resource_key = ?", resourceKey);
|
||||
delete(connection, "DELETE FROM dbo.wcs_topology_version WHERE warehouse_code = ?", warehouseCode);
|
||||
delete(connection, "DELETE FROM dbo.wcs_floor WHERE floor_id = ?", floorId);
|
||||
delete(connection, "DELETE FROM dbo.wcs_warehouse WHERE warehouse_code = ?", warehouseCode);
|
||||
}
|
||||
}
|
||||
|
||||
@Test
|
||||
void onlyOneConcurrentReservationCommitsForCapacityOneResource() throws Exception {
|
||||
configureIsolatedDatabase();
|
||||
applyMigration();
|
||||
createCapacityOneFixture();
|
||||
|
||||
ExecutorService executor = Executors.newFixedThreadPool(2);
|
||||
CountDownLatch ready = new CountDownLatch(2);
|
||||
CountDownLatch start = new CountDownLatch(1);
|
||||
try {
|
||||
Future<Boolean> first = executor.submit(() -> reserve("IT-R1", "IT-V1", ready, start));
|
||||
Future<Boolean> second = executor.submit(() -> reserve("IT-R2", "IT-V2", ready, start));
|
||||
ready.await();
|
||||
start.countDown();
|
||||
|
||||
int committed = (first.get() ? 1 : 0) + (second.get() ? 1 : 0);
|
||||
assertEquals(1, committed);
|
||||
assertEquals(1, scalarInt(
|
||||
"SELECT reserved_count FROM dbo.wcs_resource_state WHERE resource_key = ?",
|
||||
resourceKey));
|
||||
assertEquals(1, scalarInt(
|
||||
"SELECT COUNT(*) FROM dbo.wcs_resource_reservation "
|
||||
+ "WHERE resource_key = ? AND reservation_state = 'GRANTED'",
|
||||
resourceKey));
|
||||
} finally {
|
||||
executor.shutdownNow();
|
||||
}
|
||||
}
|
||||
|
||||
private void configureIsolatedDatabase() {
|
||||
Assumptions.assumeTrue("true".equalsIgnoreCase(
|
||||
System.getenv("WCS_SQLSERVER_TEST_ENABLED")),
|
||||
"set WCS_SQLSERVER_TEST_ENABLED=true only for an isolated SQL Server database");
|
||||
jdbcUrl = requireEnvironment("WCS_SQLSERVER_TEST_URL");
|
||||
username = requireEnvironment("WCS_SQLSERVER_TEST_USER");
|
||||
password = requireEnvironment("WCS_SQLSERVER_TEST_PASSWORD");
|
||||
}
|
||||
|
||||
private void applyMigration() throws Exception {
|
||||
String root = System.getProperty("maven.multiModuleProjectDirectory", ".");
|
||||
try (Connection connection = connect(); Statement statement = connection.createStatement()) {
|
||||
statement.execute(readMigration(root, "V001__wcs_task_foundation.sql"));
|
||||
statement.execute(readMigration(root, "V002__wcs_topology_and_traffic.sql"));
|
||||
}
|
||||
}
|
||||
|
||||
private static String readMigration(String root, String filename) throws Exception {
|
||||
Path current = Paths.get(root).toAbsolutePath().normalize();
|
||||
while (current != null) {
|
||||
Path migration = current.resolve(Paths.get("sql", "wcs", filename));
|
||||
if (Files.isRegularFile(migration)) {
|
||||
return new String(Files.readAllBytes(migration), StandardCharsets.UTF_8);
|
||||
}
|
||||
current = current.getParent();
|
||||
}
|
||||
throw new java.nio.file.NoSuchFileException(filename);
|
||||
}
|
||||
|
||||
private void createCapacityOneFixture() throws Exception {
|
||||
String suffix = UUID.randomUUID().toString().replace("-", "");
|
||||
warehouseCode = "IT" + suffix;
|
||||
floorId = "IT-F-" + suffix;
|
||||
resourceKey = "IT:RESOURCE:" + suffix;
|
||||
try (Connection connection = connect()) {
|
||||
insert(connection,
|
||||
"INSERT INTO dbo.wcs_warehouse "
|
||||
+ "(warehouse_code, warehouse_name, active_configuration_version, state_version) "
|
||||
+ "VALUES (?, ?, NULL, 0)",
|
||||
warehouseCode, "traffic concurrency test");
|
||||
insert(connection,
|
||||
"INSERT INTO dbo.wcs_floor "
|
||||
+ "(floor_id, warehouse_code, floor_code, floor_level) VALUES (?, ?, ?, 1)",
|
||||
floorId, warehouseCode, "IT-F1");
|
||||
insert(connection,
|
||||
"INSERT INTO dbo.wcs_topology_version "
|
||||
+ "(warehouse_code, topology_version, publication_state, published_time) "
|
||||
+ "VALUES (?, 1, 'PUBLISHED', ?)",
|
||||
warehouseCode, Instant.now());
|
||||
insert(connection,
|
||||
"INSERT INTO dbo.wcs_resource_definition "
|
||||
+ "(resource_key, warehouse_code, topology_version, resource_type, capacity) "
|
||||
+ "VALUES (?, ?, 1, 'NODE', 1)",
|
||||
resourceKey, warehouseCode);
|
||||
insert(connection,
|
||||
"INSERT INTO dbo.wcs_resource_state "
|
||||
+ "(warehouse_code, topology_version, resource_key, capacity, "
|
||||
+ "reserved_count, occupied_count, fencing_token, state_version) "
|
||||
+ "VALUES (?, 1, ?, 1, 0, 0, 0, 0)",
|
||||
warehouseCode, resourceKey);
|
||||
}
|
||||
}
|
||||
|
||||
private boolean reserve(String reservationId, String vehicleId,
|
||||
CountDownLatch ready, CountDownLatch start) throws Exception {
|
||||
try (Connection connection = connect()) {
|
||||
connection.setAutoCommit(false);
|
||||
connection.setTransactionIsolation(Connection.TRANSACTION_READ_COMMITTED);
|
||||
ready.countDown();
|
||||
start.await();
|
||||
try {
|
||||
int claimed;
|
||||
try (PreparedStatement statement = connection.prepareStatement(
|
||||
"UPDATE dbo.wcs_resource_state WITH (UPDLOCK, ROWLOCK) "
|
||||
+ "SET reserved_count = reserved_count + 1, "
|
||||
+ "fencing_token = fencing_token + 1, state_version = state_version + 1 "
|
||||
+ "WHERE warehouse_code = ? AND topology_version = 1 "
|
||||
+ "AND resource_key = ? "
|
||||
+ "AND reserved_count + occupied_count < capacity")) {
|
||||
statement.setString(1, warehouseCode);
|
||||
statement.setString(2, resourceKey);
|
||||
claimed = statement.executeUpdate();
|
||||
}
|
||||
if (claimed == 0) {
|
||||
connection.rollback();
|
||||
return false;
|
||||
}
|
||||
try (PreparedStatement statement = connection.prepareStatement(
|
||||
"INSERT INTO dbo.wcs_resource_reservation "
|
||||
+ "(reservation_id, warehouse_code, topology_version, resource_key, "
|
||||
+ "order_id, step_id, vehicle_id, reservation_state, fencing_token, "
|
||||
+ "created_time, updated_time) "
|
||||
+ "SELECT ?, ?, 1, ?, 'IT-O', 'IT-S', ?, 'GRANTED', fencing_token, ?, ? "
|
||||
+ "FROM dbo.wcs_resource_state WHERE warehouse_code = ? "
|
||||
+ "AND topology_version = 1 AND resource_key = ?")) {
|
||||
Instant now = Instant.now();
|
||||
statement.setString(1, reservationId + '-' + warehouseCode);
|
||||
statement.setString(2, warehouseCode);
|
||||
statement.setString(3, resourceKey);
|
||||
statement.setString(4, vehicleId);
|
||||
statement.setTimestamp(5, Timestamp.from(now));
|
||||
statement.setTimestamp(6, Timestamp.from(now));
|
||||
statement.setString(7, warehouseCode);
|
||||
statement.setString(8, resourceKey);
|
||||
statement.executeUpdate();
|
||||
}
|
||||
connection.commit();
|
||||
return true;
|
||||
} catch (Exception error) {
|
||||
connection.rollback();
|
||||
throw error;
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
private int scalarInt(String sql, String value) throws Exception {
|
||||
try (Connection connection = connect();
|
||||
PreparedStatement statement = connection.prepareStatement(sql)) {
|
||||
statement.setString(1, value);
|
||||
try (ResultSet result = statement.executeQuery()) {
|
||||
if (!result.next()) {
|
||||
throw new AssertionError("query returned no rows: " + sql);
|
||||
}
|
||||
return result.getInt(1);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
private Connection connect() throws SQLException {
|
||||
return DriverManager.getConnection(jdbcUrl, username, password);
|
||||
}
|
||||
|
||||
private static void insert(Connection connection, String sql,
|
||||
Object... values) throws SQLException {
|
||||
try (PreparedStatement statement = connection.prepareStatement(sql)) {
|
||||
for (int index = 0; index < values.length; index++) {
|
||||
Object value = values[index];
|
||||
statement.setObject(index + 1,
|
||||
value instanceof Instant ? Timestamp.from((Instant) value) : value);
|
||||
}
|
||||
statement.executeUpdate();
|
||||
}
|
||||
}
|
||||
|
||||
private static void delete(Connection connection, String sql,
|
||||
String value) throws SQLException {
|
||||
try (PreparedStatement statement = connection.prepareStatement(sql)) {
|
||||
statement.setString(1, value);
|
||||
statement.executeUpdate();
|
||||
}
|
||||
}
|
||||
|
||||
private static String requireEnvironment(String name) {
|
||||
String value = System.getenv(name);
|
||||
Assumptions.assumeTrue(value != null && !value.trim().isEmpty(),
|
||||
name + " must point to the isolated SQL Server test database");
|
||||
return value.trim();
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user