fix(lastleg): scope inventory holds by cycle and align pick lock ordering
This commit is contained in:
+2
@@ -117,6 +117,8 @@ public class DispatchPdaWorkflowService {
|
||||
}
|
||||
long picked = picks.totalItem(customerId, task.getId(), itemId);
|
||||
if (picked + request.packages() > item.getPackageCount()) throw conflict("Picked packages exceed dispatch pallet packages");
|
||||
pickAttribution.preparePicking(customerId, request.positionId(),
|
||||
"RELABEL_ORDER".equals(item.getSourceBusinessType()) ? item.getOrderId() : null);
|
||||
String keyBase = "DISPATCH:" + order.getDispatchNo() + ":" + itemId + ":" + request.idempotencyKey();
|
||||
InventoryCommandRequests.Move move = new InventoryCommandRequests.Move(request.positionId(),
|
||||
request.packages(), order.getDispatchNo(), order.getDispatchNo(), null,
|
||||
|
||||
+19
@@ -25,6 +25,25 @@ public class DispatchPickOrderAttributionService {
|
||||
@Resource private LastlegOrderMapper orders;
|
||||
@Resource private LastlegOrderStatusHistoryMapper history;
|
||||
|
||||
/** Acquire source order locks before any inventory lock, matching order hold/release. */
|
||||
@org.springframework.transaction.annotation.Transactional(propagation = org.springframework.transaction.annotation.Propagation.MANDATORY)
|
||||
public void preparePicking(String customerId, long positionId, Long fallbackOrderId) {
|
||||
var sourceIds = new java.util.TreeSet<Long>();
|
||||
attributions.availableReceipts(customerId, positionId).forEach(receipt -> sourceIds.add(receipt.getOrderId()));
|
||||
if (fallbackOrderId != null && fallbackOrderId > 0) sourceIds.add(fallbackOrderId);
|
||||
if (sourceIds.isEmpty()) throw conflict("PDA pick has no attributable source order");
|
||||
for (Long orderId : sourceIds) {
|
||||
LastlegOrder order = orders.lockById(customerId, orderId);
|
||||
if (order == null || Boolean.TRUE.equals(order.getDetained())) {
|
||||
throw conflict("PDA pick source order is missing or held");
|
||||
}
|
||||
}
|
||||
if (attributions.availableReceipts(customerId, positionId).stream()
|
||||
.anyMatch(receipt -> !sourceIds.contains(receipt.getOrderId()))) {
|
||||
throw conflict("PDA pick source orders changed; reload the task");
|
||||
}
|
||||
}
|
||||
|
||||
/** Call after the inventory position has been locked/moved and the PDA pick row has been inserted. */
|
||||
public List<OrderPackages> attribute(String customerId, DispatchPdaPick pick) {
|
||||
return attribute(customerId, pick, null);
|
||||
|
||||
+10
-8
@@ -144,23 +144,25 @@ public class InventoryCommandService {
|
||||
|
||||
@Transactional
|
||||
public void holdInboundOrderInventory(String customerId, String operatorId, long inboundOrderId,
|
||||
String businessOrderNo, String remark) {
|
||||
String businessOrderNo, String remark, String holdCycle) {
|
||||
for (InventoryPosition position : commands.lockInboundAvailablePositions(customerId, inboundOrderId)) {
|
||||
long packages = value(position.getAvailablePackages());
|
||||
if (packages == 0) continue;
|
||||
hold(customerId, operatorId, new InventoryCommandRequests.Move(position.getId(), packages,
|
||||
businessOrderNo, null, null, "ORDER_HOLD:" + inboundOrderId + ":" + position.getId(), remark));
|
||||
businessOrderNo, null, holdCycle, holdCycle + ":" + position.getId(), remark));
|
||||
}
|
||||
}
|
||||
|
||||
@Transactional
|
||||
public void releaseInboundOrderHold(String customerId, String operatorId, long inboundOrderId,
|
||||
String businessOrderNo, String remark) {
|
||||
for (InventoryPosition position : commands.lockInboundHeldPositions(customerId, inboundOrderId)) {
|
||||
long packages = value(position.getHoldPackages());
|
||||
if (packages == 0) continue;
|
||||
release(customerId, operatorId, new InventoryCommandRequests.Move(position.getId(), packages,
|
||||
businessOrderNo, null, null, "ORDER_HOLD_RELEASE:" + inboundOrderId + ":" + position.getId(), remark));
|
||||
String businessOrderNo, String remark, String holdCycle) {
|
||||
// HOLD is shared by several business flows. Only release facts owned by this order hold cycle.
|
||||
for (InventoryLedger held : ledgers.orderHoldCycle(customerId, inboundOrderId, holdCycle)) {
|
||||
if (held.getPositionId() == null || value(held.getDeltaPackages()) <= 0) {
|
||||
throw conflict("Order hold inventory fact is invalid");
|
||||
}
|
||||
release(customerId, operatorId, new InventoryCommandRequests.Move(held.getPositionId(), held.getDeltaPackages(),
|
||||
businessOrderNo, null, holdCycle, holdCycle + ":RELEASE:" + held.getPositionId(), remark));
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
+5
@@ -12,6 +12,11 @@ import java.util.List;
|
||||
|
||||
@Mapper
|
||||
public interface InventoryLedgerMapper extends BaseMapper<InventoryLedger> {
|
||||
@Select("SELECT * FROM ll_inventory_ledger WHERE customer_id=#{customerId} AND inbound_order_id=#{inboundOrderId} " +
|
||||
"AND business_type='HOLD' AND source_event_id=#{holdCycle} ORDER BY position_id,id")
|
||||
List<InventoryLedger> orderHoldCycle(@Param("customerId") String customerId,
|
||||
@Param("inboundOrderId") long inboundOrderId, @Param("holdCycle") String holdCycle);
|
||||
|
||||
@Select("SELECT * FROM ll_inventory_ledger WHERE customer_id=#{customerId} AND idempotency_key=#{idempotencyKey} LIMIT 1")
|
||||
InventoryLedger findIdempotency(@Param("customerId") String customerId,
|
||||
@Param("idempotencyKey") String idempotencyKey);
|
||||
|
||||
+11
-2
@@ -78,6 +78,15 @@ public class LastlegOrderCommandService {
|
||||
|| text(order.getOutboundOrderNo()) != null)) {
|
||||
throw conflict("尾程订单已进入出库流程，不能新增暂扣");
|
||||
}
|
||||
String holdCycle = "ORDER_HOLD:" + orderId + ":" + (version + 1);
|
||||
if (!held) {
|
||||
LastlegOrderHoldHistory original = holdHistory.selectOne(new LambdaQueryWrapper<LastlegOrderHoldHistory>()
|
||||
.eq(LastlegOrderHoldHistory::getCustomerId, order.getCustomerId())
|
||||
.eq(LastlegOrderHoldHistory::getOrderId, orderId).eq(LastlegOrderHoldHistory::getHeld, true)
|
||||
.orderByDesc(LastlegOrderHoldHistory::getOrderVersion).last("LIMIT 1"));
|
||||
if (original == null) throw conflict("暂扣来源记录缺失，不能释放其他业务冻结库存");
|
||||
holdCycle = "ORDER_HOLD:" + orderId + ":" + original.getOrderVersion();
|
||||
}
|
||||
LocalDateTime now = LocalDateTime.now();
|
||||
boolean chargeable = held && request.chargeable();
|
||||
String reason = request.reason().trim();
|
||||
@@ -86,10 +95,10 @@ public class LastlegOrderCommandService {
|
||||
}
|
||||
if (held) {
|
||||
inventory.holdInboundOrderInventory(order.getCustomerId(), operatorId, orderId,
|
||||
order.getOrderNo(), reason);
|
||||
order.getOrderNo(), reason, holdCycle);
|
||||
} else {
|
||||
inventory.releaseInboundOrderHold(order.getCustomerId(), operatorId, orderId,
|
||||
order.getOrderNo(), reason);
|
||||
order.getOrderNo(), reason, holdCycle);
|
||||
}
|
||||
int nextVersion = version + 1;
|
||||
LastlegOrderCommandDtos.HoldResponse response = new LastlegOrderCommandDtos.HoldResponse(
|
||||
|
||||
+28
@@ -23,6 +23,7 @@ import java.util.Arrays;
|
||||
|
||||
import static org.junit.jupiter.api.Assertions.assertEquals;
|
||||
import static org.junit.jupiter.api.Assertions.assertTrue;
|
||||
import static org.junit.jupiter.api.Assertions.assertThrows;
|
||||
import static org.mockito.ArgumentMatchers.*;
|
||||
import static org.mockito.Mockito.*;
|
||||
|
||||
@@ -80,6 +81,33 @@ class DispatchPickOrderAttributionServiceTest {
|
||||
&& "PARTIALLY_SHIPPED".equals(value.getToStatus()) && value.getRemark().contains("6/10")));
|
||||
}
|
||||
|
||||
@Test void pickingLocksDistinctSourcesInOrderBeforeRecheckingReceipts() {
|
||||
when(attributions.availableReceipts("t1",91L)).thenReturn(List.of(receipt(2L,102L,1L,1L),
|
||||
receipt(1L,101L,1L,1L),receipt(3L,102L,1L,1L)));
|
||||
when(orders.lockById(eq("t1"),anyLong())).thenAnswer(call -> new LastlegOrder());
|
||||
service().preparePicking("t1",91L,null);
|
||||
var sequence=inOrder(attributions,orders);
|
||||
sequence.verify(attributions).availableReceipts("t1",91L);
|
||||
sequence.verify(orders).lockById("t1",101L);sequence.verify(orders).lockById("t1",102L);
|
||||
sequence.verify(attributions).availableReceipts("t1",91L);
|
||||
verify(orders,times(2)).lockById(eq("t1"),anyLong());
|
||||
}
|
||||
@Test void heldOrMissingSourcePreventsPickingPreparation() {
|
||||
when(attributions.availableReceipts("t1",91L)).thenReturn(List.of(receipt(1L,101L,1L,1L)));
|
||||
var held=new LastlegOrder();held.setDetained(true);
|
||||
when(orders.lockById("t1",101L)).thenReturn(held).thenReturn(null);
|
||||
assertThrows(org.springframework.web.server.ResponseStatusException.class,()->service().preparePicking("t1",91L,null));
|
||||
assertThrows(org.springframework.web.server.ResponseStatusException.class,()->service().preparePicking("t1",91L,null));
|
||||
verifyNoInteractions(history);
|
||||
}
|
||||
@Test void changedSourceSetAfterLockingFailsClosed() {
|
||||
when(attributions.availableReceipts("t1",91L)).thenReturn(List.of(receipt(1L,101L,1L,1L)))
|
||||
.thenReturn(List.of(receipt(2L,102L,1L,1L)));
|
||||
when(orders.lockById("t1",101L)).thenReturn(new LastlegOrder());
|
||||
assertThrows(org.springframework.web.server.ResponseStatusException.class,()->service().preparePicking("t1",91L,null));
|
||||
verify(orders,never()).increasePicked(anyString(),anyLong(),anyLong());
|
||||
}
|
||||
|
||||
private DispatchPickOrderAttributionService service() {
|
||||
DispatchPickOrderAttributionService service = new DispatchPickOrderAttributionService();
|
||||
ReflectionTestUtils.setField(service, "attributions", attributions);
|
||||
|
||||
+1
-1
@@ -292,7 +292,7 @@ class InventoryCommandServiceTest {
|
||||
when(commands.lockPosition("100", 201L)).thenReturn(position);
|
||||
when(commands.lockBalance("100", 200L)).thenReturn(balance);
|
||||
|
||||
service.holdInboundOrderInventory("100", "7", 101L, "ORD-1", "质检暂扣");
|
||||
service.holdInboundOrderInventory("100", "7", 101L, "ORD-1", "质检暂扣", "ORDER_HOLD:101:1");
|
||||
|
||||
assertEquals(0L, position.getAvailablePackages());
|
||||
assertEquals(5L, position.getHoldPackages());
|
||||
|
||||
+224
@@ -0,0 +1,224 @@
|
||||
package com.baiyue.lastleg.inventory.command;
|
||||
|
||||
import com.baomidou.mybatisplus.annotation.TableField;
|
||||
import com.baomidou.mybatisplus.annotation.TableName;
|
||||
import com.baomidou.mybatisplus.core.MybatisConfiguration;
|
||||
import com.baomidou.mybatisplus.extension.spring.MybatisSqlSessionFactoryBean;
|
||||
import com.baiyue.lastleg.dto.LastlegOrderCommandDtos;
|
||||
import com.baiyue.lastleg.dispatch.pda.service.DispatchPickOrderAttributionService;
|
||||
import com.baiyue.lastleg.dispatch.pda.mapper.DispatchPdaPickOrderMapper;
|
||||
import com.baiyue.lastleg.dispatch.pda.entity.DispatchPdaPickOrder;
|
||||
import com.baiyue.lastleg.inboundplan.pda.entity.InboundStockReceipt;
|
||||
import com.baiyue.lastleg.entity.*;
|
||||
import com.baiyue.lastleg.mapper.*;
|
||||
import com.baiyue.lastleg.service.LastlegOrderCommandService;
|
||||
import com.baiyue.lastleg.inventory.command.dto.InventoryCommandRequests;
|
||||
import com.baiyue.lastleg.inventory.command.mapper.InventoryCommandMapper;
|
||||
import com.baiyue.lastleg.inventory.command.service.InventoryCommandService;
|
||||
import com.baiyue.lastleg.inventory.entity.*;
|
||||
import com.baiyue.lastleg.inventory.mapper.*;
|
||||
import com.baiyue.lastleg.inboundplan.pda.gateway.WarehouseResourceGateway;
|
||||
import com.fasterxml.jackson.databind.ObjectMapper;
|
||||
import org.junit.jupiter.api.BeforeEach;
|
||||
import org.junit.jupiter.api.Test;
|
||||
import org.mybatis.spring.SqlSessionTemplate;
|
||||
import org.springframework.aop.framework.ProxyFactory;
|
||||
import org.springframework.jdbc.core.JdbcTemplate;
|
||||
import org.springframework.jdbc.datasource.DataSourceTransactionManager;
|
||||
import org.springframework.jdbc.datasource.DriverManagerDataSource;
|
||||
import org.springframework.test.util.ReflectionTestUtils;
|
||||
import org.springframework.transaction.annotation.AnnotationTransactionAttributeSource;
|
||||
import org.springframework.transaction.interceptor.TransactionInterceptor;
|
||||
import org.springframework.transaction.support.TransactionTemplate;
|
||||
import org.springframework.web.server.ResponseStatusException;
|
||||
|
||||
import java.lang.reflect.Modifier;
|
||||
import java.time.LocalDateTime;
|
||||
import java.util.Arrays;
|
||||
import java.util.List;
|
||||
import java.util.UUID;
|
||||
import java.util.concurrent.*;
|
||||
import java.util.stream.Collectors;
|
||||
import static org.junit.jupiter.api.Assertions.*;
|
||||
import static org.mockito.Mockito.mock;
|
||||
|
||||
/** Real local SQL/transactions. Exercises the inventory + source-order boundary used by PDA picking,
|
||||
* not the fleet/dispatch controller, warehouse remote transactions or TEST MySQL. */
|
||||
class InventoryHoldConcurrencyPersistenceTest {
|
||||
private JdbcTemplate jdbc;
|
||||
private TransactionTemplate tx;
|
||||
private InventoryCommandService inventory;
|
||||
private LastlegOrderCommandService orders;
|
||||
private LastlegOrderMapper orderMapper;
|
||||
private DispatchPickOrderAttributionService attribution;
|
||||
|
||||
@BeforeEach void setup() throws Exception {
|
||||
var ds = new DriverManagerDataSource("jdbc:h2:mem:hold-" + UUID.randomUUID()
|
||||
+ ";MODE=MySQL;DB_CLOSE_DELAY=-1;LOCK_TIMEOUT=2000", "sa", "");
|
||||
jdbc = new JdbcTemplate(ds);
|
||||
for (Class<?> entity : List.of(InventoryBalance.class, InventoryPosition.class, InventoryLedger.class,
|
||||
InventoryOperation.class, LastlegOrder.class, LastlegOrderCommandRecord.class,
|
||||
LastlegOrderHoldHistory.class, LastlegOrderOperationLog.class, InboundStockReceipt.class, DispatchPdaPickOrder.class)) {
|
||||
String columns = Arrays.stream(entity.getDeclaredFields())
|
||||
.filter(f -> !Modifier.isStatic(f.getModifiers()))
|
||||
.filter(f -> f.getAnnotation(TableField.class) == null || f.getAnnotation(TableField.class).exist())
|
||||
.map(f -> f.getName().replaceAll("([A-Z])", "_$1").toLowerCase() + " "
|
||||
+ (f.getType() == String.class ? "VARCHAR(16000)" : f.getType() == LocalDateTime.class ? "TIMESTAMP"
|
||||
: f.getType() == java.math.BigDecimal.class ? "DECIMAL(24,6)" : "BIGINT")
|
||||
+ (f.getName().equals("id") ? " PRIMARY KEY" : ""))
|
||||
.collect(Collectors.joining(","));
|
||||
jdbc.execute("CREATE TABLE " + entity.getAnnotation(TableName.class).value() + " (" + columns + ")");
|
||||
}
|
||||
jdbc.execute("CREATE UNIQUE INDEX operation_key ON ll_inventory_operation(customer_id,idempotency_key)");
|
||||
var config = new MybatisConfiguration(); config.setMapUnderscoreToCamelCase(true);
|
||||
for (Class<?> mapper : List.of(InventoryBalanceMapper.class, InventoryPositionMapper.class,
|
||||
InventoryLedgerMapper.class, InventoryOperationMapper.class, InventoryCommandMapper.class,
|
||||
LastlegOrderMapper.class, LastlegOrderCommandRecordMapper.class, LastlegOrderHoldHistoryMapper.class,
|
||||
LastlegOrderOperationLogMapper.class, DispatchPdaPickOrderMapper.class)) config.addMapper(mapper);
|
||||
var factory = new MybatisSqlSessionFactoryBean(); factory.setDataSource(ds); factory.setConfiguration(config);
|
||||
var session = new SqlSessionTemplate(factory.getObject());
|
||||
var manager = new DataSourceTransactionManager(ds); tx = new TransactionTemplate(manager);
|
||||
var stock = new InventoryCommandService();
|
||||
ReflectionTestUtils.setField(stock,"balances",session.getMapper(InventoryBalanceMapper.class));
|
||||
ReflectionTestUtils.setField(stock,"positions",session.getMapper(InventoryPositionMapper.class));
|
||||
ReflectionTestUtils.setField(stock,"ledgers",session.getMapper(InventoryLedgerMapper.class));
|
||||
ReflectionTestUtils.setField(stock,"operations",session.getMapper(InventoryOperationMapper.class));
|
||||
ReflectionTestUtils.setField(stock,"commands",session.getMapper(InventoryCommandMapper.class));
|
||||
ReflectionTestUtils.setField(stock,"warehouseGateway",mock(WarehouseResourceGateway.class));
|
||||
ReflectionTestUtils.setField(stock,"objectMapper",new ObjectMapper().findAndRegisterModules());
|
||||
inventory = proxy(stock,manager);
|
||||
orderMapper = session.getMapper(LastlegOrderMapper.class);
|
||||
var sourceLocks = new DispatchPickOrderAttributionService();
|
||||
ReflectionTestUtils.setField(sourceLocks,"orders",orderMapper);
|
||||
ReflectionTestUtils.setField(sourceLocks,"attributions",session.getMapper(DispatchPdaPickOrderMapper.class));
|
||||
attribution = proxy(sourceLocks,manager);
|
||||
var target = new LastlegOrderCommandService();
|
||||
ReflectionTestUtils.setField(target,"orders",orderMapper);
|
||||
ReflectionTestUtils.setField(target,"commands",session.getMapper(LastlegOrderCommandRecordMapper.class));
|
||||
ReflectionTestUtils.setField(target,"holdHistory",session.getMapper(LastlegOrderHoldHistoryMapper.class));
|
||||
ReflectionTestUtils.setField(target,"operationLogs",session.getMapper(LastlegOrderOperationLogMapper.class));
|
||||
ReflectionTestUtils.setField(target,"inventory",inventory);
|
||||
ReflectionTestUtils.setField(target,"objectMapper",new ObjectMapper().findAndRegisterModules());
|
||||
orders = proxy(target,manager);
|
||||
jdbc.update("INSERT INTO ll_order(id,customer_id,order_no,current_status,package_count,picked_package_count,shipped_package_count,detained,version) VALUES(11,'C1','ORDER-1','INBOUND_COMPLETED',12,0,0,0,0)");
|
||||
jdbc.update("INSERT INTO ll_inventory_balance(id,customer_id,overseas_warehouse_id,inbound_order_id,inbound_code,sku_code,available_packages,hold_packages,allocated_packages,picked_packages,in_transit_packages,receiving_packages,procurement_packages,version) VALUES(20,'C1','WH1',11,'IN-1','SKU',12,0,0,0,0,0,0,0)");
|
||||
for (long id : new long[]{21,22}) jdbc.update("INSERT INTO ll_inventory_position(id,customer_id,balance_id,pallet_id,pallet_no,batch_no,status,available_packages,hold_packages,allocated_packages,picked_packages,receiving_packages,version) VALUES(?,'C1',20,?,'PALLET','BATCH','OCCUPIED',?,0,0,0,0,0)",id,id,id==21?5:7);
|
||||
jdbc.update("INSERT INTO ll_inbound_stock_receipt(id,customer_id,order_id,inventory_position_id,received_packages,received_at) VALUES(31,'C1',11,21,5,CURRENT_TIMESTAMP),(32,'C1',11,22,7,CURRENT_TIMESTAMP)");
|
||||
}
|
||||
@SuppressWarnings("unchecked") private <T> T proxy(T target, DataSourceTransactionManager manager) {
|
||||
var factory = new ProxyFactory(target); factory.setProxyTargetClass(true);
|
||||
factory.addAdvice(new TransactionInterceptor(manager,new AnnotationTransactionAttributeSource()));
|
||||
return (T) factory.getProxy();
|
||||
}
|
||||
private void hold(boolean held,int version,String key) {
|
||||
orders.hold("C1",11,key,new LastlegOrderCommandDtos.HoldRequest(held,false,"quality",version),"operator","Operator");
|
||||
}
|
||||
private InventoryCommandRequests.Move move(long position,long packages,String key) {
|
||||
return new InventoryCommandRequests.Move(position,packages,"DISPATCH-1","DISPATCH-1",null,key,"fixture");
|
||||
}
|
||||
private void scan(Runnable afterStock) {
|
||||
tx.executeWithoutResult(ignored -> {
|
||||
attribution.preparePicking("C1",21,null);
|
||||
inventory.allocate("C1","picker",move(21,5,"scan-allocate"));
|
||||
inventory.pick("C1","picker",move(21,5,"scan-pick"));
|
||||
inventory.ship("C1","picker",move(21,5,"scan-ship"));
|
||||
afterStock.run();
|
||||
// Same production source-order update used by DispatchPickOrderAttributionService.
|
||||
if (orderMapper.increasePicked("C1",11L,5) != 1) throw new ResponseStatusException(org.springframework.http.HttpStatus.CONFLICT,"source held");
|
||||
});
|
||||
}
|
||||
private long amount(String column) { return jdbc.queryForObject("SELECT " + column + " FROM ll_inventory_balance WHERE id=20",Long.class); }
|
||||
private long count(String table) { return jdbc.queryForObject("SELECT COUNT(*) FROM " + table,Long.class); }
|
||||
private void conserved(long shipped) {
|
||||
for (String column : List.of("available_packages","hold_packages","allocated_packages","picked_packages")) {
|
||||
assertEquals(amount(column),jdbc.queryForObject("SELECT SUM("+column+") FROM ll_inventory_position WHERE balance_id=20",Long.class));
|
||||
assertTrue(amount(column)>=0);
|
||||
}
|
||||
assertEquals(12,amount("available_packages")+amount("hold_packages")+amount("allocated_packages")+amount("picked_packages")+shipped);
|
||||
assertEquals(shipped,jdbc.queryForObject("SELECT COALESCE(-SUM(delta_packages),0) FROM ll_inventory_ledger WHERE business_type='SHIP'",Long.class));
|
||||
}
|
||||
@Test void secondHoldCycleMustMoveInventoryAgainAndNotReuseTheFirstCycle() {
|
||||
hold(true,0,"hold-one"); hold(false,1,"release-one"); hold(true,2,"hold-two");
|
||||
assertEquals(0,amount("available_packages")); assertEquals(12,amount("hold_packages"));
|
||||
hold(true,2,"hold-two"); // Original command replay remains harmless.
|
||||
assertEquals(6,count("ll_inventory_operation")); conserved(0);
|
||||
hold(false,3,"release-two");
|
||||
assertEquals(12,amount("available_packages"));assertEquals(0,amount("hold_packages"));
|
||||
assertEquals(8,count("ll_inventory_operation"));conserved(0);
|
||||
}
|
||||
@Test void releasingOrderHoldMustPreserveAnotherBusinessHold() {
|
||||
inventory.hold("C1","operator",move(21,2,"RELABEL-HOLD"));
|
||||
hold(true,0,"order-hold"); hold(false,1,"order-release");
|
||||
assertEquals(2,amount("hold_packages")); assertEquals(10,amount("available_packages")); conserved(0);
|
||||
}
|
||||
@Test void secondPositionLedgerFailureRollsBackOrderStateAndFirstPosition() {
|
||||
jdbc.execute("ALTER TABLE ll_inventory_ledger ADD CONSTRAINT reject_second CHECK(position_id<>22)");
|
||||
assertThrows(RuntimeException.class,()->hold(true,0,"failed-hold"));
|
||||
assertEquals(12,amount("available_packages")); assertEquals(0,amount("hold_packages"));
|
||||
assertEquals(0,count("ll_inventory_operation")); assertEquals(0,count("ll_inventory_ledger"));
|
||||
assertEquals(0,count("ll_order_hold_history")); assertEquals(0,count("ll_order_command_record"));
|
||||
assertFalse(Boolean.TRUE.equals(orderMapper.selectById(11L).getDetained())); conserved(0);
|
||||
}
|
||||
@Test void failedReleaseRollsBackEarlierReleaseAndKeepsTheOrderHeld() {
|
||||
hold(true,0,"hold-before-failed-release");
|
||||
jdbc.execute("ALTER TABLE ll_inventory_ledger ADD CONSTRAINT reject_release CHECK(business_type<>'RELEASE' OR position_id<>22)");
|
||||
assertThrows(RuntimeException.class,()->hold(false,1,"failed-release"));
|
||||
assertEquals(12,amount("hold_packages"));assertEquals(0,amount("available_packages"));
|
||||
assertEquals(2,count("ll_inventory_operation"));assertEquals(2,count("ll_inventory_ledger"));
|
||||
assertEquals(1,count("ll_order_hold_history"));assertTrue(orderMapper.selectById(11L).getDetained());conserved(0);
|
||||
}
|
||||
@Test void committedHoldMakesConcurrentScanFailWithoutPartialInventoryWrites() throws Exception {
|
||||
var held = new CountDownLatch(1); var finish = new CountDownLatch(1); var entered = new CountDownLatch(1);
|
||||
ExecutorService pool = Executors.newFixedThreadPool(2);
|
||||
try {
|
||||
Future<?> holding=pool.submit(()->tx.executeWithoutResult(s->{hold(true,0,"hold-first");held.countDown();await(finish);}));
|
||||
await(held);
|
||||
Future<Throwable> scanning=pool.submit(()->capture(()->{entered.countDown();scan(()->{});}));
|
||||
await(entered); finish.countDown(); holding.get(8,TimeUnit.SECONDS);
|
||||
assertInstanceOf(ResponseStatusException.class,scanning.get(8,TimeUnit.SECONDS));
|
||||
assertEquals(12,amount("hold_packages")); assertEquals(2,count("ll_inventory_operation")); conserved(0);
|
||||
} finally { finish.countDown();pool.shutdownNow();assertTrue(pool.awaitTermination(10,TimeUnit.SECONDS)); }
|
||||
}
|
||||
@Test void committedScanMakesConcurrentHoldFailAndOnlyShipsOnce() throws Exception {
|
||||
var picked=new CountDownLatch(1);var finish=new CountDownLatch(1);var entered=new CountDownLatch(1);
|
||||
ExecutorService pool=Executors.newFixedThreadPool(2);
|
||||
try {
|
||||
Future<?> scanning=pool.submit(()->tx.executeWithoutResult(s->{scan(()->{});picked.countDown();await(finish);}));
|
||||
await(picked);
|
||||
Future<Throwable> holding=pool.submit(()->capture(()->{entered.countDown();hold(true,0,"hold-after-scan");}));
|
||||
await(entered);finish.countDown();scanning.get(8,TimeUnit.SECONDS);
|
||||
assertInstanceOf(ResponseStatusException.class,holding.get(8,TimeUnit.SECONDS));
|
||||
assertEquals(7,amount("available_packages"));assertEquals(0,count("ll_order_hold_history"));conserved(5);
|
||||
} finally {finish.countDown();pool.shutdownNow();assertTrue(pool.awaitTermination(10,TimeUnit.SECONDS));}
|
||||
}
|
||||
@Test void concurrentReleaseAndScanPreserveOtherBusinessHold() throws Exception {
|
||||
inventory.hold("C1","operator",move(22,2,"OTHER-HOLD"));hold(true,0,"order-held");
|
||||
var released=new CountDownLatch(1);var finish=new CountDownLatch(1);var entered=new CountDownLatch(1);
|
||||
ExecutorService pool=Executors.newFixedThreadPool(2);
|
||||
try {
|
||||
Future<?> releasing=pool.submit(()->tx.executeWithoutResult(s->{hold(false,1,"release-before-scan");released.countDown();await(finish);}));
|
||||
await(released);
|
||||
Future<Throwable> scanning=pool.submit(()->capture(()->{entered.countDown();scan(()->{});}));
|
||||
await(entered);finish.countDown();releasing.get(8,TimeUnit.SECONDS);
|
||||
assertNull(scanning.get(8,TimeUnit.SECONDS));
|
||||
assertEquals(5,amount("available_packages"));assertEquals(2,amount("hold_packages"));conserved(5);
|
||||
} finally {finish.countDown();pool.shutdownNow();assertTrue(pool.awaitTermination(10,TimeUnit.SECONDS));}
|
||||
}
|
||||
@Test void overlappingHoldAndScanSerializeOnSourceOrderBeforeTouchingStock() throws Exception {
|
||||
var orderLocked=new CountDownLatch(1);var scanEntering=new CountDownLatch(1);
|
||||
ExecutorService pool=Executors.newFixedThreadPool(2);
|
||||
try {
|
||||
Future<Throwable> holding=pool.submit(()->capture(()->tx.executeWithoutResult(s->{
|
||||
orderMapper.lockById("C1",11L);orderLocked.countDown();await(scanEntering);
|
||||
hold(true,0,"crossing-hold");})));
|
||||
await(orderLocked);
|
||||
Future<Throwable> scanning=pool.submit(()->capture(()->{scanEntering.countDown();scan(()->{});}));
|
||||
Throwable scanError=scanning.get(10,TimeUnit.SECONDS),holdError=holding.get(10,TimeUnit.SECONDS);
|
||||
assertNull(holdError,"hold technical failure: "+holdError);
|
||||
assertInstanceOf(ResponseStatusException.class,scanError);
|
||||
assertEquals(12,amount("hold_packages"));conserved(0);
|
||||
} finally {scanEntering.countDown();pool.shutdownNow();assertTrue(pool.awaitTermination(10,TimeUnit.SECONDS));}
|
||||
}
|
||||
private static Throwable capture(Runnable action) {try{action.run();return null;}catch(RuntimeException failure){return failure;}}
|
||||
private static void await(CountDownLatch gate) {try{assertTrue(gate.await(6,TimeUnit.SECONDS),"transaction gate timed out");}catch(InterruptedException e){Thread.currentThread().interrupt();throw new IllegalStateException(e);}}
|
||||
}
|
||||
+4
-2
@@ -84,7 +84,7 @@ class LastlegOrderCommandServiceTest {
|
||||
verify(orders).updateHold(eq("100"), eq(11L), eq(true), eq(true), eq("海关查验"), eq(3), any());
|
||||
verify(holdHistory).insert(any(LastlegOrderHoldHistory.class));
|
||||
verify(commands).insert(any(LastlegOrderCommandRecord.class));
|
||||
verify(inventory).holdInboundOrderInventory("100", "900", 11L, "LL-11", "海关查验");
|
||||
verify(inventory).holdInboundOrderInventory("100", "900", 11L, "LL-11", "海关查验", "ORDER_HOLD:11:4");
|
||||
verify(orders, never()).lockContainerOrders(any());
|
||||
}
|
||||
|
||||
@@ -93,6 +93,8 @@ class LastlegOrderCommandServiceTest {
|
||||
LastlegOrder order = order(11L, 3);
|
||||
order.setCurrentStatus("INBOUND_COMPLETED");
|
||||
order.setDetained(true);
|
||||
var originalHold = new LastlegOrderHoldHistory(); originalHold.setOrderVersion(2);
|
||||
when(holdHistory.selectOne(any())).thenReturn(originalHold);
|
||||
when(orders.selectById(11L)).thenReturn(order);
|
||||
when(orders.lockById("100", 11L)).thenReturn(order);
|
||||
when(orders.updateHold(eq("100"), eq(11L), eq(false), eq(false), eq("质检通过"), eq(3),
|
||||
@@ -101,7 +103,7 @@ class LastlegOrderCommandServiceTest {
|
||||
service.hold("100", 11L, "release-11",
|
||||
new LastlegOrderCommandDtos.HoldRequest(false, false, "质检通过", 3), "900", "管理员");
|
||||
|
||||
verify(inventory).releaseInboundOrderHold("100", "900", 11L, "LL-11", "质检通过");
|
||||
verify(inventory).releaseInboundOrderHold("100", "900", 11L, "LL-11", "质检通过", "ORDER_HOLD:11:2");
|
||||
}
|
||||
|
||||
@Test
|
||||
|
||||
Reference in New Issue
Block a user