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

fix(selfop): allow controlled manual collection in test

Browse Source
This commit is contained in:
ouyangxq
2026-10-12 23:21:57 +08:00
parent 6e9f4b7932
commit 86e05a77e6
15 changed files with 160 additions and 20 deletions
Download Patch File Download Diff File Expand all files Collapse all files
lastleg-service/src/main/java/com/baiyue/lastleg/selfoperated/analytics/model/SelfopAnalyticsProperties.java
+1 -1
View File
@@ -35,7 +35,7 @@ public class SelfopAnalyticsProperties {
private int minimumBuildIntervalSeconds = 1800;
/** Read-only UI projection; collectionMode is the sole runtime mode input. */
public boolean isExecutionEnabled() {
return "VALIDATION".equals(collectionMode)||("BUSINESS".equals(collectionMode)&&"production".equals(environment));
return "VALIDATION".equals(collectionMode)||("BUSINESS".equals(collectionMode)&&SelfopCollectionPolicy.businessEnvironment(this));
}
private boolean salesPublicationEnabled = false;
private String integrationBaseUrl = "http://integration-service";
lastleg-service/src/main/java/com/baiyue/lastleg/selfoperated/analytics/model/SelfopCollectionPolicy.java
+32 -1
View File
@@ -19,7 +19,38 @@ public final class SelfopCollectionPolicy {
public static void requireExecution(SelfopAnalyticsProperties config,String purpose) {
require("VALIDATION".equals(purpose)||"BUSINESS".equals(purpose),"SELFOP_PURPOSE_INVALID");
require(purpose.equals(config.getCollectionMode()),"SELFOP_COLLECTION_MODE_REJECTED");
if("BUSINESS".equals(purpose))require("production".equals(config.getEnvironment()),"SELFOP_BUSINESS_ENVIRONMENT_REJECTED");
if("BUSINESS".equals(purpose))require(businessEnvironment(config),"SELFOP_BUSINESS_ENVIRONMENT_REJECTED");
}
public static boolean businessEnvironment(SelfopAnalyticsProperties config) {
return "test".equals(config.getEnvironment())||"production".equals(config.getEnvironment());
}
public static void requireAutomaticPlanning(SelfopAnalyticsProperties config) {
requireExecution(config,"BUSINESS");
require("production".equals(config.getEnvironment()),"SELFOP_AUTOMATIC_PLANNING_ENVIRONMENT_REJECTED");
}
/** TEST may poll only persisted manual roots and their existing routing/refresh descendants. */
public static boolean controlledBusinessTask(SelfopSyncTask task,java.util.function.Function<Long,SelfopSyncTask> lookup) {
if(task==null)return false;
String account=task.getSourceAccount();var seen=new java.util.HashSet<Long>();
for(int depth=0;task!=null&&depth<64;depth++) {
if(!"BUSINESS".equals(task.getExecutionPurpose())||task.getValidationRootTaskId()!=null
||!java.util.Objects.equals(account,task.getSourceAccount())||"SCHEDULED".equals(task.getTaskType()))return false;
if(task.getId()!=null&&!seen.add(task.getId()))return false;
String key=task.getRequestKey();
if(key!=null&&key.startsWith("manual:")&&task.getRequestedBy()!=null&&!task.getRequestedBy().isBlank())return true;
Long parent=task.getUpstreamTaskId();
if(parent!=null) {
if(!"WAREHOUSE_SYNC".equals(task.getTaskRole())||!java.util.Objects.equals(key,"route:"+parent+":"+task.getWarehouseCode()))return false;
}else {
if(!"ACCOUNT_DISCOVERY".equals(task.getTaskRole())||!"RECHECK".equals(task.getTaskType())||key==null)return false;
Map<?,?> query=read(task.getQueryScopeJson(),Map.class);
if(query==null||!(query.get("parentTaskId") instanceof Number id))return false;
parent=id.longValue();
if(!key.startsWith("refresh:"+parent+":")&&!key.startsWith("relocate:"+parent+":"))return false;
}
task=lookup.apply(parent);
}
return false;
}
public static String configuredPurpose(SelfopAnalyticsProperties config) {
String purpose=config.getCollectionMode();requireExecution(config,purpose);return purpose;
lastleg-service/src/main/java/com/baiyue/lastleg/selfoperated/analytics/query/CollectionQueryService.java
+4 -3
View File
@@ -86,7 +86,7 @@ public class CollectionQueryService {
Task shared="OUTBOUND".equals(family)&&sharedWindowRelated(active,scope)?active:null;
String blocked=executionBlock(warehouse,protection);
boolean enabled=blocked==null;
boolean publication="BUSINESS".equals(config.getCollectionMode())&&"production".equals(config.getEnvironment())
boolean publication="BUSINESS".equals(config.getCollectionMode())&&config.isExecutionEnabled()
&&("INVENTORY".equals(family)||config.isSalesPublicationEnabled());
String initial="INVENTORY".equals(family)?published!=null&&"COMPLETE".equals(published.coverageStatus())?"COMPLETE":"UNKNOWN"
:scope==null||scope.initialFromUtc()==null||scope.initialUntilUtc()==null?"UNKNOWN"
@@ -97,7 +97,7 @@ public class CollectionQueryService {
String freshness=published==null?"NOT_COLLECTED":scope.staleAfterSeconds()==null||scope.staleAfterSeconds()<=0?"UNKNOWN"
:observation==null?"UNKNOWN":!clock.instant().isBefore(observation.plusSeconds(scope.staleAfterSeconds()))?"STALE":"FRESH";
rows.add(new CollectionResponses.Warehouse(warehouse,warehouse,family,enabled,publication,blocked,
enabled?nextSchedule(warehouse,family):null,clock.instant(),scope==null?null:utc(scope.initialFromUtc()),
enabled&&"production".equals(config.getEnvironment())?nextSchedule(warehouse,family):null,clock.instant(),scope==null?null:utc(scope.initialFromUtc()),
scope==null?null:utc(scope.initialUntilUtc()),initial,published==null?null:utc(published.coveredFromUtc()),
published==null?null:utc(published.coveredUntilUtc()),last==null?null:last.taskNo(),last==null||context.latest().get(last.id())==null?null:utc(context.latest().get(last.id()).finishedAtUtc()),
published==null?null:utc(published.publishedAtUtc()),"INVENTORY".equals(family)&&published!=null?utc(published.fetchStartedAtUtc()):null,
@@ -234,6 +234,7 @@ public class CollectionQueryService {
if(!"WAREHOUSE_SYNC".equals(task.taskRole()))return "ACCOUNT_SHARED_TASK";
String disabled=executionBlock(task.warehouseCode(),protection);if(disabled!=null)return disabled;
if(!"BUSINESS".equals(config.getCollectionMode()))return "COLLECTION_MODE_NOT_BUSINESS";
if("test".equals(config.getEnvironment())&&"SCHEDULED".equals(task.taskType()))return "TEST_MANUAL_TASK_REQUIRED";
if(!Set.of("DEAD","RETRY_WAIT").contains(task.status()))return "TASK_NOT_RETRYABLE";
if(scope==null||!Objects.equals(scope.scopeHash(),task.scopeHash()))return "SCOPE_CHANGED";
if(scope.leaseOwner()!=null&&scope.leaseUntilUtc()!=null&&clock.instant().isBefore(utc(scope.leaseUntilUtc())))return "SCOPE_LEASE_BUSY";
@@ -273,7 +274,7 @@ public class CollectionQueryService {
private String executionBlock(String warehouse,CollectionResponses.Protection protection) {
if("OFF".equals(config.getCollectionMode()))return "COLLECTION_OFF";
if(!Set.of("BUSINESS","VALIDATION").contains(config.getCollectionMode()))return "COLLECTION_MODE_INVALID";
if("BUSINESS".equals(config.getCollectionMode())&&!"production".equals(config.getEnvironment()))return "BUSINESS_ENVIRONMENT_REJECTED";
if("BUSINESS".equals(config.getCollectionMode())&&!com.baiyue.lastleg.selfoperated.analytics.model.SelfopCollectionPolicy.businessEnvironment(config))return "BUSINESS_ENVIRONMENT_REJECTED";
if(!config.isExecutionEnabled())return "EXECUTION_DISABLED";
if(!config.getWarehouses().containsKey(warehouse))return "WAREHOUSE_NOT_CONFIGURED";
var w=config.getWarehouses().get(warehouse);
lastleg-service/src/main/java/com/baiyue/lastleg/selfoperated/analytics/request/SelfopEgressService.java
+1 -1
View File
@@ -110,7 +110,7 @@ public class SelfopEgressService {
public void finishSample(Long rootTaskId,String reason){var root=mapper.lockTask(rootTaskId);var v=validation(root);if("DEAD".equals(root.getStatus()))return;stop(root,v,safeReason(reason),false,now());}
private void gate(String account,String purpose,String grant,LocalDateTime now){
if("OFF".equals(config.getCollectionMode())||!Objects.equals(config.getCollectionMode(),purpose)||!("VALIDATION".equals(purpose)||"BUSINESS".equals(purpose)))throw reject("EGRESS_MODE_REJECTED");
if("BUSINESS".equals(purpose)&&!"production".equals(config.getEnvironment()))throw reject("BUSINESS_ENVIRONMENT_REJECTED");
if("BUSINESS".equals(purpose)&&!("production".equals(config.getEnvironment())||"test".equals(config.getEnvironment())))throw reject("BUSINESS_ENVIRONMENT_REJECTED");
if(!Objects.equals(config.getSourceAccount(),account)||account==null||account.isBlank())throw reject("SOURCE_ACCOUNT_MISMATCH");
if(grant==null||!Objects.equals(grant,config.getExclusiveGrantId())||config.getExclusiveGrantExpiresAtUtc()==null||!now.isBefore(config.getExclusiveGrantExpiresAtUtc())||!config.isExclusiveHandoffVerified()||!config.isPriorEgressClosed()||!config.isInflightDrained()||!config.isTransferredBudgetVerified())throw reject("EXCLUSIVE_HANDOFF_UNCONFIRMED");
if(config.getDailyRequestBudget()<1||config.getBusinessMinRequestIntervalMs()<1||config.getRequestBudget()<1||config.getTimeBudgetSeconds()<1||config.getMaxRateLimitFailures()<1||config.getCooldownSeconds()<1)throw reject("ACCOUNT_BUDGET_UNCONFIGURED");
lastleg-service/src/main/java/com/baiyue/lastleg/selfoperated/analytics/sync/SelfopCommandService.java
+2
View File
@@ -56,6 +56,7 @@ public class SelfopCommandService {
require("WAREHOUSE_SYNC".equals(original.getTaskRole())&&c.warehouseCode().equals(original.getWarehouseCode()),"SELFOP_TASK_SCOPE_MISMATCH");
require(!"WAITING_SOURCE".equals(original.getStatus()),"SELFOP_RETRY_WAITING_SOURCE");
require("BUSINESS".equals(original.getExecutionPurpose())&&"BUSINESS".equals(config.getCollectionMode()),"SELFOP_BUSINESS_PURPOSE_REQUIRED");
require(!"test".equals(config.getEnvironment())||SelfopCollectionPolicy.controlledBusinessTask(original,tasks::selectById),"SELFOP_TEST_MANUAL_TASK_REQUIRED");
return taskService.retry(original.getId(),c.requestKey(),operator,c.reason());}
String purpose=SelfopCollectionPolicy.configuredPurpose(config);
require(("VALIDATION".equals(purpose))==("VALIDATION".equals(c.action())),"SELFOP_VALIDATION_ENTRY_REQUIRED");
@@ -118,5 +119,6 @@ public class SelfopCommandService {
Map.entry("SELFOP_TASK_ALREADY_SUCCEEDED","任务已成功完成，无需重试"),
Map.entry("SELFOP_COLLECTION_MODE_REJECTED","当前采集模式不允许此操作"),
Map.entry("SELFOP_BUSINESS_ENVIRONMENT_REJECTED","当前环境不允许正式采集"),
Map.entry("SELFOP_TEST_MANUAL_TASK_REQUIRED","测试环境仅允许重试人工采集任务及其派生任务"),
Map.entry("SELFOP_BUDGET_INVALID","采集预算配置无效，请联系管理员"));
}
lastleg-service/src/main/java/com/baiyue/lastleg/selfoperated/analytics/sync/SelfopSchedulePlanner.java
+8 -3
View File
@@ -34,6 +34,7 @@ public class SelfopSchedulePlanner {
if("BUSINESS".equals(config.getCollectionMode()))SelfopCollectionPolicy.requireExecution(config,"BUSINESS");
List<SelfopSyncScope> configured;
if("VALIDATION".equals(config.getCollectionMode())){taskService.initializeAccount();configured=List.of();}
else if("test".equals(config.getEnvironment()))configured=List.of();
else configured=taskService.initialize();
var now=utcNow();
for(var scope:configured){
@@ -41,10 +42,14 @@ public class SelfopSchedulePlanner {
if("INVENTORY".equals(scope.getDataFamily()))plan(scope,now);
else if(config.getSourceQueryTimezone()!=null&&Objects.equals(scope.getSourceTimezone(),config.getSourceQueryTimezone())){plan(scope,now);review(scope,now);}
}
for(var task:taskService.due(10))try{execution.execute(task.getId());}catch(IllegalStateException leaseBusy){/* Another worker owns the durable lease. */}
for(var task:taskService.due(10)){
if("BUSINESS".equals(config.getCollectionMode())&&"test".equals(config.getEnvironment())
&&!SelfopCollectionPolicy.controlledBusinessTask(task,tasks::selectById))continue;
try{execution.execute(task.getId());}catch(IllegalStateException leaseBusy){/* Another worker owns the durable lease. */}
}
}
public void plan(SelfopSyncScope preview,LocalDateTime now){
SelfopCollectionPolicy.requireExecution(config,"BUSINESS");
SelfopCollectionPolicy.requireAutomaticPlanning(config);
tx.executeWithoutResult(s->{var scope=locks.lockScope(preview.getId());ZoneId business=ZoneId.of(scope.getBusinessTimezone());
if("OUTBOUND".equals(scope.getDataFamily())&&scope.getInitialUntilUtc()==null){
scope.setInitialUntilUtc(now);scope.setInitialFromUtc(now.atZone(ZoneOffset.UTC).withZoneSameInstant(business).minusMonths(2).withZoneSameInstant(ZoneOffset.UTC).toLocalDateTime());
@@ -91,7 +96,7 @@ public class SelfopSchedulePlanner {
t.setRequestHash(hash(Arrays.asList(t.getQueryScopeJson(),t.getScopeHash())));return taskService.create(t);
}
public void review(SelfopSyncScope preview,LocalDateTime now){
SelfopCollectionPolicy.requireExecution(config,"BUSINESS");
SelfopCollectionPolicy.requireAutomaticPlanning(config);
tx.executeWithoutResult(s->{var scope=locks.lockScope(preview.getId());
if(scope.getKnownOrderRoundMaxId()==null){var max=orders.selectOne(new QueryWrapper<SelfopOutboundOrder>().eq("source_account",scope.getSourceAccount()).eq("warehouse_code",scope.getWarehouseCode()).orderByDesc("id").last("LIMIT 1"));
scope.setKnownOrderRoundMaxId(max==null?0L:max.getId());scope.setKnownOrderCursorId(0L);scope.setKnownOrderRoundStartedAtUtc(now);}