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

fix(selfop): expose scoped validation tasks without published scopes

Browse Source
This commit is contained in:
ouyangxq
2026-10-12 22:57:18 +08:00
parent b71380de31
commit cbc316b34d
4 changed files with 160 additions and 25 deletions
Download Patch File Download Diff File Expand all files Collapse all files
lastleg-service/src/main/java/com/baiyue/lastleg/selfoperated/analytics/query/CollectionQueryService.java
+36 -9
View File
@@ -62,8 +62,13 @@ public class CollectionQueryService {
List<Long> scopeIds=scopes.stream().filter(s->config.getWarehouses().containsKey(s.warehouseCode())).map(Scope::id).toList();
List<Task> currentTasks=scopeIds.isEmpty()?List.of():mapper.currentTasks(query.sourceAccount(),scopeIds);
List<Task> successes=scopeIds.isEmpty()?List.of():mapper.lastSuccessfulTasks(query.sourceAccount(),scopeIds);
List<String> visibleWarehouses=config.getWarehouses().keySet().stream()
.filter(w->query.warehouseCode()==null||w.equals(query.warehouseCode())).sorted().toList();
List<Task> 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<Task> all=new ArrayList<>(currentTasks);all.addAll(successes);if(active!=null)all.add(active);
List<Task> 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<String> 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&&current.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<Task> 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<CollectionResponses.Attempt> 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<String,Scope> 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";
lastleg-service/src/main/java/com/baiyue/lastleg/selfoperated/analytics/query/CollectionReadMapper.java
+40 -11
View File
@@ -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' "
+ "<if test='f.family != null'>AND s.data_family=#{f.family} </if>"
+ "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 + " "
+ "<if test='f.family != null'>AND ((" + TASK_FAMILY + ")=#{f.family} OR (" + VALIDATION_ROOT
+ " AND #{f.family} IN ('OUTBOUND','INVENTORY'))) </if>"
+ "<if test='f.type != null'>AND t.task_type=#{f.type} </if>"
+ "<if test='f.taskNo != null'>AND t.task_no=#{f.taskNo} </if>";
String TASK_FILTER = FILTER + "<if test='f.status != null'>AND t.status=#{f.status} </if>"
+ "<if test='f.from != null'>AND t.created_at &gt;= #{f.from} </if>"
+ "<if test='f.until != null'>AND t.created_at &lt; #{f.until} </if>";
String RUN_FILTER = FILTER + "<if test='f.status != null'>AND r.status=#{f.status} </if>"
String RUN_FILTER = FILTER + " AND (s.id IS NOT NULL OR (" + VALIDATION_ROOT + " AND r.scope_id IS NULL)) "
+ "<if test='f.status != null'>AND r.status=#{f.status} </if>"
+ "<if test='f.from != null'>AND r.created_at &gt;= #{f.from} </if>"
+ "<if test='f.until != null'>AND r.created_at &lt; #{f.until} </if>";
String TASK_BOUND = " AND (t.created_at &lt; #{b.createdAt} OR (t.created_at=#{b.createdAt} AND t.id &lt;= #{b.id})) ";
@@ -72,20 +90,24 @@ public interface CollectionReadMapper {
@Select("<script>SELECT " + ATTEMPT + RUN_FROM + RUN_FILTER + RUN_BOUND
+ " ORDER BY r.created_at DESC,r.id DESC LIMIT #{f.size} OFFSET #{f.offset}</script>")
List<Attempt> 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("<script>SELECT " + TASK + " 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"
+ " WHERE t.source_account=#{account} AND t.id IN <foreach collection='ids' item='id' open='(' close=')' separator=','>#{id}</foreach></script>")
List<Task> tasksByIds(@Param("account") String account,@Param("ids") List<Long> ids);
@Select("<script>SELECT " + ATTEMPT + " 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.execution_purpose=r.execution_purpose AND t.task_role=r.run_role"
+ " JOIN (SELECT id,ROW_NUMBER() OVER(PARTITION BY task_id ORDER BY attempt_no DESC,id DESC) rn FROM ll_selfop_sync_run"
+ " WHERE task_id IN <foreach collection='ids' item='id' open='(' close=')' separator=','>#{id}</foreach>) ranked"
+ " WHERE source_account=#{account} AND task_id IN <foreach collection='ids' item='id' open='(' close=')' separator=','>#{id}</foreach>) ranked"
+ " ON ranked.id=r.id AND ranked.rn=1 WHERE r.source_account=#{account}</script>")
List<Attempt> latestAttempts(@Param("account") String account,@Param("ids") List<Long> ids);
@Select("<script>SELECT " + ATTEMPT + " 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.execution_purpose=r.execution_purpose AND t.task_role=r.run_role"
+ " WHERE r.source_account=#{account} AND r.status='PUBLISHED' AND r.execution_purpose='BUSINESS' AND r.run_role='WAREHOUSE_SYNC'"
+ " AND r.id IN <foreach collection='ids' item='id' open='(' close=')' separator=','>#{id}</foreach></script>")
List<Attempt> publishedRuns(@Param("account") String account,@Param("ids") List<Long> ids);
@@ -94,6 +116,13 @@ public interface CollectionReadMapper {
+ " WHERE scope_id IN <foreach collection='ids' item='id' open='(' close=')' separator=','>#{id}</foreach>) ranked"
+ " ON ranked.id=t.id AND ranked.rn=1 WHERE t.source_account=#{account}</script>")
List<Task> currentTasks(@Param("account") String account,@Param("ids") List<Long> ids);
@Select("<script>SELECT " + TASK + TASK_FROM
+ " JOIN (SELECT t.id,ROW_NUMBER() OVER(PARTITION BY " + TASK_WAREHOUSE
+ " ORDER BY t.created_at DESC,t.id DESC) rn FROM ll_selfop_sync_task t"
+ " WHERE t.source_account=#{account} AND " + VALIDATION_ROOT + " AND (" + TASK_WAREHOUSE + ") IN"
+ " <foreach collection='warehouses' item='w' open='(' close=')' separator=','>#{w}</foreach>) ranked"
+ " ON ranked.id=t.id AND ranked.rn=1 WHERE t.source_account=#{account} AND " + VALIDATION_ROOT + "</script>")
List<Task> latestValidationTasks(@Param("account") String account,@Param("warehouses") List<String> warehouses);
@Select("<script>SELECT " + TASK + TASK_FROM
+ " JOIN (SELECT id,ROW_NUMBER() OVER(PARTITION BY scope_id ORDER BY updated_at DESC,id DESC) rn FROM ll_selfop_sync_task"
+ " WHERE status='SUCCEEDED' AND execution_purpose='BUSINESS' AND result_published_run_id IS NOT NULL AND scope_id IN"