From 674914c65194e259d4cf697d49afb8ca3c6690e6 Mon Sep 17 00:00:00 2001 From: ouyangxq <775153912@qq.com> Date: Wed, 14 Oct 2026 09:46:49 +0800 Subject: [PATCH] fix(selfop): allow explicit retry of failed continuous inventory roots --- .../analytics/sync/SelfopTaskService.java | 26 +++++++++---- .../analytics/sync/SelfopTaskServiceTest.java | 37 +++++++++++++++++++ 2 files changed, 56 insertions(+), 7 deletions(-) diff --git a/lastleg-service/src/main/java/com/baiyue/lastleg/selfoperated/analytics/sync/SelfopTaskService.java b/lastleg-service/src/main/java/com/baiyue/lastleg/selfoperated/analytics/sync/SelfopTaskService.java index 0a9986582..111b0fbb5 100644 --- a/lastleg-service/src/main/java/com/baiyue/lastleg/selfoperated/analytics/sync/SelfopTaskService.java +++ b/lastleg-service/src/main/java/com/baiyue/lastleg/selfoperated/analytics/sync/SelfopTaskService.java @@ -224,18 +224,23 @@ public class SelfopTaskService { lockBusinessAccount(id); var commandQuery=new QueryWrapper().eq("source_account",preview.getSourceAccount()).eq("request_key","retry:"+requestKey); var prior=tasks.selectOne(commandQuery); - if(prior!=null){require(prior.getRequestHash().equals(hash(List.of(id,requestKey,reason))),"SELFOP_IDEMPOTENCY_CONFLICT");businessGuard(preview);return preview;} + if(prior!=null){require(prior.getRequestHash().equals(hash(List.of(id,requestKey,reason))),"SELFOP_IDEMPOTENCY_CONFLICT");businessGuard(preview,false);return preview;} // Match executor lock ordering: account/warehouse scope first, task second. String activeOwner;LocalDateTime activeUntil; if("ACCOUNT_DISCOVERY".equals(preview.getTaskRole())){var scope=locks.lockDiscovery(preview.getDiscoveryScopeId());require(scope!=null,"SELFOP_SCOPE_MISSING");activeOwner=scope.getLeaseOwner();activeUntil=scope.getLeaseUntilUtc();} else{var scope=locks.lockScope(preview.getScopeId());require(scope!=null,"SELFOP_SCOPE_MISSING");activeOwner=scope.getLeaseOwner();activeUntil=scope.getLeaseUntilUtc();} prior=tasks.selectOne(new QueryWrapper().eq("source_account",preview.getSourceAccount()).eq("request_key","retry:"+requestKey).last("FOR UPDATE")); if(prior!=null){require(prior.getRequestHash().equals(hash(List.of(id,requestKey,reason))),"SELFOP_IDEMPOTENCY_CONFLICT");return tasks.selectById(id);} - businessGuard(preview); + businessGuard(preview,false,true); require(SelfopLeaseService.available(activeOwner,activeUntil,utcNow()),"SELFOP_RETRY_LEASE_BUSY"); var original=locks.lockTask(id);require(original!=null,"SELFOP_TASK_MISSING"); requireBusiness(original);requireExecution(config,original.getExecutionPurpose()); require(!"SUCCEEDED".equals(original.getStatus()),"SELFOP_TASK_ALREADY_SUCCEEDED"); + // The audit creation checks the parent again. Persist the authorized resume in this same + // transaction before creating it; any audit failure rolls the resume back as well. + original.setStatus("READY");original.setNextRetryAtUtc(utcNow()); + if(!continuousBusinessRoot(original))original.setRequestedBy(operator); + original.setUpdatedAt(utcNow());tasks.updateById(original); // Resume original logical task, retaining apply barrier and stage set. Record the manual command separately. var audit=new SelfopSyncTask();audit.setSourceAccount(original.getSourceAccount());audit.setTaskRole(original.getTaskRole());audit.setScopeId(original.getScopeId()); audit.setExecutionPurpose(original.getExecutionPurpose()); @@ -243,9 +248,6 @@ public class SelfopTaskService { audit.setRequestKey("retry:"+requestKey);audit.setRequestHash(hash(List.of(id,requestKey,reason)));audit.setTaskType("MANUAL_RETRY");audit.setScopeHash(original.getScopeHash()); audit.setQueryScopeJson(json(Map.of("resumesTaskNo",original.getTaskNo(),"reason",reason)));audit.setStatus("SUCCEEDED");audit.setApplyState("NOT_APPLICABLE");audit.setRequestedBy(operator); create(audit); - original.setStatus("READY");original.setNextRetryAtUtc(utcNow()); - if(!continuousBusinessRoot(original))original.setRequestedBy(operator); - original.setUpdatedAt(utcNow());tasks.updateById(original); return original; } public List due(int limit){return continuousBusiness(config)?locks.dueContinuous(utcNow(),limit,config.getSourceAccount()):locks.due(utcNow(),limit,config.getCollectionMode());} @@ -301,7 +303,8 @@ public class SelfopTaskService { var task=tasks.selectById(id);businessGuard(task); } private void businessGuard(SelfopSyncTask task){businessGuard(task,true);} - private void businessGuard(SelfopSyncTask task,boolean terminate){ + private void businessGuard(SelfopSyncTask task,boolean terminate){businessGuard(task,terminate,false);} + private void businessGuard(SelfopSyncTask task,boolean terminate,boolean inventoryRetry){ if(!testBusiness(task))return; require(controlledBusinessTask(task,tasks::selectById,continuousBusiness(config)),"SELFOP_TEST_MANUAL_TASK_REQUIRED"); require(task.getBusinessRootTaskId()!=null,"BUSINESS_SESSION_ROOT_REQUIRED"); @@ -318,7 +321,7 @@ public class SelfopTaskService { var budget=read(root.getBusinessBudgetJson(),SelfopBusinessBudget.class);budget.validate(); require(!budget.continuous||(continuousBusiness(config)&&continuousBusinessRoot(root)),"SELFOP_CONTINUOUS_COLLECTION_DISABLED"); String reason=budget.stopReason; - if(reason==null&&"DEAD".equals(root.getStatus()))reason="BUSINESS_SESSION_STOPPED"; + if(reason==null&&"DEAD".equals(root.getStatus())&&!(inventoryRetry&&failedInventoryScanRoot(root)))reason="BUSINESS_SESSION_STOPPED"; if(reason==null)return; if(!terminate)throw new BusinessSessionStoppedException(reason); var now=utcNow();budget.stopReason=reason; @@ -331,6 +334,15 @@ public class SelfopTaskService { .set("status","DEAD").set("next_retry_at_utc",null).set("updated_at",now)); throw new BusinessSessionStoppedException(reason); } + private boolean failedInventoryScanRoot(SelfopSyncTask root){ + if(!continuousBusiness(config)||!continuousBusinessRoot(root)||!"WAREHOUSE_SYNC".equals(root.getTaskRole()))return false; + var scope=scopes.selectById(root.getScopeId()); + if(scope==null||!"INVENTORY".equals(scope.getDataFamily()))return false; + var latest=runs.selectOne(new QueryWrapper().eq("task_id",root.getId()) + .orderByDesc("attempt_no","id").last("LIMIT 1")); + return latest!=null&&"FAILED".equals(latest.getStatus()) + &&Set.of("SELFOP_STOCK_DUPLICATE","SELFOP_STOCK_COUNT_MISMATCH","SELFOP_PAGE_TOTAL_CHANGED").contains(latest.getErrorCode()); + } private static void checkTaskLease(SelfopSyncTask task,SelfopLeaseService.Lease lease){ require(task!=null&&(lease.account()?"ACCOUNT_DISCOVERY".equals(task.getTaskRole())&&Objects.equals(task.getDiscoveryScopeId(),lease.scopeId()) :"WAREHOUSE_SYNC".equals(task.getTaskRole())&&Objects.equals(task.getScopeId(),lease.scopeId())),"SELFOP_TASK_LEASE_SCOPE_MISMATCH"); diff --git a/lastleg-service/src/test/java/com/baiyue/lastleg/selfoperated/analytics/sync/SelfopTaskServiceTest.java b/lastleg-service/src/test/java/com/baiyue/lastleg/selfoperated/analytics/sync/SelfopTaskServiceTest.java index f1d89dfb2..1814fb3db 100644 --- a/lastleg-service/src/test/java/com/baiyue/lastleg/selfoperated/analytics/sync/SelfopTaskServiceTest.java +++ b/lastleg-service/src/test/java/com/baiyue/lastleg/selfoperated/analytics/sync/SelfopTaskServiceTest.java @@ -506,6 +506,43 @@ class SelfopTaskServiceTest { service.fail(1L,20L,"SELFOP_STOCK_DUPLICATE",false,lease); assertThat(root.getStatus()).isEqualTo("DEAD"); } + @Test void commandRetriesDeadContinuousInventoryThroughRealTaskServiceAndRecordsAudit(){ + var root=continuousInventoryRoot();root.setStatus("DEAD");root.setAttemptCount(2); + var failed=run("FAILED");failed.setErrorCode("SELFOP_STOCK_DUPLICATE"); + when(runs.selectOne(any())).thenReturn(failed); + when(tasks.selectOne(any())).thenAnswer(inv->{var q=(QueryWrapper)inv.getArgument(0); + return q.getSqlSegment().contains("task_no")?root:null;}); + config.setSourceAccount("synthetic-account");config.getWarehouses().put("WH-A",new SelfopAnalyticsProperties.Warehouse()); + String budget=root.getBusinessBudgetJson();var planned=root.getPlannedAtUtc(); + var commands=new SelfopCommandService(service,scopes,accounts,config,tasks); + var command=new SelfopCommandService.Command("recover-stock","WH-A","RETRY",root.getTaskNo(),null,null,null,null,"retry completed unstable inventory scan"); + assertThat(commands.request(command,"operator")).isSameAs(root); + assertThat(root.getStatus()).isEqualTo("READY");assertThat(root.getAttemptCount()).isEqualTo(2); + assertThat(root.getBusinessBudgetJson()).isEqualTo(budget);assertThat(root.getPlannedAtUtc()).isEqualTo(planned); + assertThat(root.getRequestedBy()).isEqualTo(SelfopCollectionPolicy.CONTINUOUS_PLANNER); + var order=inOrder(tasks);order.verify(tasks).updateById(root); + order.verify(tasks).insert(argThat((SelfopSyncTask audit)->"MANUAL_RETRY".equals(audit.getTaskType()) + &&"SUCCEEDED".equals(audit.getStatus())&&Long.valueOf(1).equals(audit.getUpstreamTaskId()) + &&Long.valueOf(1).equals(audit.getBusinessRootTaskId()))); + } + @Test void explicitInventoryRetryCannotClearPersistedStopOrRetryUnknownFailure(){ + for(String error:List.of("LINGXING_WMS_RATE_LIMITED","INTEGRATION_TRANSPORT_FAILURE","SELFOP_EXECUTION_FAILED")){ + var root=continuousInventoryRoot();root.setStatus("DEAD");String budget=root.getBusinessBudgetJson(); + var failed=run("FAILED");failed.setErrorCode(error);when(runs.selectOne(any())).thenReturn(failed); + assertThatThrownBy(()->service.retry(1L,"retry-protected","operator","offline retry")) + .isInstanceOf(SelfopTaskService.BusinessSessionStoppedException.class); + assertThat(root.getStatus()).isEqualTo("DEAD");assertThat(root.getBusinessBudgetJson()).isEqualTo(budget); + } + var root=continuousInventoryRoot();root.setStatus("DEAD"); + var budget=read(root.getBusinessBudgetJson(),com.baiyue.lastleg.selfoperated.analytics.request.SelfopBusinessBudget.class); + budget.stopReason="BUSINESS_SESSION_STOPPED";root.setBusinessBudgetJson(json(budget));String frozen=root.getBusinessBudgetJson(); + var failed=run("FAILED");failed.setErrorCode("SELFOP_STOCK_DUPLICATE");when(runs.selectOne(any())).thenReturn(failed); + assertThatThrownBy(()->service.retry(1L,"retry-stopped","operator","offline retry")) + .isInstanceOf(SelfopTaskService.BusinessSessionStoppedException.class); + assertThat(root.getBusinessBudgetJson()).isEqualTo(frozen); + verify(tasks,never()).updateById(any(SelfopSyncTask.class));verify(tasks,never()).insert(any(SelfopSyncTask.class)); + verify(tasks,never()).update(any(),any()); + } private SelfopSyncTask continuousInventoryRoot(){ var root=boundedRoot(0,utcNow().plusMinutes(3));config.setContinuousBusinessCollectionEnabled(true); root.setTaskType("SCHEDULED");root.setStatus("RUNNING");root.setApplyState("NOT_APPLICABLE");