java-project/baiyue-firstleg
4
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix: converge inbound pallets and retries

Browse Source
This commit is contained in:
ouyangxq
2026-09-11 00:06:57 +08:00
parent ed12769858
commit 2bd12f8a5f
11 changed files with 65 additions and 15 deletions
Download Patch File Download Diff File Expand all files Collapse all files
deploy/sql/海外区域统一增量-MySQL8.sql
+8 -2
View File
@@ -859,6 +859,8 @@ SET @ddl = (SELECT IF(EXISTS (SELECT 1 FROM information_schema.columns WHERE tab
PREPARE inbound_schema_stmt FROM @ddl; EXECUTE inbound_schema_stmt; DEALLOCATE PREPARE inbound_schema_stmt;
SET @ddl = (SELECT IF(EXISTS (SELECT 1 FROM information_schema.columns WHERE table_schema=DATABASE() AND table_name='ll_inbound_plan_pallet' AND column_name='actual_package_count'), 'SELECT 1', 'ALTER TABLE ll_inbound_plan_pallet ADD COLUMN actual_package_count INT NOT NULL DEFAULT 0 COMMENT ''实收箱数'''));
PREPARE inbound_schema_stmt FROM @ddl; EXECUTE inbound_schema_stmt; DEALLOCATE PREPARE inbound_schema_stmt;
SET @ddl = (SELECT IF(EXISTS (SELECT 1 FROM information_schema.columns WHERE table_schema=DATABASE() AND table_name='ll_inbound_plan_pallet' AND column_name='version'), 'SELECT 1', 'ALTER TABLE ll_inbound_plan_pallet ADD COLUMN version INT NOT NULL DEFAULT 0 COMMENT ''收货累计值乐观锁版本'''));
PREPARE inbound_schema_stmt FROM @ddl; EXECUTE inbound_schema_stmt; DEALLOCATE PREPARE inbound_schema_stmt;
SET @ddl = (SELECT IF(EXISTS (SELECT 1 FROM information_schema.columns WHERE table_schema=DATABASE() AND table_name='ll_inbound_plan_pallet_allocation' AND column_name='planned_package_count'), 'SELECT 1', 'ALTER TABLE ll_inbound_plan_pallet_allocation ADD COLUMN planned_package_count INT NOT NULL DEFAULT 0 COMMENT ''计划分配箱数'''));
PREPARE inbound_schema_stmt FROM @ddl; EXECUTE inbound_schema_stmt; DEALLOCATE PREPARE inbound_schema_stmt;
SET @ddl = (SELECT IF(EXISTS (SELECT 1 FROM information_schema.columns WHERE table_schema=DATABASE() AND table_name='ll_inbound_plan_pallet_allocation' AND column_name='counted_package_count'), 'SELECT 1', 'ALTER TABLE ll_inbound_plan_pallet_allocation ADD COLUMN counted_package_count INT NOT NULL DEFAULT 0 COMMENT ''清点箱数'''));
@@ -4638,8 +4640,12 @@ BEGIN
IF NOT EXISTS (SELECT 1 FROM information_schema.columns WHERE table_schema='baiyue_lastleg' AND table_name='ll_relabel_application' AND column_name='total_file_url' AND column_type='varchar(1000)') THEN
SIGNAL SQLSTATE '45000' SET MESSAGE_TEXT='缺少字段 ll_relabel_application.total_file_url';
END IF;
IF NOT EXISTS (SELECT 1 FROM information_schema.columns WHERE table_schema='baiyue_lastleg' AND table_name='ll_inbound_plan_pallet' AND column_name='pallet_no' AND column_type='varchar(128)' AND is_nullable='NO') THEN
SIGNAL SQLSTATE '45000' SET MESSAGE_TEXT='入仓计划托盘码长度未就绪';
IF (SELECT COUNT(*) FROM information_schema.columns WHERE table_schema='baiyue_lastleg'
AND table_name='ll_inbound_plan_pallet' AND column_name IN ('pallet_no','version')) <> 2
OR NOT EXISTS (SELECT 1 FROM information_schema.columns WHERE table_schema='baiyue_lastleg'
AND table_name='ll_inbound_plan_pallet' AND column_name='pallet_no'
AND column_type='varchar(128)' AND is_nullable='NO') THEN
SIGNAL SQLSTATE '45000' SET MESSAGE_TEXT='入仓计划托盘码或乐观锁版本字段未就绪';
END IF;
IF NOT EXISTS (SELECT 1 FROM information_schema.statistics WHERE table_schema='baiyue_lastleg' AND table_name='ll_delivery_document_revision' AND index_name='uk_delivery_document_command') THEN
SIGNAL SQLSTATE '45000' SET MESSAGE_TEXT='缺少索引 uk_delivery_document_command';
file/src/main/java/com/baiyue/file/mapper/FileExportOutboxMapper.java
+7 -6
View File
@@ -44,12 +44,13 @@ public interface FileExportOutboxMapper extends BaseMapper<FileExportOutbox> {
int releaseExpiredClaims();
@Update("""
UPDATE file_export_outbox o JOIN file_export_task t
ON t.id=o.task_id AND t.dispatch_version=o.dispatch_version
SET o.status='PENDING',o.next_retry_at=CURRENT_TIMESTAMP(3),o.lease_owner=NULL,o.lease_until=NULL
WHERE o.status='SENT' AND o.sent_at<=DATE_SUB(CURRENT_TIMESTAMP(3), INTERVAL #{staleSeconds} SECOND)
AND t.status='PENDING' AND t.next_retry_at<=DATE_SUB(CURRENT_TIMESTAMP(3), INTERVAL #{graceSeconds} SECOND)
LIMIT 20
UPDATE file_export_outbox
SET status='PENDING',next_retry_at=CURRENT_TIMESTAMP(3),lease_owner=NULL,lease_until=NULL
WHERE id IN (SELECT id FROM (SELECT o.id FROM file_export_outbox o JOIN file_export_task t
ON t.id=o.task_id AND t.dispatch_version=o.dispatch_version
WHERE o.status='SENT' AND o.sent_at<=DATE_SUB(CURRENT_TIMESTAMP(3), INTERVAL #{staleSeconds} SECOND)
AND t.status='PENDING' AND t.next_retry_at<=DATE_SUB(CURRENT_TIMESTAMP(3), INTERVAL #{graceSeconds} SECOND)
ORDER BY o.sent_at,o.id LIMIT 20) stale)
""")
int redispatchSent(@Param("staleSeconds") int staleSeconds, @Param("graceSeconds") int graceSeconds);
file/src/test/java/com/baiyue/file/mapper/ExportMapperContractTest.java
+3
View File
@@ -22,6 +22,9 @@ class ExportMapperContractTest {
String sql=update(FileExportOutboxMapper.class,"redispatchSent");
assertTrue(sql.contains("o.status='SENT'"));
assertTrue(sql.contains("t.dispatch_version=o.dispatch_version"));
assertTrue(sql.contains("WHERE id IN (SELECT id FROM (SELECT o.id"));
assertTrue(sql.contains("ORDER BY o.sent_at,o.id LIMIT 20"));
assertFalse(sql.contains("SET o.status"));
assertFalse(sql.contains("dispatch_version=dispatch_version+1"));
assertFalse(sql.contains("QUARANTINED"));
}
lastleg-service/src/main/resources/db/lastleg.sql
+2
View File
@@ -766,6 +766,8 @@ SET @ddl = (SELECT IF(EXISTS (SELECT 1 FROM information_schema.columns WHERE tab
PREPARE inbound_schema_stmt FROM @ddl; EXECUTE inbound_schema_stmt; DEALLOCATE PREPARE inbound_schema_stmt;
SET @ddl = (SELECT IF(EXISTS (SELECT 1 FROM information_schema.columns WHERE table_schema=DATABASE() AND table_name='ll_inbound_plan_pallet' AND column_name='actual_package_count'), 'SELECT 1', 'ALTER TABLE ll_inbound_plan_pallet ADD COLUMN actual_package_count INT NOT NULL DEFAULT 0 COMMENT ''实收箱数'''));
PREPARE inbound_schema_stmt FROM @ddl; EXECUTE inbound_schema_stmt; DEALLOCATE PREPARE inbound_schema_stmt;
SET @ddl = (SELECT IF(EXISTS (SELECT 1 FROM information_schema.columns WHERE table_schema=DATABASE() AND table_name='ll_inbound_plan_pallet' AND column_name='version'), 'SELECT 1', 'ALTER TABLE ll_inbound_plan_pallet ADD COLUMN version INT NOT NULL DEFAULT 0 COMMENT ''收货累计值乐观锁版本'''));
PREPARE inbound_schema_stmt FROM @ddl; EXECUTE inbound_schema_stmt; DEALLOCATE PREPARE inbound_schema_stmt;
SET @ddl = (SELECT IF(EXISTS (SELECT 1 FROM information_schema.columns WHERE table_schema=DATABASE() AND table_name='ll_inbound_plan_pallet_allocation' AND column_name='planned_package_count'), 'SELECT 1', 'ALTER TABLE ll_inbound_plan_pallet_allocation ADD COLUMN planned_package_count INT NOT NULL DEFAULT 0 COMMENT ''计划分配箱数'''));
PREPARE inbound_schema_stmt FROM @ddl; EXECUTE inbound_schema_stmt; DEALLOCATE PREPARE inbound_schema_stmt;
SET @ddl = (SELECT IF(EXISTS (SELECT 1 FROM information_schema.columns WHERE table_schema=DATABASE() AND table_name='ll_inbound_plan_pallet_allocation' AND column_name='counted_package_count'), 'SELECT 1', 'ALTER TABLE ll_inbound_plan_pallet_allocation ADD COLUMN counted_package_count INT NOT NULL DEFAULT 0 COMMENT ''清点箱数'''));
lastleg-service/src/test/java/com/baiyue/lastleg/inboundplan/InboundPlanSchemaContractTest.java
+1
View File
@@ -21,6 +21,7 @@ class InboundPlanSchemaContractTest {
assertTrue(schema.contains("required_package_count"));
assertTrue(schema.contains("planned_package_count"));
assertTrue(schema.contains("warehouse_resource_lock_id"));
assertTrue(schema.contains("ll_inbound_plan_pallet' and column_name='version'"));
assertTrue(schema.contains("create table if not exists ll_inbound_pda_task"));
assertTrue(schema.contains("create table if not exists ll_inbound_pda_operation"));
assertTrue(schema.contains("create table if not exists ll_inbound_pallet_merge"));
lastleg-service/src/test/java/com/baiyue/lastleg/schema/Phase5OverseasUpgradeSqlContractTest.java
+1
View File
@@ -53,6 +53,7 @@ class Phase5OverseasUpgradeSqlContractTest {
"ALTER TABLE ll_dropship_order_profile ADD COLUMN manifest_package_count",
"ALTER TABLE ll_dropship_order_profile ADD COLUMN manifest_fragment_count",
"ALTER TABLE ll_dropship_order_profile ADD COLUMN manifest_hash",
"ALTER TABLE ll_inbound_plan_pallet ADD COLUMN version",
"ADD KEY idx_ll_dropship_profile_manifest(source_pallet_no,manifest_hash)",
"ADD UNIQUE KEY uk_ll_dropship_sku_inventory(profile_id,sku_code)",
"MODIFY COLUMN source_pallet_no VARCHAR(128) NOT NULL");
warehouse-overseas-service/AGENTS.md
+1 -1
View File
@@ -15,7 +15,7 @@
| 内部资源锁定、释放、报废 | `controller/InternalWarehouseResourceLockController.java` |
| PDA 托盘状态流转与补偿 | `controller/InternalPdaPalletTransitionController.java`、`service/WarehousePalletTransitionService.java`、`mapper/WarehousePalletTransitionMapper.java`；内部前缀 `/internal/v1/pda-pallet-transitions` |
| 托盘解析与业务绑定 | `controller/InternalWarehousePalletController.java`、`dto/WarehousePalletResolveResponse.java`、`entity/WarehousePalletBindingEntity.java` |
| 柜级入库计划批量托盘 | `controller/InternalWarehousePalletBatchController.java`、`service/WarehousePalletBatchService.java`、`warehouse-pallet-batch-internal-v1.yaml`；以稳定批次键幂等执行 `ensure-reserved → confirm`，只从可预留真实托盘/库位建立逐项资源锁；Lastleg 本地计划事务失败时仅补偿仍为 `RESERVED` 的未确认资源，已 `CONFIRMED` 批次不得回退，同柜重放不得产生第二批资源 |
| 柜级入库计划批量托盘 | `controller/InternalWarehousePalletBatchController.java`、`service/WarehousePalletBatchService.java`、`warehouse-pallet-batch-internal-v1.yaml`；以稳定批次键幂等执行 `ensure-reserved → confirm`，权威托盘码已存在且资源可预留时必须复用，未提供或不存在时才创建；Lastleg 本地计划事务失败时仅补偿仍为 `RESERVED` 的未确认资源，已 `CONFIRMED` 批次不得回退，同柜重放不得产生第二批资源 |
| 仓储资源服务 | `service/WarehouseResourceService.java`、`WarehouseResourceLockService.java` |
| 内部服务令牌 | `security/InternalServiceTokenGuard.java` |
| 数据库基线与中文注释契约 | `src/main/resources/db/warehouse.sql`、`src/test/java/com/baiyue/warehouse/schema/SchemaChineseCommentContractTest.java` |
warehouse-overseas-service/src/main/java/com/baiyue/warehouse/repository/WarehouseResourceRepository.java
+7
View File
@@ -130,6 +130,13 @@ public class WarehouseResourceRepository {
.orderByAsc(WarehousePalletEntity::getId));
}
public WarehousePalletEntity findPublicPalletByCode(long warehouseId, String palletCode) {
return pallets.selectOne(new LambdaQueryWrapper<WarehousePalletEntity>()
.eq(WarehousePalletEntity::getCustomerId, 0L)
.eq(WarehousePalletEntity::getWarehouseId, warehouseId)
.eq(WarehousePalletEntity::getPalletCode, palletCode));
}
public WarehousePalletEntity findPublicPallet(long palletId) {
return pallets.selectOne(new LambdaQueryWrapper<WarehousePalletEntity>()
.eq(WarehousePalletEntity::getCustomerId, 0L).eq(WarehousePalletEntity::getId, palletId));
warehouse-overseas-service/src/main/java/com/baiyue/warehouse/service/WarehousePalletBatchService.java
+10 -6
View File
@@ -88,12 +88,16 @@ public class WarehousePalletBatchService {
SharedReservation shared = isDropshipPallet(requested.palletCode())
? sharedDropship.get(requested.palletCode()) : null;
if (shared == null) {
Long palletId = resources.createPallet(WarehouseResourceService.PUBLIC_RESOURCE_SCOPE,
normalized.warehouseId(), new PalletRequests.SaveRequest(requested.palletCode(), "WOOD",
AUTO_PALLET_CAPACITY_CBM, AUTO_PALLET_UNIT_PRICE_USD, "IDLE", null, AUTO_PALLET_REMARK,
requested.fbaWarehouseCode()));
WarehousePalletEntity createdPallet = resources.pallet(WarehouseResourceService.PUBLIC_RESOURCE_SCOPE,
normalized.warehouseId(), palletId);
WarehousePalletEntity createdPallet = resources.reusablePublicPallet(
normalized.warehouseId(), requested.palletCode());
if (createdPallet == null) {
Long palletId = resources.createPallet(WarehouseResourceService.PUBLIC_RESOURCE_SCOPE,
normalized.warehouseId(), new PalletRequests.SaveRequest(requested.palletCode(), "WOOD",
AUTO_PALLET_CAPACITY_CBM, AUTO_PALLET_UNIT_PRICE_USD, "IDLE", null, AUTO_PALLET_REMARK,
requested.fbaWarehouseCode()));
createdPallet = resources.pallet(WarehouseResourceService.PUBLIC_RESOURCE_SCOPE,
normalized.warehouseId(), palletId);
}
WarehouseResourceLockResponse lock = locks.reserve(new WarehouseResourceLockRequests.ReserveRequest(
requested.customerId(), normalized.warehouseId(), createdPallet.getId(), createdPallet.getLocationId(),
BUSINESS_TYPE, requested.businessKey(), reservationKey(batch, requested.sequenceNo())));
warehouse-overseas-service/src/main/java/com/baiyue/warehouse/service/WarehouseResourceService.java
+6
View File
@@ -191,6 +191,12 @@ public class WarehouseResourceService {
return palletReference(pallet);
}
public WarehousePalletEntity reusablePublicPallet(long warehouseId, String palletCode) {
if (!StringUtils.hasText(palletCode)) return null;
requireWarehouseReference(warehouseId);
return repository.findPublicPalletByCode(warehouseId, palletCode.trim());
}
public WarehousePalletResolveResponse resolvePublicPallet(long palletId) {
WarehousePalletEntity pallet = repository.findPublicPallet(palletId);
if (pallet == null) throw new BizException(ResultCode.NOT_FOUND, "托盘不存在");
warehouse-overseas-service/src/test/java/com/baiyue/warehouse/service/WarehousePalletBatchServiceTest.java
+19
View File
@@ -132,6 +132,25 @@ class WarehousePalletBatchServiceTest {
assertThat(response.items()).extracting(value -> value.palletCode()).containsExactly("PAL-301");
}
@Test
void compensatedSourcePalletCodeReusesTheExistingIdlePhysicalPallet() {
when(batches.insertIgnore(any())).thenReturn(1);
WarehousePalletEntity existing = pallet(301L, 401L, "DP-260910-0001");
existing.setStatus("IDLE");
when(resources.reusablePublicPallet(WAREHOUSE, "DP-260910-0001")).thenReturn(existing);
when(locks.reserve(any())).thenReturn(lockResponse(CUSTOMER_A, 501L));
var request = new WarehousePalletBatchRequests.EnsureReserved(WAREHOUSE, "retry-container", List.of(
new WarehousePalletBatchRequests.Item(CUSTOMER_A, "container:dropship-a:1", "PLAN-A", 1,
"STANDARD", null, "DP-260910-0001")));
var response = service.ensureReserved(request);
assertThat(response.items()).extracting(value -> value.palletId()).containsExactly(301L);
assertThat(response.items()).extracting(value -> value.palletCode()).containsExactly("DP-260910-0001");
verify(resources, never()).createPallet(anyLong(), anyLong(), any());
verify(locks).reserve(any());
}
@Test
void concurrentSameRequestReplaysWinningBatchWithoutCreatingResources() {
AtomicReference<WarehousePalletBatchEntity> inserted = new AtomicReference<>();