From cbc316b34d3858577b80834df65268de272e5ed8 Mon Sep 17 00:00:00 2001 From: ouyangxq <775153912@qq.com> Date: Mon, 12 Oct 2026 22:57:18 +0800 Subject: [PATCH] fix(selfop): expose scoped validation tasks without published scopes --- .../query/CollectionQueryService.java | 45 ++++++++--- .../analytics/query/CollectionReadMapper.java | 51 +++++++++--- .../openapi/self-operated-analytics-v1.yaml | 8 +- .../query/CollectionQueryServiceTest.java | 81 ++++++++++++++++++- 4 files changed, 160 insertions(+), 25 deletions(-) diff --git a/lastleg-service/src/main/java/com/baiyue/lastleg/selfoperated/analytics/query/CollectionQueryService.java b/lastleg-service/src/main/java/com/baiyue/lastleg/selfoperated/analytics/query/CollectionQueryService.java index 600117510..3a62e0a44 100644 --- a/lastleg-service/src/main/java/com/baiyue/lastleg/selfoperated/analytics/query/CollectionQueryService.java +++ b/lastleg-service/src/main/java/com/baiyue/lastleg/selfoperated/analytics/query/CollectionQueryService.java @@ -62,8 +62,13 @@ public class CollectionQueryService { List scopeIds=scopes.stream().filter(s->config.getWarehouses().containsKey(s.warehouseCode())).map(Scope::id).toList(); List currentTasks=scopeIds.isEmpty()?List.of():mapper.currentTasks(query.sourceAccount(),scopeIds); List successes=scopeIds.isEmpty()?List.of():mapper.lastSuccessfulTasks(query.sourceAccount(),scopeIds); + List visibleWarehouses=config.getWarehouses().keySet().stream() + .filter(w->query.warehouseCode()==null||w.equals(query.warehouseCode())).sorted().toList(); + List validationTasks=visibleWarehouses.isEmpty()?List.of():mapper.latestValidationTasks(query.sourceAccount(),visibleWarehouses); + validationTasks=validationTasks.stream().filter(t->validationAccountRoot(t)&&query.sourceAccount().equals(t.sourceAccount()) + &&visibleWarehouses.contains(t.warehouseCode())).toList(); Task active=mapper.sharedActiveAccount(query.sourceAccount()); - List all=new ArrayList<>(currentTasks);all.addAll(successes);if(active!=null)all.add(active); + List all=new ArrayList<>(currentTasks);all.addAll(successes);all.addAll(validationTasks);if(active!=null)all.add(active); Context context=context(query.sourceAccount(),all,scopes.stream().map(Scope::publishedRunId).filter(Objects::nonNull).toList()); Set barriers=config.getWarehouses().isEmpty()?Set.of():new HashSet<>(mapper.barrierWarehouses(query.sourceAccount(),new ArrayList<>(config.getWarehouses().keySet()))); for(String warehouse:new TreeSet<>(config.getWarehouses().keySet())) { @@ -72,6 +77,8 @@ public class CollectionQueryService { if(query.dataFamily()!=null&&!family.equals(query.dataFamily()))continue; Scope scope=persisted.get(warehouse+":"+family); Task current=currentTasks.stream().filter(t->warehouse.equals(t.warehouseCode())&&family.equals(t.dataFamily())).findFirst().orElse(null); + Task sampled=validationTasks.stream().filter(t->warehouse.equals(t.warehouseCode())).findFirst().orElse(null); + if(sampled!=null&&(current==null||sampled.createdAt()!=null&¤t.createdAt()!=null&&sampled.createdAt().isAfter(current.createdAt())))current=sampled; Task last=successes.stream().filter(t->warehouse.equals(t.warehouseCode())&&family.equals(t.dataFamily())).findFirst().orElse(null); Attempt published=scope==null?null:context.published().get(scope.publishedRunId()); if(published!=null&&(!warehouse.equals(published.warehouseCode())||!Objects.equals(scope.id(),published.scopeId()) @@ -96,7 +103,7 @@ public class CollectionQueryService { published==null?null:utc(published.publishedAtUtc()),"INVENTORY".equals(family)&&published!=null?utc(published.fetchStartedAtUtc()):null, "INVENTORY".equals(family)&&published!=null?utc(published.fetchFinishedAtUtc()):null,freshness, "OUTBOUND".equals(family)&&scope!=null&&(Integer.valueOf(1).equals(scope.rebuildRequired())||barriers.contains(warehouse)),reviewStatus(scope), - current==null?null:taskView(current,scope,protection,context),shared==null?null:taskView(shared,null,protection,context),protection, + current==null?null:taskView(current,scope,protection,context,family),shared==null?null:taskView(shared,null,protection,context),protection, "INVENTORY".equals(family)?inventorySamples.get(warehouse):null)); } } @@ -132,7 +139,7 @@ public class CollectionQueryService { if(bound==null)return empty(query); var scopes=scopeMap(query); var protection=protection(query.sourceAccount()); List tasks=mapper.tasks(f,bound);Context context=context(query.sourceAccount(),tasks,List.of()); - var rows=tasks.stream().map(t->taskView(t,scopes.get(t.dataFamily()),protection,context)).toList(); + var rows=tasks.stream().map(t->{requireTaskScope(t,query);return taskView(t,scopes.get(t.dataFamily()),protection,context,query.dataFamily());}).toList(); return new CollectionResponses.Page<>(current(query),size(query),mapper.taskCount(f,bound),utc(bound.createdAt()),bound.id(),clock.instant(),rows); } public CollectionResponses.TaskDetail taskDetail(String taskNo,CollectionQuery query) { @@ -140,12 +147,13 @@ public class CollectionQueryService { if(taskNo==null||taskNo.isBlank()||taskNo.length()>64)throw new BizException("任务号无效"); Task task=mapper.task(query.sourceAccount(),query.warehouseCode(),taskNo); if(task==null)throw new BizException("任务不存在或不属于所选仓库"); + requireTaskScope(task,query); Scope scope=scopeMap(query).get(task.dataFamily()); - if(scope==null||(query.dataFamily()!=null&&!query.dataFamily().equals(task.dataFamily())))throw new BizException("任务范围与所选仓库不一致"); + if(scope==null&&!validationAccountRoot(task))throw new BizException("任务范围与所选仓库不一致"); Context context=context(query.sourceAccount(),List.of(task),List.of()); Task shared=relatedUpstream(task,context); var protection=protection(query.sourceAccount()); - return new CollectionResponses.TaskDetail(taskView(task,scope,protection,context),shared==null?null:taskView(shared,null,protection,context),protection,clock.instant()); + return new CollectionResponses.TaskDetail(taskView(task,scope,protection,context,query.dataFamily()),shared==null?null:taskView(shared,null,protection,context),protection,clock.instant()); } public CollectionResponses.Page runs(CollectionQuery query) { validate(query,true); @@ -156,7 +164,11 @@ public class CollectionQueryService { if(!taskIds.isEmpty())for(Task t:mapper.tasksByIds(query.sourceAccount(),taskIds))tasks.put(t.id(),t); Map scopes=scopeMap(query);var protection=protection(query.sourceAccount()); Context context=context(query.sourceAccount(),new ArrayList<>(tasks.values()),List.of()); - var rows=runs.stream().map(r->{Task t=tasks.get(r.taskId());return attemptView(r,t, + var rows=runs.stream().map(r->{Task t=tasks.get(r.taskId()); + if(t!=null)requireTaskScope(t,query); + if("VALIDATION".equals(r.executionPurpose())&&(t==null||!query.warehouseCode().equals(r.warehouseCode()))) + throw new BizException("尝试范围与所选仓库不一致"); + return attemptView(r,t, t!=null&&retryBlock(t,scopes.get(t.dataFamily()),protection,context)==null);}).toList(); return new CollectionResponses.Page<>(current(query),size(query),mapper.runCount(f,bound),utc(bound.createdAt()),bound.id(),clock.instant(),rows); } @@ -186,9 +198,24 @@ public class CollectionQueryService { private Task relatedUpstream(Task task,Context context) { if(task.upstreamTaskId()==null)return null; Task parent=context.tasks().get(task.upstreamTaskId()); - return parent!=null&&"ACCOUNT_DISCOVERY".equals(parent.taskRole())&&Objects.equals(task.sourceAccount(),parent.sourceAccount())?parent:null; + return parent!=null&&"ACCOUNT_DISCOVERY".equals(parent.taskRole())&&Objects.equals(task.sourceAccount(),parent.sourceAccount()) + &&(!"VALIDATION".equals(parent.executionPurpose())||validationAccountRoot(parent)&&Objects.equals(task.warehouseCode(),parent.warehouseCode()))?parent:null; + } + private static boolean validationAccountRoot(Task task) { + return task!=null&&"VALIDATION".equals(task.executionPurpose())&&"VALIDATION".equals(task.taskType()) + &&"ACCOUNT_DISCOVERY".equals(task.taskRole())&&Objects.equals(task.id(),task.validationRootTaskId()) + &&task.warehouseCode()!=null&&!task.warehouseCode().isBlank()&&"OUTBOUND".equals(task.dataFamily()); + } + private void requireTaskScope(Task task,CollectionQuery query) { + if(!query.sourceAccount().equals(task.sourceAccount())||!query.warehouseCode().equals(task.warehouseCode()) + ||"ACCOUNT_DISCOVERY".equals(task.taskRole())&&!validationAccountRoot(task) + ||query.dataFamily()!=null&&!query.dataFamily().equals(task.dataFamily())&&!validationAccountRoot(task)) + throw new BizException("任务范围与所选仓库不一致"); } private CollectionResponses.Task taskView(Task t,Scope scope,CollectionResponses.Protection protection,Context context) { + return taskView(t,scope,protection,context,null); + } + private CollectionResponses.Task taskView(Task t,Scope scope,CollectionResponses.Protection protection,Context context,String family) { Attempt latest=context.latest().get(t.id()); String blocked=retryBlock(t,scope,protection,context); Task root=t.validationRootTaskId()==null?null:context.tasks().get(t.validationRootTaskId()); @@ -196,15 +223,15 @@ public class CollectionQueryService { String resultState=t.resultPublishedRunId()==null?"NOT_PUBLISHED":result==null||scope==null||!Objects.equals(scope.id(),result.scopeId()) ||!Objects.equals(t.warehouseCode(),result.warehouseCode())||!Objects.equals(t.scopeHash(),result.scopeHash())?"RESULT_UNAVAILABLE" :latest!=null&&"REUSED".equals(latest.status())?"REUSED":latest!=null&&"PUBLISHED".equals(latest.status())?"PUBLISHED":"RESULT_AVAILABLE"; - return new CollectionResponses.Task(t.taskNo(),t.taskType(),t.dataFamily(),t.warehouseCode(),t.taskRole(),t.executionPurpose(), + return new CollectionResponses.Task(t.taskNo(),t.taskType(),validationAccountRoot(t)&&family!=null?family:t.dataFamily(),t.warehouseCode(),t.taskRole(),t.executionPurpose(), utc(t.queryFromUtc()),utc(t.queryUntilUtc()),t.status(),t.applyState(),t.attemptCount(),latest==null?null:latest.attemptNo(), resultState,t.resultPublishedRunId(),utc(t.nextRetryAtUtc()), blocked==null?List.of("RETRY"):List.of(),blocked,utc(t.progressUpdatedAtUtc()),progress(t.progressJson()), validation(root==null?t:root),latest==null?null:attemptView(latest,t,blocked==null)); } private String retryBlock(Task task,Scope scope,CollectionResponses.Protection protection,Context context) { - if(!"WAREHOUSE_SYNC".equals(task.taskRole()))return "ACCOUNT_SHARED_TASK"; if(!"BUSINESS".equals(task.executionPurpose()))return "VALIDATION_NOT_RETRYABLE"; + 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(!Set.of("DEAD","RETRY_WAIT").contains(task.status()))return "TASK_NOT_RETRYABLE"; diff --git a/lastleg-service/src/main/java/com/baiyue/lastleg/selfoperated/analytics/query/CollectionReadMapper.java b/lastleg-service/src/main/java/com/baiyue/lastleg/selfoperated/analytics/query/CollectionReadMapper.java index c1eeb40b4..916790867 100644 --- a/lastleg-service/src/main/java/com/baiyue/lastleg/selfoperated/analytics/query/CollectionReadMapper.java +++ b/lastleg-service/src/main/java/com/baiyue/lastleg/selfoperated/analytics/query/CollectionReadMapper.java @@ -28,28 +28,46 @@ public interface CollectionReadMapper { String SCOPE = "id,warehouse_code,data_family,scope_hash,business_timezone,schedule_cron,stale_after_seconds," + "initial_from_utc,initial_until_utc,discovery_from_utc,discovery_until_utc,rebuild_required,published_run_id," + "known_order_round_finished_at_utc,history_round_finished_at_utc,review_target_days,lease_owner,lease_until_utc"; - String TASK = "t.id,t.task_no,t.source_account,t.warehouse_code,s.data_family,t.task_role,t.task_type," + // Only a single frozen warehouse on an ordinary validation root establishes membership. + // Warehouse-directory roots have no such membership and cannot enter any warehouse view. + String VALIDATION_ROOT = "(t.execution_purpose='VALIDATION' AND t.task_type='VALIDATION'" + + " AND t.task_role='ACCOUNT_DISCOVERY' AND t.validation_root_task_id=t.id" + + " AND t.scope_id IS NULL AND t.warehouse_code IS NULL" + + " AND JSON_UNQUOTE(JSON_EXTRACT(t.query_scope_json,'$.action'))='VALIDATION'" + + " AND JSON_TYPE(JSON_EXTRACT(t.query_scope_json,'$.warehouses'))='ARRAY'" + + " AND JSON_LENGTH(JSON_EXTRACT(t.query_scope_json,'$.warehouses'))=1" + + " AND JSON_TYPE(JSON_EXTRACT(t.query_scope_json,'$.warehouses[0]'))='STRING')"; + String TASK_WAREHOUSE = "CASE WHEN " + VALIDATION_ROOT + + " THEN JSON_UNQUOTE(JSON_EXTRACT(t.query_scope_json,'$.warehouses[0]')) ELSE t.warehouse_code END"; + String TASK_FAMILY = "CASE WHEN " + VALIDATION_ROOT + " THEN 'OUTBOUND' ELSE s.data_family END"; + String TASK_MEMBERSHIP = "((t.task_role='WAREHOUSE_SYNC' AND s.id IS NOT NULL) OR " + VALIDATION_ROOT + ")"; + String TASK = "t.id,t.task_no,t.source_account," + TASK_WAREHOUSE + " AS warehouse_code," + TASK_FAMILY + " AS data_family,t.task_role,t.task_type," + "t.execution_purpose,t.scope_hash,t.status,t.apply_state,t.attempt_count,t.query_from_utc,t.query_until_utc," + "t.next_retry_at_utc,t.upstream_task_id,t.validation_root_task_id,t.result_published_run_id," + "t.progress_json,t.progress_updated_at_utc,t.validation_budget_json,t.created_at,t.updated_at"; String ATTEMPT = "r.id,r.task_id,t.task_no,r.attempt_no,r.status,r.last_phase,r.execution_purpose," + "r.fetch_started_at_utc,r.fetch_finished_at_utc,r.build_started_at_utc,r.finished_at_utc," + "r.published_at_utc,r.heartbeat_at_utc,r.progress_snapshot_json,r.coverage_status," - + "r.covered_from_utc,r.covered_until_utc,r.error_code,r.error_message,r.created_at,r.warehouse_code,r.scope_id,r.scope_hash"; - String TASK_FROM = " FROM ll_selfop_sync_task t JOIN ll_selfop_sync_scope s ON s.id=t.scope_id " + + "r.covered_from_utc,r.covered_until_utc,r.error_code,r.error_message,r.created_at," + + "CASE WHEN " + VALIDATION_ROOT + " THEN " + TASK_WAREHOUSE + " ELSE r.warehouse_code END AS warehouse_code,r.scope_id,r.scope_hash"; + String TASK_FROM = " FROM ll_selfop_sync_task t LEFT JOIN ll_selfop_sync_scope s ON s.id=t.scope_id " + "AND s.source_account=t.source_account AND s.warehouse_code=t.warehouse_code "; String RUN_FROM = " FROM ll_selfop_sync_run r JOIN ll_selfop_sync_task t ON t.id=r.task_id " - + "AND t.source_account=r.source_account AND t.warehouse_code=r.warehouse_code " - + "JOIN ll_selfop_sync_scope s ON s.id=r.scope_id AND s.id=t.scope_id "; - String FILTER = " WHERE t.source_account=#{f.account} AND t.warehouse_code=#{f.warehouse} " - + "AND t.task_role='WAREHOUSE_SYNC' " - + "AND s.data_family=#{f.family} " + + "AND t.source_account=r.source_account AND r.execution_purpose=t.execution_purpose AND r.run_role=t.task_role" + + " AND (t.warehouse_code=r.warehouse_code OR (t.warehouse_code IS NULL AND r.warehouse_code IS NULL)) " + + "LEFT JOIN ll_selfop_sync_scope s ON s.id=r.scope_id AND s.id=t.scope_id" + + " AND s.source_account=t.source_account AND s.warehouse_code=t.warehouse_code "; + String FILTER = " WHERE t.source_account=#{f.account} AND (" + TASK_WAREHOUSE + ")=#{f.warehouse} " + + "AND " + TASK_MEMBERSHIP + " " + + "AND ((" + TASK_FAMILY + ")=#{f.family} OR (" + VALIDATION_ROOT + + " AND #{f.family} IN ('OUTBOUND','INVENTORY'))) " + "AND t.task_type=#{f.type} " + "AND t.task_no=#{f.taskNo} "; String TASK_FILTER = FILTER + "AND t.status=#{f.status} " + "AND t.created_at >= #{f.from} " + "AND t.created_at < #{f.until} "; - String RUN_FILTER = FILTER + "AND r.status=#{f.status} " + String RUN_FILTER = FILTER + " AND (s.id IS NOT NULL OR (" + VALIDATION_ROOT + " AND r.scope_id IS NULL)) " + + "AND r.status=#{f.status} " + "AND r.created_at >= #{f.from} " + "AND r.created_at < #{f.until} "; String TASK_BOUND = " AND (t.created_at < #{b.createdAt} OR (t.created_at=#{b.createdAt} AND t.id <= #{b.id})) "; @@ -72,20 +90,24 @@ public interface CollectionReadMapper { @Select("") List runs(@Param("f") Filter filter, @Param("b") Bound bound); - @Select("SELECT " + TASK + TASK_FROM + " WHERE t.source_account=#{account} AND t.warehouse_code=#{warehouse} AND t.task_no=#{taskNo}") + @Select("SELECT " + TASK + TASK_FROM + " WHERE t.source_account=#{account} AND (" + TASK_WAREHOUSE + + ")=#{warehouse} AND " + TASK_MEMBERSHIP + " AND t.task_no=#{taskNo}") Task task(@Param("account") String account, @Param("warehouse") String warehouse, @Param("taskNo") String taskNo); @Select("SELECT api_cooldown_until_utc,api_pause_reason,api_budget_json FROM ll_selfop_outbound_discovery_scope WHERE source_account=#{account} LIMIT 1") Protection protection(@Param("account") String account); @Select("") List tasksByIds(@Param("account") String account,@Param("ids") List ids); @Select("") List latestAttempts(@Param("account") String account,@Param("ids") List ids); @Select("") List publishedRuns(@Param("account") String account,@Param("ids") List ids); @@ -94,6 +116,13 @@ public interface CollectionReadMapper { + " WHERE scope_id IN #{id}) ranked" + " ON ranked.id=t.id AND ranked.rn=1 WHERE t.source_account=#{account}") List currentTasks(@Param("account") String account,@Param("ids") List ids); + @Select("") + List latestValidationTasks(@Param("account") String account,@Param("warehouses") List warehouses); @Select("