fix(return): replay concurrent merges and extend integrity acceptance
This commit is contained in:
+3
@@ -32,6 +32,9 @@ public interface ReturnInventoryCartonMapper extends BaseMapper<ReturnInventoryC
|
||||
@Select("SELECT * FROM ll_return_inventory_carton WHERE carton_mark=#{cartonMark} LIMIT 1 FOR UPDATE")
|
||||
ReturnInventoryCarton lockMark(@Param("cartonMark") String cartonMark);
|
||||
|
||||
@Select("SELECT * FROM ll_return_inventory_carton WHERE id=#{id} FOR UPDATE")
|
||||
ReturnInventoryCarton lockId(@Param("id") long id);
|
||||
|
||||
@Select("SELECT * FROM ll_return_inventory_carton WHERE source_receipt_id=#{receiptId} LIMIT 1")
|
||||
ReturnInventoryCarton findReceipt(@Param("receiptId") long receiptId);
|
||||
|
||||
|
||||
+3
@@ -10,4 +10,7 @@ import org.apache.ibatis.annotations.Select;
|
||||
public interface ReturnInventoryOperationMapper extends BaseMapper<ReturnInventoryOperation> {
|
||||
@Select("SELECT * FROM ll_return_inventory_operation WHERE operator_id=#{operatorId} AND idempotency_key=#{key} LIMIT 1")
|
||||
ReturnInventoryOperation findKey(@Param("operatorId") String operatorId, @Param("key") String key);
|
||||
|
||||
@Select("SELECT * FROM ll_return_inventory_operation WHERE operator_id=#{operatorId} AND idempotency_key=#{key} LIMIT 1 FOR UPDATE")
|
||||
ReturnInventoryOperation lockKey(@Param("operatorId") String operatorId, @Param("key") String key);
|
||||
}
|
||||
|
||||
+11
-5
@@ -224,7 +224,8 @@ public class ReturnInventoryService {
|
||||
operation.setRequestHash(requestHash); operation.setOperationType("MERGE"); operation.setCreatedAt(now); operation.setUpdatedAt(now);
|
||||
try { operations.insert(operation); }
|
||||
catch (DuplicateKeyException concurrent) {
|
||||
ReturnInventoryOperation existing = operations.findKey(identity.operatorKey(), key);
|
||||
// The duplicate insert waited for the winner; the earlier snapshot cannot see its commit.
|
||||
ReturnInventoryOperation existing = operations.lockKey(identity.operatorKey(), key);
|
||||
if (existing != null) return replayMerge(existing, requestHash);
|
||||
throw concurrent;
|
||||
}
|
||||
@@ -342,14 +343,19 @@ public class ReturnInventoryService {
|
||||
private ReturnInventoryResponses.MergeResult replayMerge(ReturnInventoryOperation operation, String requestHash) {
|
||||
if (!requestHash.equals(operation.getRequestHash())) throw conflict("幂等键已用于不同的合箱请求");
|
||||
if (operation.getResultCartonId() == null) throw conflict("合箱请求正在处理");
|
||||
ReturnInventoryCarton carton = cartons.selectById(operation.getResultCartonId());
|
||||
// The result and its label must also bypass the snapshot established before the insert race.
|
||||
ReturnInventoryCarton carton = cartons.lockId(operation.getResultCartonId());
|
||||
if (carton == null) throw new IllegalStateException("合箱结果箱不存在");
|
||||
List<ReturnInventoryCartonItem> items = cartonItems.listCarton(carton.getId());
|
||||
return new ReturnInventoryResponses.MergeResult(carton.getCartonMark(), items.get(0).getSkuCode(), carton.getTotalQuantity(), label(carton));
|
||||
List<ReturnInventoryCartonItem> items = cartonItems.lockCarton(carton.getId());
|
||||
return new ReturnInventoryResponses.MergeResult(carton.getCartonMark(), items.get(0).getSkuCode(), carton.getTotalQuantity(), label(carton, items));
|
||||
}
|
||||
|
||||
private ReturnInventoryResponses.Label label(ReturnInventoryCarton carton) {
|
||||
List<ReturnInventoryResponses.CartonItem> items = cartonItems.listCarton(carton.getId()).stream()
|
||||
return label(carton, cartonItems.listCarton(carton.getId()));
|
||||
}
|
||||
|
||||
private ReturnInventoryResponses.Label label(ReturnInventoryCarton carton, List<ReturnInventoryCartonItem> cartonLines) {
|
||||
List<ReturnInventoryResponses.CartonItem> items = cartonLines.stream()
|
||||
.map(item -> new ReturnInventoryResponses.CartonItem(item.getSkuCode(), item.getProductNameZh(), item.getQuantity())).toList();
|
||||
return new ReturnInventoryResponses.Label(carton.getCartonMark(), carton.getOverseasWarehouseId(), carton.getCustomerId(),
|
||||
carton.getSourceBatchNo(), carton.getSourceReturnNo(), value(carton.getTotalQuantity()), carton.getLength(), carton.getWidth(), carton.getHeight(), carton.getWeight(), items, LocalDateTime.now(), printableHtml(carton, items));
|
||||
|
||||
+8
@@ -2,6 +2,8 @@ package com.baiyue.lastleg.returnreceipt.service;
|
||||
|
||||
import com.baiyue.lastleg.returnreceipt.mapper.ReturnReceiptMapper;
|
||||
import com.baiyue.lastleg.returnreceipt.mapper.ReturnInventoryCartonItemMapper;
|
||||
import com.baiyue.lastleg.returnreceipt.mapper.ReturnInventoryCartonMapper;
|
||||
import com.baiyue.lastleg.returnreceipt.mapper.ReturnInventoryOperationMapper;
|
||||
import org.apache.ibatis.annotations.Select;
|
||||
import org.junit.jupiter.api.Test;
|
||||
import static org.junit.jupiter.api.Assertions.assertTrue;
|
||||
@@ -14,5 +16,11 @@ class ReturnCurrentReadContractTest {
|
||||
assertTrue(receiptSql.endsWith("FOR UPDATE"));
|
||||
assertTrue(itemSql.endsWith("FOR UPDATE"));
|
||||
assertTrue(itemSql.contains("ORDER BY sku_code"));
|
||||
String operationSql = String.join(" ", ReturnInventoryOperationMapper.class.getMethod("lockKey", String.class, String.class).getAnnotation(Select.class).value());
|
||||
String cartonSql = String.join(" ", ReturnInventoryCartonMapper.class.getMethod("lockId", long.class).getAnnotation(Select.class).value());
|
||||
assertTrue(operationSql.contains("operator_id=#{operatorId} AND idempotency_key=#{key}"));
|
||||
assertTrue(operationSql.endsWith("FOR UPDATE"));
|
||||
assertTrue(cartonSql.contains("WHERE id=#{id}"));
|
||||
assertTrue(cartonSql.endsWith("FOR UPDATE"));
|
||||
}
|
||||
}
|
||||
|
||||
+58
@@ -4,6 +4,7 @@ import com.baiyue.lastleg.pda.security.PdaWarehouseIdentity;
|
||||
import com.baiyue.lastleg.returnreceipt.dto.ReturnInventoryRequests;
|
||||
import com.baiyue.lastleg.returnreceipt.entity.ReturnInventoryCarton;
|
||||
import com.baiyue.lastleg.returnreceipt.entity.ReturnInventoryCartonItem;
|
||||
import com.baiyue.lastleg.returnreceipt.entity.ReturnInventoryOperation;
|
||||
import com.baiyue.lastleg.returnreceipt.entity.ReturnOrder;
|
||||
import com.baiyue.lastleg.returnreceipt.entity.ReturnReceipt;
|
||||
import com.baiyue.lastleg.returnreceipt.mapper.ReturnInventoryBalanceMapper;
|
||||
@@ -22,6 +23,7 @@ import org.mockito.ArgumentCaptor;
|
||||
import org.mockito.Mock;
|
||||
import org.mockito.junit.jupiter.MockitoExtension;
|
||||
import org.springframework.test.util.ReflectionTestUtils;
|
||||
import org.springframework.dao.DuplicateKeyException;
|
||||
|
||||
import java.math.BigDecimal;
|
||||
import java.util.List;
|
||||
@@ -32,6 +34,7 @@ import static org.mockito.ArgumentMatchers.anyString;
|
||||
import static org.mockito.Mockito.verify;
|
||||
import static org.mockito.Mockito.when;
|
||||
import static org.mockito.Mockito.never;
|
||||
import static org.mockito.Mockito.verifyNoInteractions;
|
||||
import static org.mockito.ArgumentMatchers.any;
|
||||
import static org.junit.jupiter.api.Assertions.assertThrows;
|
||||
import org.springframework.web.server.ResponseStatusException;
|
||||
@@ -195,6 +198,61 @@ class ReturnInventoryServiceTest {
|
||||
verify(balances).adjustCartonQuantity("10001", "42", "SKU-A", 0L);
|
||||
}
|
||||
|
||||
@Test
|
||||
void sameKeyInsertRaceReplaysCommittedResultAndLabelUsingOnlyCurrentReads() {
|
||||
var request = mergeRequest(1);
|
||||
var winner = winningOperation(request);
|
||||
when(operations.insert(any(ReturnInventoryOperation.class))).thenThrow(new DuplicateKeyException("same key"));
|
||||
when(operations.lockKey("operator-2", "concurrent-key")).thenReturn(winner);
|
||||
ReturnInventoryCarton target = new ReturnInventoryCarton();
|
||||
target.setId(31L); target.setCartonMark("RC-WINNER"); target.setTotalQuantity(1L); target.setSkuTypes(1);
|
||||
when(cartons.lockId(31L)).thenReturn(target);
|
||||
when(cartonItems.lockCarton(31L)).thenReturn(List.of(item(41L, "SKU-A", 1L)));
|
||||
|
||||
var result = service.merge(new PdaWarehouseIdentity("operator-2", "操作员2", 42L), request);
|
||||
|
||||
assertThat(result.cartonMark()).isEqualTo("RC-WINNER");
|
||||
assertThat(result.totalQuantity()).isEqualTo(1L);
|
||||
assertThat(result.label().items()).hasSize(1);
|
||||
assertThat(result.label().items().get(0).quantity()).isEqualTo(1L);
|
||||
assertThat(result.label().printableHtml()).contains("RC-WINNER", "SKU-A");
|
||||
verify(operations).findKey("operator-2", "concurrent-key");
|
||||
verify(operations).lockKey("operator-2", "concurrent-key");
|
||||
verify(cartons, never()).selectById(anyLong());
|
||||
verify(cartonItems, never()).listCarton(anyLong());
|
||||
verify(cartons, never()).insert(any(ReturnInventoryCarton.class));
|
||||
verify(cartonItems, never()).insert(any(ReturnInventoryCartonItem.class));
|
||||
verify(operations, never()).updateById(any(ReturnInventoryOperation.class));
|
||||
verifyNoInteractions(orders, receipts, balances, numbers);
|
||||
}
|
||||
|
||||
@Test
|
||||
void sameKeyInsertRaceWithDifferentPayloadIsConflictWithoutInventoryWrites() {
|
||||
when(operations.insert(any(ReturnInventoryOperation.class))).thenThrow(new DuplicateKeyException("same key"));
|
||||
when(operations.lockKey("operator-2", "concurrent-key")).thenReturn(winningOperation(mergeRequest(1)));
|
||||
var error = assertThrows(ResponseStatusException.class, () -> service.merge(
|
||||
new PdaWarehouseIdentity("operator-2", "操作员2", 42L), mergeRequest(2)));
|
||||
assertThat(error.getStatusCode().value()).isEqualTo(409);
|
||||
verifyNoInteractions(cartons, cartonItems, orders, receipts, balances, numbers);
|
||||
}
|
||||
|
||||
@Test
|
||||
void unrelatedDuplicateInsertFailureIsNotMistakenForSuccessfulReplay() {
|
||||
var duplicate = new DuplicateKeyException("different unique constraint");
|
||||
when(operations.insert(any(ReturnInventoryOperation.class))).thenThrow(duplicate);
|
||||
assertThat(assertThrows(DuplicateKeyException.class, () -> service.merge(
|
||||
new PdaWarehouseIdentity("operator-2", "操作员2", 42L), mergeRequest(1)))).isSameAs(duplicate);
|
||||
verify(operations).lockKey("operator-2", "concurrent-key");
|
||||
verifyNoInteractions(cartons, cartonItems, orders, receipts, balances, numbers);
|
||||
}
|
||||
|
||||
private ReturnInventoryOperation winningOperation(ReturnInventoryRequests.Merge request) {
|
||||
ReturnInventoryOperation operation = new ReturnInventoryOperation();
|
||||
operation.setResultCartonId(31L);
|
||||
operation.setRequestHash(ReflectionTestUtils.invokeMethod(service, "hash", request));
|
||||
return operation;
|
||||
}
|
||||
|
||||
private ReturnInventoryCarton prepareConcurrentSource() {
|
||||
ReturnReceipt receipt = new ReturnReceipt(); receipt.setId(21L);
|
||||
ReturnInventoryCarton snapshot = new ReturnInventoryCarton();
|
||||
|
||||
Reference in New Issue
Block a user