From 7ae91eb276bd0455c4e62cde51ce1d8a69ec2a5f Mon Sep 17 00:00:00 2001 From: ouyangxq <775153912@qq.com> Date: Tue, 13 Oct 2026 03:08:36 +0800 Subject: [PATCH] fix(lastleg): persist bounded test collection session budgets --- deploy/sql/README.md | 2 +- deploy/sql/海外区域统一增量-MySQL8.sql | 4 + .../analytics/model/SelfopRequestContext.java | 4 + .../analytics/model/SelfopSyncTask.java | 7 + .../query/CollectionQueryService.java | 22 ++- .../analytics/query/CollectionReadMapper.java | 21 ++- .../analytics/query/CollectionReadRows.java | 3 +- .../request/SelfopBusinessBudget.java | 36 +++++ .../analytics/request/SelfopEgressMapper.java | 4 +- .../request/SelfopEgressService.java | 32 ++++- .../analytics/sync/SelfopCommandService.java | 6 +- .../sync/SelfopDiscoveryService.java | 67 +++++---- .../sync/SelfopInventoryCollector.java | 12 +- .../analytics/sync/SelfopTaskService.java | 133 ++++++++++++++++-- .../src/main/resources/db/lastleg.sql | 4 + .../self-operated-analytics-sync-v1.yaml | 2 + ...lfoperatedAnalyticsSchemaContractTest.java | 4 +- .../query/CollectionQueryServiceTest.java | 96 ++++++++++++- .../SelfopBusinessBudgetPersistenceTest.java | 56 ++++++++ .../request/SelfopBusinessBudgetTest.java | 33 +++++ .../request/SelfopEgressServiceTest.java | 81 ++++++++++- .../sync/SelfopCommandServiceTest.java | 10 ++ .../sync/SelfopDiscoveryServiceTest.java | 27 +++- .../sync/SelfopInventoryCollectorTest.java | 19 +++ .../sync/SelfopInventoryRecoveryTest.java | 1 + .../analytics/sync/SelfopTaskServiceTest.java | 124 ++++++++++++++++ 26 files changed, 739 insertions(+), 71 deletions(-) create mode 100644 lastleg-service/src/main/java/com/baiyue/lastleg/selfoperated/analytics/request/SelfopBusinessBudget.java create mode 100644 lastleg-service/src/test/java/com/baiyue/lastleg/selfoperated/analytics/request/SelfopBusinessBudgetPersistenceTest.java create mode 100644 lastleg-service/src/test/java/com/baiyue/lastleg/selfoperated/analytics/request/SelfopBusinessBudgetTest.java diff --git a/deploy/sql/README.md b/deploy/sql/README.md index da5775a5d..818b00712 100644 --- a/deploy/sql/README.md +++ b/deploy/sql/README.md @@ -22,7 +22,7 @@ 内部拼车需求池的当前结构在 Lastleg 源基线和海外统一空库基线同步维护,独立五项权限在 IAM 源种子和国内统一基线维护。已存在基础表的指定环境结构补齐使用 `migrations/overseas/baiyue_lastleg/V10__fleet_capacity_pool.sql` 与 `migrations/cn/baiyue_iam/V4__fleet_capacity_offer_permissions.sql`;执行前确认三张新表、六个新增字段均未存在,并单独核对目标环境。随后 `V11__align_capacity_pool_collation.sql` 将预订与商业桥的客户关联字段显式统一为 `utf8mb4_unicode_ci`,与原订单域一致,避免依赖MySQL实例默认排序规则导致JOIN失败。它们不回填、删除或重置业务数据,不能重放整个空库基线。候选文件存在不表示已经执行。 -平台自营领星统计十张 `ll_selfop_*` 表直接同步 Lastleg 源基线及海外统一空库基线;时间使用 UTC 存储及已确认北京时间业务日,数量使用 DECIMAL 保留 PCS 精度,维持独立发布版本。包含账号共享预算/冷却、验证根用途与硬预算、稳定任务号、阶段进度/心跳和旧版替代时间;库存 run 增加每日代表日期、选择规则及账号/仓库/日期唯一约束,代表库存与 profile 不适用销量副本淘汰。IAM 权限源为 `iam/src/main/resources/db/selfoperated-analytics-permissions.sql`,自营业务下经营总览、销量分析、库存分析、数据采集的菜单资源与独立读写权限同步国内统一基线,仅平台管理员默认授权。本批只交付文件与离线验证,不执行 SQL、不部署;真实采集模式默认 OFF。 +平台自营领星统计十张 `ll_selfop_*` 表直接同步 Lastleg 源基线及海外统一空库基线;时间使用 UTC 存储及已确认北京时间业务日,数量使用 DECIMAL 保留 PCS 精度,维持独立发布版本。包含账号共享预算/冷却、验证根用途与硬预算、TEST BUSINESS 根会话持久预算、稳定任务号、阶段进度/心跳和旧版替代时间;任务的 `business_root_task_id` 标记共享根,只有根保存 `business_budget_json`,生产任务不使用该测试预算。库存 run 包含每日代表日期、选择规则及账号/仓库/日期唯一约束,代表库存与 profile 不适用销量副本淘汰。IAM 权限源为 `iam/src/main/resources/db/selfoperated-analytics-permissions.sql`,自营业务下经营总览、销量分析、库存分析、数据采集的菜单资源与独立读写权限同步国内统一基线,仅平台管理员默认授权。执行和部署状态只维护于工作区测试环境主契约;真实采集模式默认 OFF。 ## 唯一入口与区域边界 diff --git a/deploy/sql/海外区域统一增量-MySQL8.sql b/deploy/sql/海外区域统一增量-MySQL8.sql index 4a505e045..9fe44e578 100644 --- a/deploy/sql/海外区域统一增量-MySQL8.sql +++ b/deploy/sql/海外区域统一增量-MySQL8.sql @@ -3990,6 +3990,8 @@ CREATE TABLE ll_selfop_sync_task ( execution_purpose VARCHAR(16) NOT NULL COMMENT 'BUSINESS或VALIDATION;由服务端冻结用途,子任务继承', validation_root_task_id BIGINT NULL COMMENT '验证会话唯一预算根任务,根引用自身,正式任务为空', validation_budget_json JSON NULL COMMENT '验证根冻结硬预算、累计消耗、库存分仓页数、截止时刻及停止原因', + business_root_task_id BIGINT NULL COMMENT '测试人工正式采集共享预算根任务;生产任务为空', + business_budget_json JSON NULL COMMENT '仅测试正式采集根保存冻结请求上限、截止时间、消耗及停止原因', source_account VARCHAR(64) NOT NULL COMMENT '受控来源账号别名,不保存密钥', task_role VARCHAR(24) NOT NULL COMMENT 'ACCOUNT_DISCOVERY 或 WAREHOUSE_SYNC', scope_id BIGINT NULL COMMENT '分仓任务必须引用分仓作用域;仅账号共享发现为空', @@ -4026,6 +4028,8 @@ CREATE TABLE ll_selfop_sync_task ( KEY idx_selfop_task_apply (apply_state,id), KEY idx_selfop_task_scope_list (scope_id,created_at,id), KEY idx_selfop_task_account_list (source_account,task_role,created_at,id), + KEY idx_selfop_task_business_root (business_root_task_id,status,id), + CONSTRAINT chk_selfop_business_budget CHECK ((business_root_task_id IS NULL AND business_budget_json IS NULL) OR (execution_purpose='BUSINESS' AND business_root_task_id IS NOT NULL AND ((business_root_task_id=id AND business_budget_json IS NOT NULL) OR (business_root_task_id<>id AND business_budget_json IS NULL)))), CONSTRAINT chk_selfop_sync_task CHECK (attempt_count >= 0 AND execution_purpose IN ('BUSINESS','VALIDATION') AND ((execution_purpose = 'BUSINESS' AND validation_root_task_id IS NULL AND validation_budget_json IS NULL) OR (execution_purpose = 'VALIDATION' AND validation_root_task_id IS NOT NULL AND apply_state = 'NOT_APPLICABLE' AND result_published_run_id IS NULL AND status <> 'SUCCEEDED')) AND ((task_role = 'ACCOUNT_DISCOVERY' AND scope_id IS NULL AND warehouse_code IS NULL AND discovery_scope_id IS NOT NULL) OR (task_role = 'WAREHOUSE_SYNC' AND scope_id IS NOT NULL AND warehouse_code IS NOT NULL))) ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_bin COMMENT='一次计划或人工请求'; diff --git a/lastleg-service/src/main/java/com/baiyue/lastleg/selfoperated/analytics/model/SelfopRequestContext.java b/lastleg-service/src/main/java/com/baiyue/lastleg/selfoperated/analytics/model/SelfopRequestContext.java index 50b42c69d..c8fb94183 100644 --- a/lastleg-service/src/main/java/com/baiyue/lastleg/selfoperated/analytics/model/SelfopRequestContext.java +++ b/lastleg-service/src/main/java/com/baiyue/lastleg/selfoperated/analytics/model/SelfopRequestContext.java @@ -5,6 +5,10 @@ public record SelfopRequestContext(String sourceAccount,String executionPurpose, String sessionId,String exclusiveGrantId,Long taskId,Long runId) { public static SelfopRequestContext forTask(SelfopSyncTask task,SelfopSyncRun run,SelfopAnalyticsProperties properties) { Long root="VALIDATION".equals(task.getExecutionPurpose())?task.getValidationRootTaskId():task.getId(); + if("BUSINESS".equals(task.getExecutionPurpose())&&"test".equals(properties.getEnvironment())) { + root=task.getBusinessRootTaskId(); + SelfopSupport.require(root!=null,"BUSINESS_SESSION_ROOT_REQUIRED"); + } return new SelfopRequestContext(task.getSourceAccount(),task.getExecutionPurpose(),root, "selfop-"+root,properties.getExclusiveGrantId(),task.getId(),run.getId()); } diff --git a/lastleg-service/src/main/java/com/baiyue/lastleg/selfoperated/analytics/model/SelfopSyncTask.java b/lastleg-service/src/main/java/com/baiyue/lastleg/selfoperated/analytics/model/SelfopSyncTask.java index 9bf50a686..bf537782c 100644 --- a/lastleg-service/src/main/java/com/baiyue/lastleg/selfoperated/analytics/model/SelfopSyncTask.java +++ b/lastleg-service/src/main/java/com/baiyue/lastleg/selfoperated/analytics/model/SelfopSyncTask.java @@ -3,6 +3,8 @@ package com.baiyue.lastleg.selfoperated.analytics.model; import com.baomidou.mybatisplus.annotation.IdType; import com.baomidou.mybatisplus.annotation.TableId; import com.baomidou.mybatisplus.annotation.TableName; +import com.baomidou.mybatisplus.annotation.TableField; +import com.baomidou.mybatisplus.annotation.FieldStrategy; import lombok.Data; import java.math.BigDecimal; import java.time.LocalDate; @@ -15,6 +17,11 @@ public class SelfopSyncTask { private String executionPurpose; private Long validationRootTaskId; private String validationBudgetJson; + @TableField(updateStrategy = FieldStrategy.NEVER) + private Long businessRootTaskId; + // Only locked explicit budget SQL may update this counter; collector entities can be stale. + @TableField(updateStrategy = FieldStrategy.NEVER) + private String businessBudgetJson; private Long resultPublishedRunId; private String progressJson; private LocalDateTime progressUpdatedAtUtc; 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 40ecaf3d6..d6f39b2ee 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 @@ -235,6 +235,9 @@ public class CollectionQueryService { 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("test".equals(config.getEnvironment())) { + String budgetBlock=businessBudgetBlock(task);if(budgetBlock!=null)return budgetBlock; + } 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"; @@ -242,6 +245,19 @@ public class CollectionQueryService { if(task.upstreamTaskId()!=null&&(parent==null||!"SUCCEEDED".equals(parent.status())))return "UPSTREAM_NOT_READY"; return null; } + private String businessBudgetBlock(Task task) { + JsonNode budget=boundedObject(task.businessRootBudgetJson()); + Long maximum=number(budget,"maxHttp"),reserved=number(budget,"http"),confirmed=number(budget,"confirmedHttp"); + Instant started=budgetInstant(budget,"startedAtUtc"),deadline=budgetInstant(budget,"deadlineAtUtc"); + if(budget==null||maximum==null||maximum<=0||maximum>Integer.MAX_VALUE||reserved==null||reserved>maximum + ||confirmed==null||confirmed>reserved||started==null||deadline==null||!deadline.isAfter(started)) + return "BUSINESS_SESSION_BUDGET_MISSING"; + JsonNode stopped=budget.get("stopReason"); + if(stopped!=null&&!stopped.isNull()) + return "BUSINESS_SESSION_STOPPED"; + if(!clock.instant().isBefore(deadline)&&!Boolean.TRUE.equals(task.businessLocalWorkReady()))return "BUSINESS_SESSION_DEADLINE"; + return null; + } private CollectionResponses.Attempt attemptView(Attempt run,Task task,boolean retryAllowed) { boolean terminal=TERMINAL.contains(run.status()); boolean same=task!=null&&Objects.equals(run.attemptNo(),task.attemptCount()); @@ -251,11 +267,13 @@ public class CollectionQueryService { String health=terminal?"TERMINATED":heartbeat==null||!clock.instant().isBefore(heartbeat.plusSeconds(config.getLeaseSeconds()))?"UNCONFIRMED":"HEALTHY"; Long elapsed=started==null?null:Math.max(0,Duration.between(started,finished==null?clock.instant():finished).getSeconds()); return new CollectionResponses.Attempt(run.id(),run.taskNo(),run.attemptNo(),run.status(),run.lastPhase(),run.executionPurpose(), - started,finished,progressInstant(progress,"stageStartedAt"),updated,heartbeat,health,elapsed,run.errorCode(),safeMessage(run.errorMessage()), + started,finished,progressInstant(progress,"stageStartedAt"),updated,heartbeat,health,elapsed,run.errorCode(),safeMessage(run.errorCode(),run.errorMessage()), retryAllowed&&same&&"BUSINESS".equals(run.executionPurpose())&&Set.of("FAILED","ABORTED","YIELDED").contains(run.status()), run.coverageStatus(),utc(run.coveredFromUtc()),utc(run.coveredUntilUtc()),utc(run.publishedAtUtc()),progress); } - private String safeMessage(String error) { + private String safeMessage(String code,String error) { + if("BUSINESS_SESSION_HTTP_LIMIT".equals(code))return "本次正式采集会话的请求预算已耗尽,根任务及关联子任务已停止,不能恢复原任务"; + if("BUSINESS_SESSION_DEADLINE".equals(code))return "本次正式采集会话已到截止时间,根任务及关联子任务已停止,不能恢复原任务"; if(error==null)return null; // Errors are already sanitized by the write side; avoid exposing arbitrary payloads here. return "采集尝试失败,请根据错误码查看受控诊断记录"; 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 916790867..e882c4872 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 @@ -41,17 +41,30 @@ public interface CollectionReadMapper { + " 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 BUSINESS_LOCAL_WORK_READY = "CASE WHEN t.task_role='WAREHOUSE_SYNC' AND s.data_family='OUTBOUND'" + + " AND (t.task_type='REBUILD' OR t.apply_state='APPLIED' OR (" + + "JSON_EXTRACT(t.routing_manifest_json,'$.directoryComplete')=CAST('true' AS JSON)" + + " AND JSON_TYPE(JSON_EXTRACT(t.routing_manifest_json,'$.objects'))='OBJECT'" + + " AND NOT EXISTS (SELECT 1 FROM JSON_TABLE(t.routing_manifest_json,'$.objects.*'" + + " COLUMNS(entry JSON PATH '$')) manifest_item" + + " WHERE COALESCE(JSON_TYPE(JSON_EXTRACT(manifest_item.entry,'$.effectiveStageId')),'NULL') NOT IN ('INTEGER')" + + " OR (CAST(JSON_UNQUOTE(JSON_EXTRACT(manifest_item.entry,'$.effectiveStageId')) AS DECIMAL(65,0)) BETWEEN 1 AND 9223372036854775807) IS NOT TRUE)))" + + " THEN TRUE ELSE FALSE END"; 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"; + + "t.progress_json,t.progress_updated_at_utc,t.validation_budget_json,b.business_budget_json AS business_root_budget_json," + + BUSINESS_LOCAL_WORK_READY + " AS business_local_work_ready,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," + "CASE WHEN " + VALIDATION_ROOT + " THEN " + TASK_WAREHOUSE + " ELSE r.warehouse_code END AS warehouse_code,r.scope_id,r.scope_hash"; + String BUSINESS_ROOT_JOIN = " LEFT JOIN ll_selfop_sync_task b ON b.id=t.business_root_task_id" + + " AND b.source_account=t.source_account AND t.execution_purpose='BUSINESS'" + + " AND b.execution_purpose='BUSINESS' AND b.business_root_task_id=b.id "; 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 "; + + "AND s.source_account=t.source_account AND s.warehouse_code=t.warehouse_code " + BUSINESS_ROOT_JOIN; 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 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)) " @@ -96,8 +109,7 @@ public interface CollectionReadMapper { @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("