diff --git a/lastleg-service/src/main/java/com/baiyue/lastleg/selfoperated/analytics/model/SelfopProgressProjection.java b/lastleg-service/src/main/java/com/baiyue/lastleg/selfoperated/analytics/model/SelfopProgressProjection.java index a70f05267..bd72168e0 100644 --- a/lastleg-service/src/main/java/com/baiyue/lastleg/selfoperated/analytics/model/SelfopProgressProjection.java +++ b/lastleg-service/src/main/java/com/baiyue/lastleg/selfoperated/analytics/model/SelfopProgressProjection.java @@ -14,16 +14,29 @@ public final class SelfopProgressProjection { "VALIDATION", "MERGE", "BUILD", "PUBLISH", "WAITING", "SAMPLE_DONE", "FAILED"); /** Pure projection for callers already holding the scope/task/run locks and fence. */ public static void applyCheckpoint(SelfopSyncTask task,SelfopSyncRun run,String phase,String unit,long completed,Long total,int scan) { + applyCheckpoint(task,run,phase,unit,completed,total,scan,true); + } + public static void applyCheckpoint(SelfopSyncTask task,SelfopSyncRun run,String phase,String unit,long completed,Long total,int scan,boolean realWork) { require(PHASES.contains(phase),"SELFOP_PROGRESS_PHASE_INVALID"); require(Set.of("PAGE","ORDER","INVENTORY_ROW","BUSINESS_KEY","BATCH","NONE").contains(unit),"SELFOP_PROGRESS_UNIT_INVALID"); require(completed>=0 && scan>=0 && (total==null || total>=completed),"SELFOP_PROGRESS_COUNTS_INVALID"); Map p=progress(task); - if(!Objects.equals(number(p.get("attemptNo")),run.getAttemptNo().longValue()))p=new LinkedHashMap<>(); + if(!Objects.equals(number(p.get("attemptNo")),run.getAttemptNo().longValue())){ + var prior=p;p=new LinkedHashMap<>(); + if(prior.get("localWorkReady") instanceof Boolean + &&Objects.equals(prior.get("localWorkReadyScopeHash"),task.getScopeHash()) + &&Objects.equals(prior.get("localWorkReadyUpstreamTaskId"),task.getUpstreamTaskId()==null?null:task.getUpstreamTaskId().toString())) + for(String key:List.of("localWorkReady","localWorkReadyScopeHash","localWorkReadyUpstreamTaskId"))p.put(key,prior.get(key)); + } + String priorIdentity=SelfopStageEstimate.identity(p); String countScope=scope(task); Object started=p.get("stageStartedAt"); if(!phase.equals(p.get("phase")) || started==null)started=instant(utcNow()); + p.put("taskNo",task.getTaskNo()); p.put("attemptNo",run.getAttemptNo()); p.put("phase",phase); p.put("progressScope",countScope); p.put("stageStartedAt",started); p.put("unit",unit); p.put("completed",completed); p.put("total",total); p.put("totalKnown",total!=null); p.put("scanRound",scan); p.put("countBasis",Set.of("INVENTORY","BUILD").contains(phase)?"ATTEMPT_ONLY":"TASK_CUMULATIVE"); p.put("sliceFrom",instant(task.getQueryFromUtc())); p.put("sliceUntil",instant(task.getQueryUntilUtc())); p.put("stageFinishedAt",null); + if(!priorIdentity.equals(SelfopStageEstimate.identity(p)))p.put("stageStartedAt",instant(utcNow())); + SelfopStageEstimate.checkpoint(p,utcNow().toInstant(ZoneOffset.UTC),realWork); // A zero denominator is meaningful evidence, but cannot produce a percentage. p.put("phasePercent",total==null||total==0?null:BigDecimal.valueOf(completed).multiply(BigDecimal.valueOf(100)) .divide(BigDecimal.valueOf(total),1,RoundingMode.DOWN)); @@ -42,6 +55,13 @@ public final class SelfopProgressProjection { /** Pure terminal snapshot; historical attempt evidence is immutable. */ public static void freezeRun(SelfopSyncTask task,SelfopSyncRun run) { + // A yielded inventory attempt resumes with the same attempt number. Its live + // sampling baseline must not survive a budget pause or execution interruption. + if(task.getProgressJson()!=null){ + var live=progress(task); + if(Objects.equals(number(live.get("attemptNo")),run.getAttemptNo().longValue())&&live.remove("etaSamples")!=null) + task.setProgressJson(bounded(live)); + } if(run.getProgressSnapshotJson()!=null)return; if(task.getProgressJson()==null) { var fallback=new LinkedHashMap();fallback.put("attemptNo",run.getAttemptNo()); @@ -50,16 +70,50 @@ public final class SelfopProgressProjection { fallback.put("unit","NONE");fallback.put("completed",0);fallback.put("total",null);fallback.put("totalKnown",false); fallback.put("sliceFrom",instant(task.getQueryFromUtc()));fallback.put("sliceUntil",instant(task.getQueryUntilUtc())); fallback.put("stageStartedAt",instant(run.getFetchStartedAtUtc()));fallback.put("stageFinishedAt",null);fallback.put("stages",List.of()); + fallback.put("taskNo",task.getTaskNo());fallback.put("summaryAsOf",instant(utcNow())); run.setProgressSnapshotJson(bounded(fallback)); return; } var p=progress(task); // A later attempt's projection must never be copied into an earlier historical run. require(Objects.equals(number(p.get("attemptNo")),run.getAttemptNo().longValue()),"SELFOP_PROGRESS_ATTEMPT_MISMATCH"); + p.remove("etaSamples");p.put("summaryAsOf",instant(utcNow())); run.setProgressSnapshotJson(bounded(p));run.setUpdatedAt(utcNow()); } - private static String bounded(Map p) {String value=json(p);require(value.getBytes(StandardCharsets.UTF_8).length<=8192,"SELFOP_PROGRESS_TOO_LARGE");return value;} + /** Route commits this proof together with its immutable effective-stage references. */ + public static void localWorkReady(SelfopSyncTask task,boolean ready) { + var p=progress(task);p.put("localWorkReady",ready); + p.put("localWorkReadyScopeHash",task.getScopeHash()); + p.put("localWorkReadyUpstreamTaskId",task.getUpstreamTaskId()==null?null:task.getUpstreamTaskId().toString()); + task.setProgressJson(bounded(p)); + } + + private static String bounded(Map p) { + // Reserve the terminal timestamp before accepting a live checkpoint. Optional + // ETA and derived display fields never displace business/audit checkpoints. + int target=p.containsKey("summaryAsOf")?8192:8192-64; + String value=json(p); + if(bytes(value)>target){p.remove("etaSamples");value=json(p);} + if(bytes(value)>target){p.remove("phasePercent");value=json(p);} + if(bytes(value)>target&&p.get("stages") instanceof List prior){ + var stages=new ArrayList>(); + for(Object entry:prior){var stage=map(entry);stage.remove("phasePercent");stages.add(stage);} + p.put("stages",stages);value=json(p); + // The current top-level stage clock remains available. Duplicated stage + // display clocks can be omitted while retaining every stage count/identity. + if(bytes(value)>target){ + for(var stage:stages){stage.remove("stageStartedAt");stage.remove("stageFinishedAt");} + value=json(p); + } + } + // A pre-existing legacy checkpoint may have used the whole 8 KiB with no + // optional display fields. Missing optional snapshot time is preferable to + // failing its business termination; never invent a replacement timestamp. + if(bytes(value)>8192&&p.containsKey("summaryAsOf")){p.remove("summaryAsOf");value=json(p);} + require(bytes(value)<=8192,"SELFOP_PROGRESS_TOO_LARGE");return value; + } + private static int bytes(String value){return value.getBytes(StandardCharsets.UTF_8).length;} private static String scope(SelfopSyncTask task){return "ACCOUNT_DISCOVERY".equals(task.getTaskRole())?"ACCOUNT_SHARED":"WAREHOUSE";} @SuppressWarnings("unchecked") private static Map progress(SelfopSyncTask task){return task.getProgressJson()==null?new LinkedHashMap<>():read(task.getProgressJson(),LinkedHashMap.class);} @SuppressWarnings("unchecked") private static Map map(Object value){return value instanceof Map?new LinkedHashMap<>((Map)value):new LinkedHashMap<>();} diff --git a/lastleg-service/src/main/java/com/baiyue/lastleg/selfoperated/analytics/model/SelfopStageEstimate.java b/lastleg-service/src/main/java/com/baiyue/lastleg/selfoperated/analytics/model/SelfopStageEstimate.java new file mode 100644 index 000000000..276066eb3 --- /dev/null +++ b/lastleg-service/src/main/java/com/baiyue/lastleg/selfoperated/analytics/model/SelfopStageEstimate.java @@ -0,0 +1,114 @@ +package com.baiyue.lastleg.selfoperated.analytics.model; + +import java.math.BigDecimal; +import java.math.RoundingMode; +import java.nio.charset.StandardCharsets; +import java.time.Instant; +import java.util.*; +import static com.baiyue.lastleg.selfoperated.analytics.model.SelfopSupport.*; + +/** Bounded committed-checkpoint sampling; reads never manufacture progress. */ +public final class SelfopStageEstimate { + private SelfopStageEstimate() { } + public static final int WINDOW_SECONDS=180, MIN_SECONDS=60, MIN_ADVANCES=3, + BUCKET_SECONDS=10, MAX_POINTS=20, STALE_SECONDS=60, VALID_SECONDS=10, MAX_BYTES=1024; + public record Estimate(String status,String progressIdentity,Long completed,Long total, + Instant sampleFrom,Instant sampleUntil,Long completedDelta,BigDecimal ratePerSecond, + Long remainingSeconds,Instant estimatedAt,Instant validUntil,String unavailableReason) { } + + public static String identity(Map p) { + var values=new ArrayList(); + for(String key:List.of("taskNo","attemptNo","phase","scanRound","progressScope","unit", + "countBasis","sliceFrom","sliceUntil","total"))values.add(p.get(key)); + return hash(values); + } + + /** A reuse/uncertain checkpoint resets the baseline, never contributes throughput. */ + public static void checkpoint(Map p,Instant now,boolean realWork) { + try { + Long completed=number(p.get("completed")),total=number(p.get("total")); + if(total==null||total<=0||completed==null||completed<0||completed>=total||!Boolean.TRUE.equals(p.get("totalKnown"))||"NONE".equals(p.get("unit"))) { + p.remove("etaSamples");return; + } + String id=identity(p);long second=now.getEpochSecond(); + var old=map(p.get("etaSamples"));var points=points(old.get("p")); + if(!id.equals(old.get("i"))||!realWork||points.isEmpty() + ||second=STALE_SECONDS)points.clear(); + points.removeIf(point->second-point.get(0)>WINDOW_SECONDS); + long advances=points.isEmpty()?0:points.get(points.size()-1).get(2); + long lastAdvance=numberOr(old.get("a"),second); + if(points.isEmpty()) {points.add(List.of(second,completed,0L));lastAdvance=second;} + else { + var last=points.get(points.size()-1); + if(completed>last.get(1)) {advances++;lastAdvance=second;} + var point=List.of(second,completed,advances); + if(last.get(0)/BUCKET_SECONDS==second/BUCKET_SECONDS)points.set(points.size()-1,point); + else points.add(point); + } + while(points.size()>MAX_POINTS)points.remove(0); + var sample=new LinkedHashMap();sample.put("i",id);sample.put("a",lastAdvance);sample.put("p",points); + while(json(sample).getBytes(StandardCharsets.UTF_8).length>MAX_BYTES&&points.size()>1)points.remove(0); + if(json(sample).getBytes(StandardCharsets.UTF_8).length<=MAX_BYTES)p.put("etaSamples",sample); + else p.remove("etaSamples"); + } catch(RuntimeException malformedOptionalSample) {p.remove("etaSamples");} + } + + public static Estimate estimate(Map p,Instant now,boolean healthy,boolean paused,boolean terminal) { + String id=identity(p);Long completed=number(p.get("completed")),total=number(p.get("total")); + String reason=terminal?"TERMINAL":paused?"PAUSED":!healthy?"HEALTH_UNCONFIRMED": + total==null||!Boolean.TRUE.equals(p.get("totalKnown"))?"TOTAL_UNKNOWN":completed==null||completed<0||totalsecond-point.get(0)>WINDOW_SECONDS); + if(points.isEmpty()||second-lastAdvance>=STALE_SECONDS)reason="NO_RECENT_PROGRESS"; + else { + var first=points.get(0);var last=points.get(points.size()-1); + long elapsed=second-first.get(0),advances=last.get(2)-first.get(2); + if(second map(Object value){return value instanceof Map?(Map)value:Map.of();} + private static List> points(Object value) { + var result=new ArrayList>(); + if(value instanceof List list)for(Object item:list) { + if(!(item instanceof List point)||point.size()!=3||result.size()>=MAX_POINTS)throw new IllegalArgumentException("Invalid bounded sample"); + var parsed=List.of(Objects.requireNonNull(number(point.get(0))),Objects.requireNonNull(number(point.get(1))),Objects.requireNonNull(number(point.get(2)))); + if(parsed.stream().anyMatch(n->n<0))throw new IllegalArgumentException("Negative sample"); + if(!result.isEmpty()){ + var prior=result.get(result.size()-1); + if(parsed.get(0)<=prior.get(0)||parsed.get(1) TERMINAL = Set.of("ROUTED","PUBLISHED","REUSED","YIELDED","FAILED","ABORTED","SAMPLE_DONE"); + private static final Set ACTIVE = Set.of("RUNNING","READY","QUEUED","WAITING","WAITING_SOURCE","RETRY_WAIT","YIELDED"); private static final Set PROGRESS_FIELDS = Set.of("progressScope","countBasis","attemptNo","scanRound","phase", "completed","total","totalKnown","unit","sliceFrom","sliceUntil","stageStartedAt","stageFinishedAt", "validationState","invalidCount","unknownCount","pendingReviewCount","countUnit","summaryAsOf","directoryComplete"); @@ -76,14 +78,20 @@ public class CollectionQueryService { for(String family:List.of("OUTBOUND","INVENTORY")) { 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); + List candidates=new ArrayList<>(currentTasks.stream().filter(t->warehouse.equals(t.warehouseCode())&&family.equals(t.dataFamily())).toList()); 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; + if(sampled!=null)candidates.add(sampled); + candidates.sort(Comparator.comparingInt((Task t)->"RUNNING".equals(t.status())?0:ACTIVE.contains(t.status())?1:2) + .thenComparing(Task::createdAt,Comparator.nullsLast(Comparator.reverseOrder())) + .thenComparing(Task::id,Comparator.nullsLast(Comparator.reverseOrder()))); + Task current=candidates.isEmpty()?null:candidates.get(0); + List activity=candidates.stream().filter(t->ACTIVE.contains(t.status())).limit(8) + .map(t->taskView(t,scope,protection,context,family)).toList(); 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()) ||!Objects.equals(scope.scopeHash(),published.scopeHash())))published=null; - Task shared="OUTBOUND".equals(family)&&sharedWindowRelated(active,scope)?active:null; + Task shared=current==null?null:relatedUpstream(current,context); String blocked=executionBlock(warehouse,protection); boolean enabled=blocked==null; boolean publication="BUSINESS".equals(config.getCollectionMode())&&config.isExecutionEnabled() @@ -104,10 +112,11 @@ public class CollectionQueryService { "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,family),shared==null?null:taskView(shared,null,protection,context),protection, - "INVENTORY".equals(family)?inventorySamples.get(warehouse):null)); + "INVENTORY".equals(family)?inventorySamples.get(warehouse):null,activity)); } } - return new CollectionResponses.Overview(query.sourceAccount(),config.getEnvironment(),config.getCollectionMode(),clock.instant(),rows); + return new CollectionResponses.Overview(query.sourceAccount(),config.getEnvironment(),config.getCollectionMode(),clock.instant(),rows, + active!=null&&ACTIVE.contains(active.status())&&"BUSINESS".equals(active.executionPurpose())?taskView(active,null,protection,context):null); } private Map inventorySamples(CollectionQuery query) { @@ -181,14 +190,6 @@ public class CollectionQueryService { for(Scope scope:mapper.scopes(q.sourceAccount(),q.warehouseCode()))scopes.put(scope.dataFamily(),scope); return scopes; } - private boolean sharedWindowRelated(Task active,Scope scope) { - if(active==null||scope==null||active.queryFromUtc()==null||active.queryUntilUtc()==null)return false; - return overlaps(active.queryFromUtc(),active.queryUntilUtc(),scope.initialFromUtc(),scope.initialUntilUtc()) - ||overlaps(active.queryFromUtc(),active.queryUntilUtc(),scope.discoveryFromUtc(),scope.discoveryUntilUtc()); - } - private static boolean overlaps(LocalDateTime from,LocalDateTime until,LocalDateTime otherFrom,LocalDateTime otherUntil) { - return otherFrom!=null&&otherUntil!=null&&from.isBefore(otherUntil)&&until.isAfter(otherFrom); - } private String reviewStatus(Scope scope) { if(scope==null||scope.reviewTargetDays()==null||scope.reviewTargetDays()<1)return "UNKNOWN"; if(scope.knownOrderRoundFinishedAtUtc()==null||scope.historyRoundFinishedAtUtc()==null)return "PENDING"; @@ -199,6 +200,7 @@ public class CollectionQueryService { 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()) + &&Objects.equals(task.executionPurpose(),parent.executionPurpose()) &&(!"VALIDATION".equals(parent.executionPurpose())||validationAccountRoot(parent)&&Objects.equals(task.warehouseCode(),parent.warehouseCode()))?parent:null; } private static boolean validationAccountRoot(Task task) { @@ -223,11 +225,55 @@ 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"; + Task parent=relatedUpstream(t,context); + String dependency=dependencyState(t,parent); + String waiting=blockingReason(t,dependency,protection); + CollectionResponses.Attempt attempt=latest==null?null:attemptView(latest,t,blocked==null); 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)); + validation(root==null?t:root),attempt,dependency,waiting, + "WAITING_CONFIRMED".equals(dependency)?parent.taskNo():t.taskNo(),stageEstimate(t,latest,attempt,waiting)); + } + private String dependencyState(Task task,Task parent) { + if(task.upstreamTaskId()==null)return "WAITING_SOURCE".equals(task.status())?"WAITING_UNCONFIRMED":"NONE"; + if(parent==null)return "WAITING_UNCONFIRMED"; + if("SUCCEEDED".equals(parent.status()))return "SATISFIED"; + if(Set.of("DEAD","FAILED","ABORTED","CANCELLED").contains(parent.status()))return "FAILED"; + return "WAITING_CONFIRMED"; + } + private String blockingReason(Task task,String dependency,CollectionResponses.Protection protection) { + if(!ACTIVE.contains(task.status()))return null; + String disabled=executionBlock(task.warehouseCode(),protection); + if(disabled!=null)return disabled; + if("test".equals(config.getEnvironment())&&"BUSINESS".equals(task.executionPurpose())) { + String budget=businessBudgetBlock(task);if(budget!=null)return budget; + } + if("WAITING_UNCONFIRMED".equals(dependency))return "UPSTREAM_UNCONFIRMED"; + if("WAITING_CONFIRMED".equals(dependency))return "UPSTREAM_NOT_READY"; + if("FAILED".equals(dependency))return "UPSTREAM_FAILED"; + if("RETRY_WAIT".equals(task.status()))return "RETRY_SCHEDULED"; + if("YIELDED".equals(task.status()))return "EXECUTION_YIELDED"; + if(Set.of("READY","QUEUED","WAITING").contains(task.status()))return "QUEUED"; + return null; + } + private SelfopStageEstimate.Estimate stageEstimate(Task task,Attempt run,CollectionResponses.Attempt attempt,String blocked) { + boolean terminal=!ACTIVE.contains(task.status())||run!=null&&TERMINAL.contains(run.status()); + String source=task.progressJson(); + JsonNode bounded=boundedObject(source); + Map raw=bounded==null?new LinkedHashMap<>():json.convertValue(bounded,new com.fasterxml.jackson.core.type.TypeReference>(){}); + raw.putAll(progress(source)); + raw.put("taskNo",task.taskNo()); + boolean healthy="RUNNING".equals(task.status())&&attempt!=null&&"HEALTHY".equals(attempt.executionHealth()) + &&Objects.equals(task.attemptCount(),attempt.attemptNo())&&Objects.equals(number(bounded,"attemptNo"),attempt.attemptNo()==null?null:attempt.attemptNo().longValue()); + var estimate=SelfopStageEstimate.estimate(raw,clock.instant(),healthy,blocked!=null,terminal); + if("AVAILABLE".equals(estimate.status())&&attempt.heartbeatAt()!=null) { + Instant healthUntil=attempt.heartbeatAt().plusSeconds(config.getLeaseSeconds()); + if(healthUntil.isBefore(estimate.validUntil()))return new SelfopStageEstimate.Estimate(estimate.status(),estimate.progressIdentity(),estimate.completed(),estimate.total(), + estimate.sampleFrom(),estimate.sampleUntil(),estimate.completedDelta(),estimate.ratePerSecond(),estimate.remainingSeconds(),estimate.estimatedAt(),healthUntil,null); + } + return estimate; } private String retryBlock(Task task,Scope scope,CollectionResponses.Protection protection,Context context) { if(!"BUSINESS".equals(task.executionPurpose()))return "VALIDATION_NOT_RETRYABLE"; @@ -294,9 +340,9 @@ public class CollectionQueryService { if(!Set.of("BUSINESS","VALIDATION").contains(config.getCollectionMode()))return "COLLECTION_MODE_INVALID"; 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); - if(!"Asia/Shanghai".equals(w.getBusinessTimezone())||!"Asia/Shanghai".equals(w.getSourceTimezone())||!"PCS".equals(config.getQuantityUnit()))return "TIMEZONE_OR_UNIT_NOT_CONFIRMED"; + if(warehouse!=null&&!config.getWarehouses().containsKey(warehouse))return "WAREHOUSE_NOT_CONFIGURED"; + var w=warehouse==null?null:config.getWarehouses().get(warehouse); + if(w!=null&&(!"Asia/Shanghai".equals(w.getBusinessTimezone())||!"Asia/Shanghai".equals(w.getSourceTimezone()))||!"PCS".equals(config.getQuantityUnit()))return "TIMEZONE_OR_UNIT_NOT_CONFIRMED"; if(protection.accountPauseReason()!=null&&!protection.accountPauseReason().isBlank())return "ACCOUNT_PAUSED"; if(protection.accountCooldownUntil()!=null&&clock.instant().isBefore(protection.accountCooldownUntil()))return "ACCOUNT_COOLDOWN"; if(!config.isExclusiveHandoffVerified()||!config.isPriorEgressClosed()||!config.isInflightDrained()||!config.isTransferredBudgetVerified() 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 e882c4872..6981a31e2 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 @@ -39,16 +39,14 @@ public interface CollectionReadMapper { + " 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_FAMILY = "CASE WHEN t.task_role='ACCOUNT_DISCOVERY' 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)))" + + "OCTET_LENGTH(t.progress_json) BETWEEN 0 AND 8192 AND JSON_EXTRACT(t.progress_json,'$.localWorkReady')=CAST('true' AS JSON)" + + " AND JSON_UNQUOTE(JSON_EXTRACT(t.progress_json,'$.localWorkReadyScopeHash'))=t.scope_hash" + + " AND ((t.upstream_task_id IS NULL AND JSON_TYPE(JSON_EXTRACT(t.progress_json,'$.localWorkReadyUpstreamTaskId'))='NULL')" + + " OR (t.upstream_task_id IS NOT NULL AND JSON_UNQUOTE(JSON_EXTRACT(t.progress_json,'$.localWorkReadyUpstreamTaskId'))=CAST(t.upstream_task_id AS CHAR)))))" + " 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," @@ -123,14 +121,17 @@ public interface CollectionReadMapper { + " WHERE r.source_account=#{account} AND r.status='PUBLISHED' AND r.execution_purpose='BUSINESS' AND r.run_role='WAREHOUSE_SYNC'" + " AND r.id IN #{id}") List publishedRuns(@Param("account") String account,@Param("ids") List ids); + String ACTIVE_STATUS = "('RUNNING','READY','QUEUED','WAITING','WAITING_SOURCE','RETRY_WAIT','YIELDED')"; + String ACTIVITY_ORDER = "CASE WHEN status='RUNNING' THEN 0 WHEN status IN " + ACTIVE_STATUS + " THEN 1 ELSE 2 END"; @Select("") + + " JOIN (SELECT id,ROW_NUMBER() OVER(PARTITION BY scope_id ORDER BY " + ACTIVITY_ORDER + ",created_at DESC,id DESC) rn FROM ll_selfop_sync_task" + + " WHERE source_account=#{account} AND task_role='WAREHOUSE_SYNC' AND scope_id IN #{id}) ranked" + + " ON ranked.id=t.id AND ranked.rn <= 8 WHERE t.source_account=#{account} AND (t.status IN " + ACTIVE_STATUS + " OR ranked.rn=1)" + + " ORDER BY t.scope_id,ranked.rn") List currentTasks(@Param("account") String account,@Param("ids") List ids); @Select("") @@ -141,7 +142,7 @@ public interface CollectionReadMapper { + " #{id}) ranked" + " ON ranked.id=t.id AND ranked.rn=1 WHERE t.source_account=#{account}") List lastSuccessfulTasks(@Param("account") String account,@Param("ids") List ids); - // The one account active task is separately related to warehouses by actual persisted windows. + // Account activity is independent; only upstream_task_id proves a warehouse dependency. @Select("SELECT " + TASK + " FROM ll_selfop_outbound_discovery_scope d JOIN ll_selfop_sync_task t ON t.id=d.active_task_id" + " AND t.source_account=d.source_account LEFT JOIN ll_selfop_sync_scope s ON s.id=t.scope_id" + BUSINESS_ROOT_JOIN diff --git a/lastleg-service/src/main/java/com/baiyue/lastleg/selfoperated/analytics/query/CollectionResponses.java b/lastleg-service/src/main/java/com/baiyue/lastleg/selfoperated/analytics/query/CollectionResponses.java index 3d9aec8b4..8bc333f9a 100644 --- a/lastleg-service/src/main/java/com/baiyue/lastleg/selfoperated/analytics/query/CollectionResponses.java +++ b/lastleg-service/src/main/java/com/baiyue/lastleg/selfoperated/analytics/query/CollectionResponses.java @@ -5,13 +5,14 @@ import java.time.LocalDate; import java.time.LocalDateTime; import java.util.List; import java.util.Map; +import com.baiyue.lastleg.selfoperated.analytics.model.SelfopStageEstimate; public final class CollectionResponses { private CollectionResponses() {} public record Page(int current, int size, long total, Instant listUpperCreatedAt, Long listUpperId, Instant observedAt, List records) {} public record Overview(String sourceAccount, String environmentLabel, String collectionMode, - Instant observedAt, List warehouses) {} + Instant observedAt, List warehouses, Task accountActivity) {} public record Warehouse(String warehouseCode, String warehouseName, String dataFamily, boolean executionEnabled, boolean publicationEnabled, String disabledReason, Instant nextScheduledAt, Instant observedAt, Instant initialFrom, Instant initialUntil, @@ -19,7 +20,8 @@ public final class CollectionResponses { String lastSuccessfulTaskNo, Instant lastSucceededAt, Instant lastPublishedAt, Instant inventoryObservedFrom, Instant inventoryObservedUntil, String dataFreshness, boolean rebuildRequired, String revisionReviewStatus, Task currentTask, - Task sharedUpstream, Protection protection, InventoryDailySamples inventoryDailySamples) {} + Task sharedUpstream, Protection protection, InventoryDailySamples inventoryDailySamples, + List activeTasks) {} public record InventoryDailySamples(LocalDate windowFrom, LocalDate windowUntil, int expectedDays, int sampleCount, List missingDates, LocalDate accumulationStartedOn) {} public record Task(String taskNo, String taskType, String dataFamily, String warehouseCode, @@ -27,7 +29,9 @@ public final class CollectionResponses { String taskStatus, String applyState, Integer attemptCount, Integer latestAttemptNo, String resultPublicationState, Long resultPublishedRunId, Instant nextRetryAt, List allowedActions, String actionBlockReason, Instant progressUpdatedAt, - Map progress, Validation validation, Attempt latestAttempt) {} + Map progress, Validation validation, Attempt latestAttempt, + String dependencyState, String blockingReason, String progressSourceTaskNo, + SelfopStageEstimate.Estimate stageEstimate) {} public record TaskDetail(Task task, Task sharedUpstream, Protection protection, Instant observedAt) {} public record Attempt(Long runId, String taskNo, Integer attemptNo, String status, String phase, String executionPurpose, Instant startedAt, Instant finishedAt, Instant phaseStartedAt, diff --git a/lastleg-service/src/main/java/com/baiyue/lastleg/selfoperated/analytics/request/SelfopEgressMapper.java b/lastleg-service/src/main/java/com/baiyue/lastleg/selfoperated/analytics/request/SelfopEgressMapper.java index f1b8c564e..fe1d42378 100644 --- a/lastleg-service/src/main/java/com/baiyue/lastleg/selfoperated/analytics/request/SelfopEgressMapper.java +++ b/lastleg-service/src/main/java/com/baiyue/lastleg/selfoperated/analytics/request/SelfopEgressMapper.java @@ -11,6 +11,6 @@ public interface SelfopEgressMapper { @Update("UPDATE ll_selfop_outbound_discovery_scope SET api_budget_json=#{apiBudgetJson},api_cooldown_until_utc=#{apiCooldownUntilUtc},api_pause_reason=#{apiPauseReason},updated_at=UTC_TIMESTAMP(3) WHERE id=#{id}") int saveAccount(SelfopOutboundDiscoveryScope row); @Update("UPDATE ll_selfop_sync_task SET validation_budget_json=#{validationBudgetJson},business_budget_json=#{businessBudgetJson},status=#{status},next_retry_at_utc=#{nextRetryAtUtc},updated_at=UTC_TIMESTAMP(3) WHERE id=#{id}") int saveRoot(SelfopSyncTask row); @Update("UPDATE ll_selfop_sync_task SET status='DEAD',next_retry_at_utc=NULL,updated_at=UTC_TIMESTAMP(3) WHERE business_root_task_id=#{root} AND source_account=#{account} AND execution_purpose='BUSINESS' AND status IN ('WAITING_SOURCE','READY','RUNNING','RETRY_WAIT')") int terminateBusinessTasks(@Param("root") Long root,@Param("account") String account); - @Update("UPDATE ll_selfop_sync_run SET status='FAILED',error_code=#{reason},finished_at_utc=#{now},progress_snapshot_json=(SELECT progress_json FROM ll_selfop_sync_task WHERE id=task_id) WHERE source_account=#{account} AND execution_purpose='BUSINESS' AND task_id IN (SELECT id FROM ll_selfop_sync_task WHERE business_root_task_id=#{root} AND source_account=#{account} AND execution_purpose='BUSINESS') AND status IN ('FETCHING','VALIDATED','YIELDED','READY','BUILDING')") int terminateBusinessRuns(@Param("root") Long root,@Param("account") String account,@Param("reason") String reason,@Param("now") LocalDateTime now); - @Update("UPDATE ll_selfop_sync_run SET status=#{status},error_code=#{reason},finished_at_utc=#{now},progress_snapshot_json=(SELECT progress_json FROM ll_selfop_sync_task WHERE id=task_id) WHERE task_id IN (SELECT id FROM ll_selfop_sync_task WHERE validation_root_task_id=#{root}) AND status IN ('FETCHING','VALIDATED','YIELDED')") int terminateValidationRuns(@Param("root") Long root,@Param("status") String status,@Param("reason") String reason,@Param("now") LocalDateTime now); + @Update("UPDATE ll_selfop_sync_run SET status='FAILED',error_code=#{reason},finished_at_utc=#{now},progress_snapshot_json=(SELECT JSON_SET(JSON_REMOVE(COALESCE(progress_json,JSON_OBJECT()),'$.etaSamples'),'$.summaryAsOf',CONCAT(DATE_FORMAT(#{now},'%Y-%m-%dT%H:%i:%s.%f'),'Z')) FROM ll_selfop_sync_task WHERE id=task_id AND (progress_json IS NULL OR JSON_EXTRACT(progress_json,'$.attemptNo')=attempt_no)) WHERE source_account=#{account} AND execution_purpose='BUSINESS' AND task_id IN (SELECT id FROM ll_selfop_sync_task WHERE business_root_task_id=#{root} AND source_account=#{account} AND execution_purpose='BUSINESS') AND status IN ('FETCHING','VALIDATED','YIELDED','READY','BUILDING')") int terminateBusinessRuns(@Param("root") Long root,@Param("account") String account,@Param("reason") String reason,@Param("now") LocalDateTime now); + @Update("UPDATE ll_selfop_sync_run SET status=#{status},error_code=#{reason},finished_at_utc=#{now},progress_snapshot_json=(SELECT JSON_SET(JSON_REMOVE(COALESCE(progress_json,JSON_OBJECT()),'$.etaSamples'),'$.summaryAsOf',CONCAT(DATE_FORMAT(#{now},'%Y-%m-%dT%H:%i:%s.%f'),'Z')) FROM ll_selfop_sync_task WHERE id=task_id AND (progress_json IS NULL OR JSON_EXTRACT(progress_json,'$.attemptNo')=attempt_no)) WHERE task_id IN (SELECT id FROM ll_selfop_sync_task WHERE validation_root_task_id=#{root}) AND status IN ('FETCHING','VALIDATED','YIELDED')") int terminateValidationRuns(@Param("root") Long root,@Param("status") String status,@Param("reason") String reason,@Param("now") LocalDateTime now); } diff --git a/lastleg-service/src/main/java/com/baiyue/lastleg/selfoperated/analytics/sync/SelfopDiscoveryService.java b/lastleg-service/src/main/java/com/baiyue/lastleg/selfoperated/analytics/sync/SelfopDiscoveryService.java index 54835175f..11dec60f0 100644 --- a/lastleg-service/src/main/java/com/baiyue/lastleg/selfoperated/analytics/sync/SelfopDiscoveryService.java +++ b/lastleg-service/src/main/java/com/baiyue/lastleg/selfoperated/analytics/sync/SelfopDiscoveryService.java @@ -153,10 +153,11 @@ public class SelfopDiscoveryService { full=stage(finalFetchTask,finalFetchRun,acceptedDetail,existing==null?1:existing.getFirstPageNo(),acceptedObservation,"VALID"); if(existing!=null){full.setId(existing.getId());full.setRunId(existing.getRunId());full.setCreatedAt(existing.getCreatedAt());stages.updateById(full);}else stages.insert(full); } - ref.setEffectiveStageId(full.getId());ref.setWarehouseCode(acceptedDetail.warehouseCode());saveManifest(checkpoint.task(),checkpoint.run(),manifest); + ref.setEffectiveStageId(full.getId());ref.setWarehouseCode(acceptedDetail.warehouseCode());saveManifest(checkpoint.task(),checkpoint.run(),manifest,existing==null||!"VALID".equals(existing.getValidationStatus())); if(!currentFetchTask.getId().equals(task.getId())){var refreshManifest=new SelfopRoutingManifest();refreshManifest.setDirectoryComplete(true);refreshManifest.setPages(1);refreshManifest.setTotal(1L); var refreshRef=new SelfopRoutingManifest.Entry();refreshRef.setCustomerCode(ref.getCustomerCode());refreshRef.setDeliveryNo(ref.getDeliveryNo());refreshRef.setWarehouseCode(ref.getWarehouseCode());refreshRef.setEffectiveStageId(full.getId()); refreshManifest.getObjects().put(entry.getKey(),refreshRef);currentFetchTask.setRoutingManifestJson(json(refreshManifest)); + SelfopProgressProjection.applyCheckpoint(currentFetchTask,currentFetchRun,"DETAIL","ORDER",1,1L,refreshManifest.getScan(),existing==null||!"VALID".equals(existing.getValidationStatus())); SelfopProgressProjection.freezeRun(currentFetchTask,currentFetchRun);currentFetchRun.setStatus("ROUTED");currentFetchRun.setFetchFinishedAtUtc(utcNow());currentFetchRun.setFinishedAtUtc(utcNow());currentFetchRun.setUpdatedAt(utcNow());runs.updateById(currentFetchRun); currentFetchTask.setStatus("SUCCEEDED");currentFetchTask.setUpdatedAt(utcNow());tasks.updateById(currentFetchTask);} }); @@ -204,10 +205,14 @@ public class SelfopDiscoveryService { s.setContentHash(SelfopOutboundFingerprint.order(payload));s.setPayloadJson(json(payload));s.setValidationStatus(status);s.setCreatedAt(utcNow());s.setUpdatedAt(utcNow());return s; } private void saveManifest(SelfopSyncTask task,SelfopSyncRun active,SelfopRoutingManifest manifest){ + // Directory pages are freshly fetched. Detail reference reuse cannot establish throughput. + saveManifest(task,active,manifest,!manifest.isDirectoryComplete()); + } + private void saveManifest(SelfopSyncTask task,SelfopSyncRun active,SelfopRoutingManifest manifest,boolean realWork){ task.setRoutingManifestJson(json(manifest));task.setUpdatedAt(utcNow()); SelfopProgressProjection.applyCheckpoint(task,active,manifest.isDirectoryComplete()?"DETAIL":"DIRECTORY",manifest.isDirectoryComplete()?"ORDER":"PAGE", manifest.isDirectoryComplete()?manifest.getObjects().values().stream().filter(e->e.getEffectiveStageId()!=null).count():Math.max(0,manifest.getNextPage()-1), - manifest.isDirectoryComplete()?(long)manifest.getObjects().size():manifest.getPages()==null?null:manifest.getPages().longValue(),manifest.getScan()); + manifest.isDirectoryComplete()?(long)manifest.getObjects().size():manifest.getPages()==null?null:manifest.getPages().longValue(),manifest.getScan(),realWork); runs.updateById(active);tasks.updateById(task); } private void route(SelfopSyncTask parent,SelfopSyncRun parentRun,SelfopRoutingManifest manifest,SelfopLeaseService.Lease lease,boolean detailsResolved){ @@ -238,7 +243,8 @@ public class SelfopDiscoveryService { child.setQueryScopeJson(scope.getScopeJson());child.setScopeHash(scope.getScopeHash());child.setRequestHash(hash(List.of(parent.getId(),scope.getScopeHash())));child.setStatus(ready?"READY":"WAITING_SOURCE");child.setApplyState("NOT_STARTED"); var saved=taskService.create(child); if(!Set.of("SUCCEEDED","DEAD","SAMPLE_DONE").contains(saved.getStatus())&& !Set.of("APPLYING","APPLIED","BLOCKED").contains(saved.getApplyState())){ - saved.setRoutingManifestJson(json(own));saved.setStatus(ready?"READY":"WAITING_SOURCE");saved.setUpdatedAt(utcNow());tasks.updateById(saved);} + saved.setRoutingManifestJson(json(own));saved.setStatus(ready?"READY":"WAITING_SOURCE"); + SelfopProgressProjection.localWorkReady(saved,ready);saved.setUpdatedAt(utcNow());tasks.updateById(saved);} } }); } 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 adbe9bc88..3e15db987 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 @@ -145,8 +145,8 @@ public class SelfopTaskService { var now=utcNow();var active=runs.selectOne(new QueryWrapper().eq("task_id",taskId).eq("status","FETCHING").orderByDesc("id").last("LIMIT 1")); if(task.getScopeId()!=null && "INVENTORY".equals(scopes.selectById(task.getScopeId()).getDataFamily())) { var yielded=runs.selectOne(new QueryWrapper().eq("task_id",taskId).eq("status","YIELDED").orderByDesc("id").last("LIMIT 1")); - if(yielded!=null){yielded.setStatus("FETCHING");yielded.setFinishedAtUtc(null);yielded.setFenceToken(lease.fence());yielded.setUpdatedAt(now); - runs.update(null,new UpdateWrapper().eq("id",yielded.getId()).set("status","FETCHING").set("fence_token",lease.fence()).set("finished_at_utc",null).set("updated_at",now)); + if(yielded!=null){yielded.setStatus("FETCHING");yielded.setFinishedAtUtc(null);yielded.setProgressSnapshotJson(null);yielded.setFenceToken(lease.fence());yielded.setUpdatedAt(now); + runs.update(null,new UpdateWrapper().eq("id",yielded.getId()).set("status","FETCHING").set("fence_token",lease.fence()).set("finished_at_utc",null).set("progress_snapshot_json",null).set("updated_at",now)); task.setStatus("RUNNING");task.setUpdatedAt(now);tasks.updateById(task);return yielded;} } if(active!=null){require(!Objects.equals(active.getFenceToken(),lease.fence()),"SELFOP_TASK_ALREADY_RUNNING"); diff --git a/lastleg-service/src/main/resources/openapi/self-operated-analytics-v1.yaml b/lastleg-service/src/main/resources/openapi/self-operated-analytics-v1.yaml index 33dd624b9..cae503ab9 100644 --- a/lastleg-service/src/main/resources/openapi/self-operated-analytics-v1.yaml +++ b/lastleg-service/src/main/resources/openapi/self-operated-analytics-v1.yaml @@ -750,6 +750,32 @@ components: progress: {$ref: '#/components/schemas/Progress'} validation: {$ref: '#/components/schemas/Validation'} latestAttempt: {$ref: '#/components/schemas/Attempt'} + dependencyState: + type: string + enum: [NONE, WAITING_CONFIRMED, WAITING_UNCONFIRMED, SATISFIED, FAILED] + description: 仅直接upstreamTaskId且账号及用途一致证明依赖;时间窗口相交不构成依赖。 + blockingReason: {type: string, nullable: true, description: 当前执行阻塞原因码,与恢复操作资格actionBlockReason分开} + progressSourceTaskNo: {type: string, nullable: true, description: 确认等待时指向直接上游,否则指向本任务;计数仍保留在各自Task.progress,不复制到本仓} + stageEstimate: {$ref: '#/components/schemas/StageEstimate'} + StageEstimate: + type: object + description: 本任务当前阶段估算,不能当作全任务ETA;进度与估算整体按progressIdentity和validUntil接收。计数包含复用时测速基线重建,采样仅来自已提交检查点。 + properties: + status: {type: string, enum: [AVAILABLE, UNAVAILABLE, PAUSED]} + progressIdentity: {type: string} + completed: {type: string, nullable: true} + total: {type: string, nullable: true} + sampleFrom: {type: string, format: date-time, nullable: true} + sampleUntil: {type: string, format: date-time, nullable: true} + completedDelta: {type: string, nullable: true} + ratePerSecond: {type: number, nullable: true} + remainingSeconds: {type: string, nullable: true} + estimatedAt: {type: string, format: date-time} + validUntil: {type: string, format: date-time, nullable: true} + unavailableReason: + type: string + nullable: true + enum: [TERMINAL, PAUSED, HEALTH_UNCONFIRMED, TOTAL_UNKNOWN, INVALID_PROGRESS, STAGE_COMPLETE, SAMPLING, NO_RECENT_PROGRESS, SAMPLE_EXPIRED, INVALID_SAMPLE] TaskDetail: type: object properties: @@ -795,6 +821,11 @@ components: revisionReviewStatus: {type: string, nullable: true} inventoryDailySamples: {$ref: '#/components/schemas/InventoryDailySamples'} currentTask: {$ref: '#/components/schemas/Task'} + activeTasks: + type: array + maxItems: 8 + description: 本仓本类型活动或等待任务,RUNNING优先,再按创建时间倒序;currentTask取活动首项,无活动时取最近任务。 + items: {$ref: '#/components/schemas/Task'} sharedUpstream: {$ref: '#/components/schemas/Task'} protection: {$ref: '#/components/schemas/Protection'} Overview: @@ -805,6 +836,10 @@ components: collectionMode: {type: string, nullable: true} observedAt: {type: string, format: date-time, nullable: true} warehouses: {type: array, items: {$ref: '#/components/schemas/Warehouse'}} + accountActivity: + allOf: [{$ref: '#/components/schemas/Task'}] + nullable: true + description: 来源账号明确活动指针指向的正式采集任务,独立展示,不代表任一仓库的依赖或数量。 Progress: type: object properties: diff --git a/lastleg-service/src/test/java/com/baiyue/lastleg/selfoperated/analytics/model/SelfopStageEstimateTest.java b/lastleg-service/src/test/java/com/baiyue/lastleg/selfoperated/analytics/model/SelfopStageEstimateTest.java new file mode 100644 index 000000000..bf712d1fd --- /dev/null +++ b/lastleg-service/src/test/java/com/baiyue/lastleg/selfoperated/analytics/model/SelfopStageEstimateTest.java @@ -0,0 +1,90 @@ +package com.baiyue.lastleg.selfoperated.analytics.model; + +import java.math.BigDecimal; +import java.nio.charset.StandardCharsets; +import java.time.Instant; +import java.util.*; +import org.junit.jupiter.api.Test; +import static org.junit.jupiter.api.Assertions.*; +import static com.baiyue.lastleg.selfoperated.analytics.model.SelfopSupport.*; + +class SelfopStageEstimateTest { + private final Instant start=Instant.parse("2026-10-13T00:00:00Z"); + private Map progress(){ + var p=new LinkedHashMap();p.put("taskNo","LXCOL-ETA");p.put("attemptNo",1); + p.put("phase","DETAIL");p.put("scanRound",1);p.put("progressScope","ACCOUNT_SHARED"); + p.put("unit","ORDER");p.put("countBasis","TASK_CUMULATIVE");p.put("total",1000L);p.put("totalKnown",true); + p.put("sliceFrom","2026-10-12T00:00:00Z");p.put("sliceUntil","2026-10-13T00:00:00Z");return p; + } + private void commit(Map p,long seconds,long completed,boolean real){ + p.put("completed",completed);SelfopStageEstimate.checkpoint(p,start.plusSeconds(seconds),real); + } + private Map ready(){var p=progress();for(int i=0;i<=3;i++)commit(p,i*20,1+i*20,true);return p;} + private SelfopStageEstimate.Estimate estimate(Map p,long second){return SelfopStageEstimate.estimate(p,start.plusSeconds(second),true,false,false);} + @Test void sufficientRealAdvancesEstimateCurrentStageAndIncludeIdleTime(){ + var p=ready();var e=estimate(p,60);assertEquals("AVAILABLE",e.status());assertEquals(60L,e.completedDelta()); + assertEquals(0,new BigDecimal("1").compareTo(e.ratePerSecond()));assertEquals(939L,e.remainingSeconds()); + assertEquals(start,e.sampleFrom());assertEquals(start.plusSeconds(60),e.sampleUntil());assertEquals(start.plusSeconds(70),e.validUntil()); + var idle=estimate(p,90);assertEquals("AVAILABLE",idle.status());assertEquals(new BigDecimal("0.666666"),idle.ratePerSecond()); + assertEquals(1409L,idle.remainingSeconds());assertEquals("NO_RECENT_PROGRESS",estimate(p,120).unavailableReason()); + } + @Test void shortOrInsufficientAdvanceWindowsNeverEstimate(){ + var p=progress();commit(p,0,1,true);commit(p,20,2,true);commit(p,40,3,true); + assertEquals("SAMPLING",estimate(p,59).unavailableReason());assertEquals("SAMPLING",estimate(p,60).unavailableReason()); + commit(p,60,4,true);assertEquals("AVAILABLE",estimate(p,60).status()); + } + @Test void reuseBeforeInsideAndAfterWindowRestartsBaselineInsteadOfInflatingRate(){ + var p=progress();commit(p,0,500,false);commit(p,20,501,true);commit(p,40,502,true);commit(p,60,503,true); + assertEquals(3L,estimate(p,60).completedDelta()); + commit(p,70,900,false);assertEquals("SAMPLING",estimate(p,70).unavailableReason()); + commit(p,90,901,true);commit(p,110,902,true);commit(p,130,903,true);assertEquals(3L,estimate(p,130).completedDelta()); + commit(p,400,950,false);assertEquals("SAMPLING",estimate(p,400).unavailableReason()); + } + @Test void eachIdentityDimensionInvalidatesEarlierEstimateAndCheckpointRebuildsIt(){ + for(String key:List.of("taskNo","attemptNo","phase","scanRound","progressScope","unit","countBasis","sliceFrom","sliceUntil","total")){ + var p=ready();String old=SelfopStageEstimate.identity(p); + Object value=p.get(key);p.put(key,value instanceof Number?((Number)value).longValue()+1:"changed"); + assertNotEquals(old,SelfopStageEstimate.identity(p));assertEquals("SAMPLING",estimate(p,61).unavailableReason()); + commit(p,61,62,true);assertEquals("SAMPLING",estimate(p,61).unavailableReason()); + } + } + @Test void pausedUnhealthyTerminalUnknownAndCompleteNeverReturnActiveNumbers(){ + var p=ready(); + var paused=SelfopStageEstimate.estimate(p,start.plusSeconds(60),true,true,false);assertEquals("PAUSED",paused.status());assertNull(paused.ratePerSecond()); + var terminal=SelfopStageEstimate.estimate(p,start.plusSeconds(60),true,true,true);assertEquals("UNAVAILABLE",terminal.status());assertEquals("TERMINAL",terminal.unavailableReason()); + assertEquals("HEALTH_UNCONFIRMED",SelfopStageEstimate.estimate(p,start.plusSeconds(60),false,false,false).unavailableReason()); + p.put("totalKnown",false);assertEquals("TOTAL_UNKNOWN",estimate(p,60).unavailableReason()); + p.put("totalKnown",true);p.put("completed",1000L);assertEquals("COUNT_COMPLETE",estimate(p,60).unavailableReason()); + commit(p,61,1000,true);assertFalse(p.containsKey("etaSamples")); + } + @Test void zeroAndFullCountsNeverInferStageCompletionOrPublication(){ + for(long total:List.of(0L,1000L)){ + var p=progress();p.put("total",total);p.put("completed",total);p.put("stageFinishedAt",null); + var result=estimate(p,60); + assertEquals("UNAVAILABLE",result.status());assertEquals("COUNT_COMPLETE",result.unavailableReason()); + assertNull(result.remainingSeconds());assertNull(result.ratePerSecond());assertNull(result.validUntil()); + assertNull(p.get("stageFinishedAt"));assertFalse(p.containsKey("publishedAt")); + assertEquals("DETAIL",p.get("phase")); + } + } + @Test void bucketBoundaryHighFrequencyAndLongHistoryStayBounded(){ + var p=progress();p.put("total",100000L); + for(int second=0;second<=1000;second++)commit(p,second,second,true); + var sample=(Map)p.get("etaSamples");var points=(List)sample.get("p");assertTrue(points.size()<=20); + assertTrue(json(sample).getBytes(StandardCharsets.UTF_8).length<=1024);assertEquals("AVAILABLE",estimate(p,1000).status()); + long first=((Number)((List)points.get(0)).get(0)).longValue();assertTrue(start.plusSeconds(1000).getEpochSecond()-first<=180); + var boundary=progress();commit(boundary,9,1,true);commit(boundary,10,2,true); + assertEquals(2,((List)((Map)boundary.get("etaSamples")).get("p")).size()); + } + @Test void sixtySecondGapAndCounterRegressionRestartSampling(){ + var p=ready();commit(p,120,62,true);assertEquals("SAMPLING",estimate(p,120).unavailableReason()); + commit(p,140,1,true);assertEquals("SAMPLING",estimate(p,140).unavailableReason()); + } + @Test void corruptFractionalOutOfOrderOrOversizedSamplesFailClosed(){ + var p=ready();p.put("completed",new BigDecimal("61.5"));assertEquals("INVALID_PROGRESS",estimate(p,60).unavailableReason()); + p=ready();var sample=new LinkedHashMap<>((Map)p.get("etaSamples")); + sample.put("p",List.of(List.of(start.getEpochSecond()+40,1L,0L),List.of(start.getEpochSecond()+20,2L,1L)));p.put("etaSamples",sample); + assertEquals("INVALID_SAMPLE",estimate(p,60).unavailableReason()); + p=ready();var future=estimate(p,50);assertEquals("INVALID_PROGRESS",future.unavailableReason()); + } +} diff --git a/lastleg-service/src/test/java/com/baiyue/lastleg/selfoperated/analytics/query/CollectionQueryServiceTest.java b/lastleg-service/src/test/java/com/baiyue/lastleg/selfoperated/analytics/query/CollectionQueryServiceTest.java index 78f919279..853711e3e 100644 --- a/lastleg-service/src/test/java/com/baiyue/lastleg/selfoperated/analytics/query/CollectionQueryServiceTest.java +++ b/lastleg-service/src/test/java/com/baiyue/lastleg/selfoperated/analytics/query/CollectionQueryServiceTest.java @@ -25,6 +25,78 @@ class CollectionQueryServiceTest { CollectionQuery q(Integer page,LocalDateTime upper,Long id) { return new CollectionQuery("primary","A",null,null,null,null,null,null,page,null,upper==null?null:upper.toInstant(ZoneOffset.UTC),id); } + private Task activity(long id,String status,Long upstream,boolean account,LocalDateTime created,String progress) { + return new Task(id,"LXCOL-"+id,"primary",account?null:"A","OUTBOUND",account?"ACCOUNT_DISCOVERY":"WAREHOUSE_SYNC","BACKFILL","BUSINESS", + "hash",status,"NOT_STARTED",1,now.minusDays(1),now,null,upstream,null,null,progress,now,null,null,false,created,created); + } + @Test void currentActivityWinsOverNewerHistoryAndIsBoundedToEight() { + var candidates=new ArrayList(); + candidates.add(activity(99,"SUCCEEDED",null,false,now,null)); + for(int i=0;i<10;i++)candidates.add(activity(10+i,"WAITING_SOURCE",null,false,now.minusMinutes(i+1),null)); + candidates.add(activity(1,"RUNNING",null,false,now.minusDays(1),null)); + when(mapper.currentTasks(eq("primary"),anyList())).thenReturn(candidates); + var row=service.overview(q(null,null,null)).warehouses().get(0); + assertEquals("LXCOL-1",row.currentTask().taskNo());assertEquals(8,row.activeTasks().size()); + assertEquals("LXCOL-1",row.activeTasks().get(0).taskNo()); + assertTrue(row.activeTasks().stream().noneMatch(t->"SUCCEEDED".equals(t.taskStatus()))); + } + @Test void overlappingAccountActivityDoesNotProveWarehouseDependency() { + var child=activity(7,"WAITING_SOURCE",null,false,now,null); + var account=activity(8,"RUNNING",null,true,now,null); + when(mapper.currentTasks(eq("primary"),anyList())).thenReturn(List.of(child)); + when(mapper.sharedActiveAccount("primary")).thenReturn(account); + when(mapper.task("primary","A",child.taskNo())).thenReturn(child); + var result=service.overview(q(null,null,null));var row=result.warehouses().get(0); + assertEquals(account.taskNo(),result.accountActivity().taskNo());assertNull(row.sharedUpstream()); + assertEquals("WAITING_UNCONFIRMED",row.currentTask().dependencyState()); + var detail=service.taskDetail(child.taskNo(),q(null,null,null)); + assertNull(detail.sharedUpstream());assertEquals(row.currentTask().dependencyState(),detail.task().dependencyState()); + } + @Test void directDependenciesAreConsistentInOverviewAndDetailIncludingFailureAndCompletion() { + enableBusiness("production"); + var child=activity(7,"WAITING_SOURCE",8L,false,now,null); + when(mapper.currentTasks(eq("primary"),anyList())).thenReturn(List.of(child)); + when(mapper.task("primary","A",child.taskNo())).thenReturn(child); + for(var pair:Map.of("RUNNING","WAITING_CONFIRMED","SUCCEEDED","SATISFIED","DEAD","FAILED").entrySet()) { + var parent=activity(8,pair.getKey(),null,true,now.minusHours(1),null); + when(mapper.tasksByIds(eq("primary"),anyList())).thenReturn(List.of(parent)); + var row=service.overview(q(null,null,null)).warehouses().get(0); + var detail=service.taskDetail(child.taskNo(),q(null,null,null)); + assertEquals(parent.taskNo(),row.sharedUpstream().taskNo()); + assertEquals(parent.taskNo(),detail.sharedUpstream().taskNo()); + assertEquals(pair.getValue(),row.currentTask().dependencyState()); + assertEquals(pair.getValue(),detail.task().dependencyState()); + if("RUNNING".equals(pair.getKey())) { + assertEquals(parent.taskNo(),detail.task().progressSourceTaskNo()); + assertEquals("UPSTREAM_NOT_READY",detail.task().blockingReason()); + assertEquals("TASK_NOT_RETRYABLE",detail.task().actionBlockReason()); + } + } + when(mapper.tasksByIds(eq("primary"),anyList())).thenReturn(List.of()); + assertEquals("WAITING_UNCONFIRMED",service.taskDetail(child.taskNo(),q(null,null,null)).task().dependencyState()); + } + @Test void stageEstimateUsesCommittedSamplesWithoutExposingThemAndRejectsAttemptMismatch() throws Exception { + enableBusiness("production"); + var p=new LinkedHashMap(); + p.put("taskNo","LXCOL-7");p.put("attemptNo",1);p.put("phase","DETAIL");p.put("scanRound",1); + p.put("progressScope","WAREHOUSE");p.put("unit","ORDER");p.put("countBasis","TASK_CUMULATIVE");p.put("total",100);p.put("totalKnown",true); + Instant at=now.toInstant(ZoneOffset.UTC); + for(int i=0;i<4;i++) { + p.put("completed",10+i*10); + com.baiyue.lastleg.selfoperated.analytics.model.SelfopStageEstimate.checkpoint(p,at.minusSeconds(70-i*20),true); + } + var t=activity(7,"RUNNING",null,false,now.minusMinutes(2),new ObjectMapper().writeValueAsString(p)); + when(mapper.task("primary","A",t.taskNo())).thenReturn(t); + when(mapper.latestAttempts(eq("primary"),anyList())).thenReturn(List.of(run(1,"FETCHING",null))); + var result=service.taskDetail(t.taskNo(),q(null,null,null)).task(); + assertEquals("AVAILABLE",result.stageEstimate().status()); + assertEquals(result.progress().get("completed"),result.stageEstimate().completed()); + assertEquals(result.progress().get("total"),result.stageEstimate().total()); + assertFalse(result.progress().containsKey("etaSamples")); + when(mapper.latestAttempts(eq("primary"),anyList())).thenReturn(List.of(run(2,"FETCHING",null))); + result=service.taskDetail(t.taskNo(),q(null,null,null)).task(); + assertEquals("UNAVAILABLE",result.stageEstimate().status());assertNull(result.stageEstimate().remainingSeconds()); + } @Test void controlledTestBusinessProjectsEnabledInventoryPublicationWithoutAutomaticSchedule() { config.setCollectionMode("BUSINESS");config.setEnvironment("test");config.setQuantityUnit("PCS"); config.setExclusiveHandoffVerified(true);config.setPriorEgressClosed(true);config.setInflightDrained(true);config.setTransferredBudgetVerified(true); @@ -279,7 +351,7 @@ class CollectionQueryServiceTest { "publishedRuns","currentTasks","latestValidationTasks","lastSuccessfulTasks","sharedActiveAccount","barrierWarehouses", "inventoryDailySamples","inventoryAccumulationStarts")) { String sql=configuration.getMappedStatement(CollectionReadMapper.class.getName()+"."+method).getBoundSql(parameters).getSql(); - assertFalse(sql.contains("AS routing_manifest_json"));assertFalse(sql.contains("ll_selfop_sku_daily_sales"));assertFalse(sql.contains("ll_selfop_inventory_snapshot")); + assertFalse(sql.contains("routing_manifest_json"));assertFalse(sql.contains("JSON_TABLE"));assertFalse(sql.contains("ll_selfop_sku_daily_sales"));assertFalse(sql.contains("ll_selfop_inventory_snapshot")); assertFalse(sql.contains("ll_selfop_sync_stage")); } String runSql=configuration.getMappedStatement(CollectionReadMapper.class.getName()+".runs").getBoundSql(parameters).getSql(); @@ -289,11 +361,15 @@ class CollectionQueryServiceTest { assertTrue(sql.contains("b.source_account=t.source_account")); assertTrue(sql.contains("b.execution_purpose='BUSINESS' AND b.business_root_task_id=b.id")); assertTrue(sql.contains("AS business_local_work_ready")); - assertTrue(sql.contains("'$.objects'))='OBJECT'")); - assertTrue(sql.contains("'$.objects.*'")); - assertTrue(sql.contains("COALESCE(JSON_TYPE(JSON_EXTRACT(manifest_item.entry,'$.effectiveStageId')),'NULL')")); - assertTrue(sql.contains("BETWEEN 1 AND 9223372036854775807) IS NOT TRUE")); + assertTrue(sql.contains("OCTET_LENGTH(t.progress_json) BETWEEN 0 AND 8192")); + assertTrue(sql.contains("'$.localWorkReady')=CAST('true' AS JSON)")); + assertTrue(sql.contains("'$.localWorkReadyScopeHash'))=t.scope_hash")); + assertTrue(sql.contains("'$.localWorkReadyUpstreamTaskId'))=CAST(t.upstream_task_id AS CHAR)")); } + String currentSql=configuration.getMappedStatement(CollectionReadMapper.class.getName()+".currentTasks").getBoundSql(parameters).getSql(); + assertTrue(currentSql.contains("ranked.rn <= 8")); + assertTrue(currentSql.contains("CASE WHEN status='RUNNING' THEN 0")); + assertTrue(currentSql.contains("source_account=? AND task_role='WAREHOUSE_SYNC'")); assertTrue(runSql.contains("r.created_at <"));assertTrue(runSql.contains("r.id <="));assertTrue(runSql.contains("LIMIT ? OFFSET ?")); String samplesSql=configuration.getMappedStatement(CollectionReadMapper.class.getName()+".inventoryDailySamples").getBoundSql(parameters).getSql(); assertTrue(samplesSql.contains("r.inventory_daily_sample_date >= ?")); diff --git a/lastleg-service/src/test/java/com/baiyue/lastleg/selfoperated/analytics/request/SelfopEgressServiceTest.java b/lastleg-service/src/test/java/com/baiyue/lastleg/selfoperated/analytics/request/SelfopEgressServiceTest.java index 1dd0d91da..e79ee25fa 100644 --- a/lastleg-service/src/test/java/com/baiyue/lastleg/selfoperated/analytics/request/SelfopEgressServiceTest.java +++ b/lastleg-service/src/test/java/com/baiyue/lastleg/selfoperated/analytics/request/SelfopEgressServiceTest.java @@ -225,6 +225,10 @@ class SelfopEgressServiceTest { assertThat(tasks).contains("business_root_task_id=? AND source_account=? AND execution_purpose='BUSINESS'", "next_retry_at_utc=NULL", "status IN ('WAITING_SOURCE','READY','RUNNING','RETRY_WAIT')"); assertThat(runs).contains("business_root_task_id=? AND source_account=? AND execution_purpose='BUSINESS'", "error_code=?", "status IN ('FETCHING','VALIDATED','YIELDED','READY','BUILDING')"); assertThat(tasks).doesNotContain("SUCCEEDED");assertThat(runs).doesNotContain("PUBLISHED","REUSED"); + assertThat(runs).contains("JSON_SET(JSON_REMOVE", "'$.etaSamples'", "'$.summaryAsOf'", "DATE_FORMAT(?,", "JSON_EXTRACT(progress_json,'$.attemptNo')=attempt_no"); + args.put("status","FAILED"); + String validation=configuration.getMappedStatement(prefix+"terminateValidationRuns").getBoundSql(args).getSql(); + assertThat(validation).contains("'$.etaSamples'", "'$.summaryAsOf'", "DATE_FORMAT(?,", "validation_root_task_id=?"); } @Test void committedCompletionWithLostReplyStaysPausedAcrossIntegrationRestart(){ business();var c=reserve(Operation.WAREHOUSES,1,null,null);service.authorize(new AuthorizationRequest(c,Operation.WAREHOUSES,PROVIDER)); diff --git a/lastleg-service/src/test/java/com/baiyue/lastleg/selfoperated/analytics/sync/SelfopDiscoveryServiceTest.java b/lastleg-service/src/test/java/com/baiyue/lastleg/selfoperated/analytics/sync/SelfopDiscoveryServiceTest.java index 84823e787..b971ca49e 100644 --- a/lastleg-service/src/test/java/com/baiyue/lastleg/selfoperated/analytics/sync/SelfopDiscoveryServiceTest.java +++ b/lastleg-service/src/test/java/com/baiyue/lastleg/selfoperated/analytics/sync/SelfopDiscoveryServiceTest.java @@ -167,6 +167,8 @@ class SelfopDiscoveryServiceTest { service.collect(task,run,lease); var warehouseA=children.get("route:1:A");var warehouseB=children.get("route:1:B"); assertEquals("WAITING_SOURCE",warehouseA.getStatus());assertEquals("WAITING_SOURCE",warehouseB.getStatus()); + assertEquals(false,read(warehouseA.getProgressJson(),Map.class).get("localWorkReady")); + assertEquals(false,read(warehouseB.getProgressJson(),Map.class).get("localWorkReady")); assertTrue(read(warehouseB.getRoutingManifestJson(),SelfopRoutingManifest.class).getObjects().isEmpty()); var partial=read(task.getRoutingManifestJson(),SelfopRoutingManifest.class); assertTrue(partial.isDirectoryComplete());assertEquals(1,partial.getObjects().values().stream().filter(e->e.getEffectiveStageId()!=null).count()); @@ -175,6 +177,11 @@ class SelfopDiscoveryServiceTest { account.setRequestBudget(10);service.collect(task,run,lease); assertEquals("READY",warehouseA.getStatus());assertEquals("READY",warehouseB.getStatus()); + for(var child:List.of(warehouseA,warehouseB)){ + var proof=read(child.getProgressJson(),Map.class);assertEquals(true,proof.get("localWorkReady")); + assertEquals(child.getScopeHash(),proof.get("localWorkReadyScopeHash")); + assertEquals(task.getId().toString(),proof.get("localWorkReadyUpstreamTaskId")); + } var destinationManifest=read(warehouseB.getRoutingManifestJson(),SelfopRoutingManifest.class); String movedKey=hash(List.of("demo","customer","two"));assertEquals(Set.of(movedKey),destinationManifest.getObjects().keySet()); var moved=destinationManifest.getObjects().get(movedKey);assertNotNull(moved.getEffectiveStageId());assertEquals("B",moved.getWarehouseCode()); diff --git a/lastleg-service/src/test/java/com/baiyue/lastleg/selfoperated/analytics/sync/SelfopInventoryRecoveryTest.java b/lastleg-service/src/test/java/com/baiyue/lastleg/selfoperated/analytics/sync/SelfopInventoryRecoveryTest.java index c9b0e42d6..3920dea7b 100644 --- a/lastleg-service/src/test/java/com/baiyue/lastleg/selfoperated/analytics/sync/SelfopInventoryRecoveryTest.java +++ b/lastleg-service/src/test/java/com/baiyue/lastleg/selfoperated/analytics/sync/SelfopInventoryRecoveryTest.java @@ -69,6 +69,27 @@ class SelfopInventoryRecoveryTest { assertEquals(7L, resumed.getFenceToken()); assertEquals("FETCHING", resumed.getStatus()); assertNull(resumed.getFinishedAtUtc()); verify(runs, never()).insert(any(SelfopSyncRun.class)); } + @Test void resumedInventoryReplacesYieldSnapshotWithFinalProgressWithoutChangingHistoricalAttempt(){ + original.setFenceToken(lease.fence()); + SelfopProgressProjection.applyCheckpoint(task,original,"INVENTORY","PAGE",1,4L,1); + service.yield(1L,20L,lease); + var yielded=read(original.getProgressSnapshotJson(),java.util.Map.class); + assertEquals(1,yielded.get("completed"));assertNotNull(yielded.get("summaryAsOf")); + var history=new SelfopSyncRun();history.setId(19L);history.setAttemptNo(0); + String historical="{\"attemptNo\":0,\"completed\":9,\"summaryAsOf\":\"2026-10-12T00:00:00Z\"}"; + history.setProgressSnapshotJson(historical); + when(runs.selectOne(any())).thenReturn(null,original); + var resumed=service.begin(1L,lease);assertSame(original,resumed);assertNull(resumed.getProgressSnapshotJson()); + var update=ArgumentCaptor.forClass(com.baomidou.mybatisplus.core.conditions.Wrapper.class); + verify(runs).update(isNull(),update.capture()); + assertTrue(((com.baomidou.mybatisplus.core.conditions.update.UpdateWrapper)update.getValue()).getSqlSet().contains("progress_snapshot_json")); + SelfopProgressProjection.applyCheckpoint(task,resumed,"INVENTORY","PAGE",4,4L,1); + java.time.Instant before=java.time.Instant.now();SelfopProgressProjection.freezeRun(task,resumed); + java.time.Instant after=java.time.Instant.now();var finished=read(resumed.getProgressSnapshotJson(),java.util.Map.class); + assertEquals(4,finished.get("completed"));var asOf=java.time.Instant.parse((String)finished.get("summaryAsOf")); + assertFalse(asOf.isBefore(before));assertFalse(asOf.isAfter(after));assertNotEquals(yielded,finished); + SelfopProgressProjection.freezeRun(task,history);assertEquals(historical,history.getProgressSnapshotJson()); + } @Test void genuineFailureClearsCursorAndNewAttemptStartsAtPageOneWithItsOwnStageBatch() { original.setFenceToken(lease.fence()); diff --git a/lastleg-service/src/test/java/com/baiyue/lastleg/selfoperated/analytics/sync/SelfopProgressServiceTest.java b/lastleg-service/src/test/java/com/baiyue/lastleg/selfoperated/analytics/sync/SelfopProgressServiceTest.java index fdb0608a9..2f4bbadd8 100644 --- a/lastleg-service/src/test/java/com/baiyue/lastleg/selfoperated/analytics/sync/SelfopProgressServiceTest.java +++ b/lastleg-service/src/test/java/com/baiyue/lastleg/selfoperated/analytics/sync/SelfopProgressServiceTest.java @@ -62,4 +62,57 @@ class SelfopProgressServiceTest { verify(tasks,never()).updateById(any(SelfopSyncTask.class));verify(runs,never()).updateById(any(SelfopSyncRun.class)); var order=inOrder(leases,locks);order.verify(leases).verify(lease);order.verify(locks).lockTask(1);order.verify(locks).lockRun(2); } + @Test void terminalSnapshotHasActualFreezeTimeAndResumeCannotKeepPausedEtaBaseline(){ + var t=task();var r=run();SelfopProgressProjection.applyCheckpoint(t,r,"DETAIL","ORDER",4,10L,1); + assertTrue(read(t.getProgressJson(),Map.class).containsKey("etaSamples")); + Instant before=Instant.now();SelfopProgressProjection.freezeRun(t,r);Instant after=Instant.now(); + var snapshot=read(r.getProgressSnapshotJson(),Map.class);Instant asOf=Instant.parse((String)snapshot.get("summaryAsOf")); + assertFalse(asOf.isBefore(before));assertFalse(asOf.isAfter(after));assertFalse(snapshot.containsKey("etaSamples")); + assertFalse(read(t.getProgressJson(),Map.class).containsKey("etaSamples")); + SelfopProgressProjection.applyCheckpoint(t,r,"DETAIL","ORDER",5,10L,1); + var points=(List)((Map)read(t.getProgressJson(),Map.class).get("etaSamples")).get("p");assertEquals(1,points.size()); + var fallback=run();SelfopProgressProjection.freezeRun(task(),fallback); + assertNotNull(read(fallback.getProgressSnapshotJson(),Map.class).get("summaryAsOf")); + } + @Test void etaIsFirstToBeDroppedAtEightKiBWhileCandidateAndAuditEvidenceRemain(){ + var t=task();var r=run();SelfopProgressProjection.applyCheckpoint(t,r,"DETAIL","ORDER",1,100L,1); + var p=new java.util.LinkedHashMap(read(t.getProgressJson(),Map.class));p.remove("etaSamples"); + p.put("candidateBuild",Map.of("committedBatches",7)); + p.put("audit", "x".repeat(Math.max(0,8192-json(p).getBytes(StandardCharsets.UTF_8).length-80))); + t.setProgressJson(json(p));SelfopProgressProjection.applyCheckpoint(t,r,"DETAIL","ORDER",2,100L,1); + var result=read(t.getProgressJson(),Map.class);assertFalse(result.containsKey("etaSamples")); + assertEquals(Map.of("committedBatches",7),result.get("candidateBuild"));assertEquals(p.get("audit"),result.get("audit")); + assertTrue(t.getProgressJson().getBytes(StandardCharsets.UTF_8).length<=8192); + } + @Test void localWorkProofIsBoundToTheCommittedScopeAndUpstream(){ + var t=task();t.setScopeHash("scope-1");t.setUpstreamTaskId(99L); + SelfopProgressProjection.localWorkReady(t,true);var p=read(t.getProgressJson(),Map.class); + assertEquals(true,p.get("localWorkReady"));assertEquals("scope-1",p.get("localWorkReadyScopeHash"));assertEquals("99",p.get("localWorkReadyUpstreamTaskId")); + SelfopProgressProjection.applyCheckpoint(t,run(),"MERGE","ORDER",0,4L,0); + assertEquals(true,read(t.getProgressJson(),Map.class).get("localWorkReady")); + var next=run();next.setAttemptNo(2);SelfopProgressProjection.applyCheckpoint(t,next,"MERGE","ORDER",1,4L,0); + assertEquals(true,read(t.getProgressJson(),Map.class).get("localWorkReady")); + t.setScopeHash("scope-2");next.setAttemptNo(3);SelfopProgressProjection.applyCheckpoint(t,next,"MERGE","ORDER",1,4L,0); + assertFalse(read(t.getProgressJson(),Map.class).containsKey("localWorkReady")); + SelfopProgressProjection.localWorkReady(t,false);assertEquals(false,read(t.getProgressJson(),Map.class).get("localWorkReady")); + } + @Test void exactLegacyEightKiBCheckpointFreezesWithoutLosingBusinessOrAuditCounts(){ + var t=task();var r=run();SelfopProgressProjection.applyCheckpoint(t,r,"DETAIL","ORDER",4,10L,1); + var p=new java.util.LinkedHashMap(read(t.getProgressJson(),Map.class));p.remove("etaSamples"); + p.put("candidateBuild",Map.of("committedBatches",7));p.put("audit",""); + p.put("audit","x".repeat(8192-json(p).getBytes(StandardCharsets.UTF_8).length));t.setProgressJson(json(p)); + assertEquals(8192,t.getProgressJson().getBytes(StandardCharsets.UTF_8).length); + assertDoesNotThrow(()->SelfopProgressProjection.freezeRun(t,r));var frozen=read(r.getProgressSnapshotJson(),Map.class); + assertNotNull(frozen.get("summaryAsOf"));assertEquals(4,frozen.get("completed"));assertEquals(10,frozen.get("total")); + assertEquals(p.get("candidateBuild"),frozen.get("candidateBuild"));assertEquals(p.get("audit"),frozen.get("audit")); + assertTrue(r.getProgressSnapshotJson().getBytes(StandardCharsets.UTF_8).length<=8192); + var stage=(Map)((List)frozen.get("stages")).get(0);assertEquals(4,stage.get("completed"));assertEquals(10,stage.get("total")); + } + @Test void minimalFullSizeLegacyCheckpointNeverFailsBusinessEndForOptionalTimestamp(){ + var t=task();var r=run();var p=new java.util.LinkedHashMap(); + p.put("attemptNo",1);p.put("completed",4);p.put("total",10);p.put("audit",""); + p.put("audit","x".repeat(8192-json(p).getBytes(StandardCharsets.UTF_8).length));t.setProgressJson(json(p)); + assertDoesNotThrow(()->SelfopProgressProjection.freezeRun(t,r));var frozen=read(r.getProgressSnapshotJson(),Map.class); + assertEquals(p,frozen);assertEquals(8192,r.getProgressSnapshotJson().getBytes(StandardCharsets.UTF_8).length); + } }