fix(self-operated): remove business collection deadline limits

This commit is contained in:
2026-10-13 21:39:51 +08:00
parent 242a76221e
commit f1fb5662ca
12 changed files with 150 additions and 114 deletions
@@ -7,7 +7,7 @@ import java.net.URI;
import java.time.Duration; import java.time.Duration;
import java.time.Instant; import java.time.Instant;
/** An operator-attested exclusive window; it is not a distributed account lock. */ /** Exclusive grant identity; externalExclusiveUntil bounds VALIDATION only, not BUSINESS. */
@ConfigurationProperties(prefix = "baiyue.integration.lingxing-wms.egress") @ConfigurationProperties(prefix = "baiyue.integration.lingxing-wms.egress")
public record LingxingEgressProperties(@DefaultValue("OFF") CollectionMode collectionMode, public record LingxingEgressProperties(@DefaultValue("OFF") CollectionMode collectionMode,
@DefaultValue("test") String environmentLabel, String sourceAccount, String externalExclusiveGrantId, @DefaultValue("test") String environmentLabel, String sourceAccount, String externalExclusiveGrantId,
@@ -11,7 +11,7 @@ import java.time.Instant;
import java.util.concurrent.Semaphore; import java.util.concurrent.Semaphore;
import java.util.concurrent.TimeUnit; import java.util.concurrent.TimeUnit;
/** Local serialization supplements, never replaces, the externally attested window and durable budget. */ /** Local serialization supplements the exclusive grant and durable account protection. */
@Component @Component
public final class LingxingEgressGate { public final class LingxingEgressGate {
private static final Logger log = LoggerFactory.getLogger(LingxingEgressGate.class); private static final Logger log = LoggerFactory.getLogger(LingxingEgressGate.class);
@@ -62,7 +62,7 @@ public final class LingxingEgressGate {
|| !hash.equals(permit.requestHash()) || permit.allowedAttempts() != 1 || !permit.exclusiveConfirmed() || !hash.equals(permit.requestHash()) || permit.allowedAttempts() != 1 || !permit.exclusiveConfirmed()
|| !permit.budgetConsumed() || permit.expiresAt() == null || !clock.instant().isBefore(permit.expiresAt()) || !permit.budgetConsumed() || permit.expiresAt() == null || !clock.instant().isBefore(permit.expiresAt())
|| permit.expiresAt().isAfter(clock.instant().plusSeconds(30)) || permit.expiresAt().isAfter(clock.instant().plusSeconds(30))
|| (!continuousBusiness() && permit.expiresAt().isAfter(properties.externalExclusiveUntil()))) { || (!business() && permit.expiresAt().isAfter(properties.externalExclusiveUntil()))) {
recoverableRatePause = false; paused = true; throw failure("BUDGET_PERMIT_INVALID"); recoverableRatePause = false; paused = true; throw failure("BUDGET_PERMIT_INVALID");
} }
// A new persisted permit proves the coordinator accepted the account's cooldown/budget/root state. // A new persisted permit proves the coordinator accepted the account's cooldown/budget/root state.
@@ -108,11 +108,13 @@ public final class LingxingEgressGate {
|| !context.sourceAccount().equals(properties.sourceAccount())) throw failure("EGRESS_CONTEXT_INVALID"); || !context.sourceAccount().equals(properties.sourceAccount())) throw failure("EGRESS_CONTEXT_INVALID");
if (properties.externalExclusiveGrantId() == null || properties.externalExclusiveGrantId().isBlank() if (properties.externalExclusiveGrantId() == null || properties.externalExclusiveGrantId().isBlank()
|| !properties.externalExclusiveGrantId().equals(context.exclusiveGrantId()) || !properties.externalExclusiveGrantId().equals(context.exclusiveGrantId())
|| (!continuousBusiness() && (properties.externalExclusiveUntil() == null || !clock.instant().isBefore(properties.externalExclusiveUntil()))) || (!business() && (properties.externalExclusiveUntil() == null || !clock.instant().isBefore(properties.externalExclusiveUntil())))
|| properties.budgetAuthorityBaseUri() == null) throw failure("EXCLUSIVE_WINDOW_UNCONFIRMED"); || properties.budgetAuthorityBaseUri() == null) throw failure("EXCLUSIVE_WINDOW_UNCONFIRMED");
} }
private boolean continuousBusiness() { private boolean business() {
return properties.collectionMode() == CollectionMode.BUSINESS && properties.continuousBusinessCollectionEnabled(); // The account authority validates the original task identity, including manual roots.
// BUSINESS has no operator session cutoff; each individual permit remains finite.
return properties.collectionMode() == CollectionMode.BUSINESS;
} }
private static boolean identifier(String value, int maximum) { private static boolean identifier(String value, int maximum) {
return value != null && value.length() <= maximum && value.matches("[A-Za-z0-9][A-Za-z0-9._:-]*"); return value != null && value.length() <= maximum && value.matches("[A-Za-z0-9][A-Za-z0-9._:-]*");
@@ -97,13 +97,27 @@ class LingxingEgressGateTest {
.hasMessageContaining("ACCOUNT_PAUSED"); .hasMessageContaining("ACCOUNT_PAUSED");
assertThat(ledger.authorizations).isEqualTo(1); assertThat(ledger.authorizations).isEqualTo(1);
} }
@Test void disabledContinuousBusinessStillRequiresUnexpiredGrant() { @Test void manualBusinessWithContinuousFlagOffCrossesOldCutoffButRejectsExpiredSinglePermit() {
var gate = new LingxingEgressGate(new LingxingEgressProperties(CollectionMode.BUSINESS, "test", var gate = new LingxingEgressGate(new LingxingEgressProperties(CollectionMode.BUSINESS, "test",
"demo-account", "demo-grant", NOW, URI.create("http://127.0.0.1"), Duration.ofMillis(1200), null, false), "demo-account", "demo-grant", NOW, URI.create("http://127.0.0.1"), Duration.ofMillis(1200), null, false),
ledger, clock, clock::advanceMillis); ledger, clock, clock::advanceMillis);
assertThatThrownBy(() -> gate.execute(businessContext("disabled"), Operation.WAREHOUSES, HASH, h -> true)) var manual = new EgressContext("demo-account", CollectionMode.BUSINESS, "6", "selfop-6", "manual-after-cutoff", "demo-grant");
.hasMessageContaining("EXCLUSIVE_WINDOW_UNCONFIRMED"); assertThat(gate.execute(manual, Operation.OUTBOUND_DETAIL, HASH,
assertThat(ledger.authorizations).isZero(); h -> { h.startHttp(); return h.metadata(); }).actualHttpAttempts()).isEqualTo(1);
assertThat(ledger.completions).singleElement().satisfies(c -> assertThat(c.context()).isEqualTo(manual));
ledger.mutation = p -> new AttemptPermit(p.context(), p.operation(), p.requestHash(), 1,
clock.instant(), true, true);
assertThatThrownBy(() -> gate.execute(new EgressContext("demo-account", CollectionMode.BUSINESS, "6", "selfop-6", "manual-expired-permit", "demo-grant"),
Operation.OUTBOUND_DETAIL, HASH, h -> { h.startHttp(); return true; })).hasMessageContaining("BUDGET_PERMIT_INVALID");
assertThat(ledger.authorizations).isEqualTo(2);assertThat(ledger.completions).hasSize(1);
}
@Test void manualBusinessWithoutGrantCutoffStillRejectsPermitExpiringWhileSigning() {
var gate = new LingxingEgressGate(new LingxingEgressProperties(CollectionMode.BUSINESS, "test",
"demo-account", "demo-grant", null, URI.create("http://127.0.0.1"), Duration.ofMillis(1200), null, false),
ledger, clock, clock::advanceMillis);
assertThatThrownBy(() -> gate.execute(businessContext("manual-signing-expired"), Operation.OUTBOUND_DETAIL, HASH,
h -> { clock.advanceMillis(30000); h.startHttp(); return true; })).hasMessageContaining("ATTEMPT_NOT_SENDABLE");
assertThat(ledger.completions).singleElement().satisfies(c -> assertThat(c.actualHttpAttempts()).isZero());
} }
@Test void continuousBusinessRetains1200MillisSpacingAndOneHttpPerPermit() { @Test void continuousBusinessRetains1200MillisSpacingAndOneHttpPerPermit() {
var gate = new LingxingEgressGate(new LingxingEgressProperties(CollectionMode.BUSINESS, "test", var gate = new LingxingEgressGate(new LingxingEgressProperties(CollectionMode.BUSINESS, "test",
@@ -179,8 +193,6 @@ class LingxingEgressGateTest {
assertThatThrownBy(()->gate.execute(context("wrong-purpose"),Operation.WAREHOUSES,HASH,h->true)).hasMessageContaining("EGRESS_CONTEXT_INVALID"); assertThatThrownBy(()->gate.execute(context("wrong-purpose"),Operation.WAREHOUSES,HASH,h->true)).hasMessageContaining("EGRESS_CONTEXT_INVALID");
assertThatThrownBy(()->gate.execute(new EgressContext("other-account",CollectionMode.BUSINESS,"root","session","wrong-account","demo-grant"),Operation.WAREHOUSES,HASH,h->true)).hasMessageContaining("EGRESS_CONTEXT_INVALID"); assertThatThrownBy(()->gate.execute(new EgressContext("other-account",CollectionMode.BUSINESS,"root","session","wrong-account","demo-grant"),Operation.WAREHOUSES,HASH,h->true)).hasMessageContaining("EGRESS_CONTEXT_INVALID");
assertThatThrownBy(()->gate.execute(new EgressContext("demo-account",CollectionMode.BUSINESS,"root","session","wrong-grant","other-grant"),Operation.WAREHOUSES,HASH,h->true)).hasMessageContaining("EXCLUSIVE_WINDOW_UNCONFIRMED"); assertThatThrownBy(()->gate.execute(new EgressContext("demo-account",CollectionMode.BUSINESS,"root","session","wrong-grant","other-grant"),Operation.WAREHOUSES,HASH,h->true)).hasMessageContaining("EXCLUSIVE_WINDOW_UNCONFIRMED");
clock.advanceMillis(3600000);
assertThatThrownBy(()->gate.execute(businessContext("expired-window"),Operation.WAREHOUSES,HASH,h->true)).hasMessageContaining("EXCLUSIVE_WINDOW_UNCONFIRMED");
assertThat(ledger.authorizations).isZero();assertThat(ledger.completions).isEmpty(); assertThat(ledger.authorizations).isZero();assertThat(ledger.completions).isEmpty();
} }
@Test void testBusinessUnconfirmedCompletionStillRequiresManualReviewAfterTimePasses() { @Test void testBusinessUnconfirmedCompletionStillRequiresManualReviewAfterTimePasses() {
@@ -304,12 +304,11 @@ public class CollectionQueryService {
if(continuous&&!continuousBusiness())return "CONTINUOUS_COLLECTION_DISABLED"; if(continuous&&!continuousBusiness())return "CONTINUOUS_COLLECTION_DISABLED";
if(budget==null||maximum==null||maximum<0||maximum>Integer.MAX_VALUE||reserved==null||reserved>Integer.MAX_VALUE if(budget==null||maximum==null||maximum<0||maximum>Integer.MAX_VALUE||reserved==null||reserved>Integer.MAX_VALUE
||confirmed==null||confirmed>reserved||started==null ||confirmed==null||confirmed>reserved||started==null
||(continuous?budget.hasNonNull("deadlineAtUtc"):deadline==null||!deadline.isAfter(started))) ||(budget.hasNonNull("deadlineAtUtc")&&(continuous||deadline==null||!deadline.isAfter(started))))
return "BUSINESS_SESSION_BUDGET_MISSING"; return "BUSINESS_SESSION_BUDGET_MISSING";
JsonNode stopped=budget.get("stopReason"); JsonNode stopped=budget.get("stopReason");
if(stopped!=null&&!stopped.isNull()) if(stopped!=null&&!stopped.isNull())
return "BUSINESS_SESSION_STOPPED"; return "BUSINESS_SESSION_STOPPED";
if(!continuous&&!clock.instant().isBefore(deadline)&&!Boolean.TRUE.equals(task.businessLocalWorkReady()))return "BUSINESS_SESSION_DEADLINE";
return null; return null;
} }
private CollectionResponses.Attempt attemptView(Attempt run,Task task,boolean retryAllowed) { private CollectionResponses.Attempt attemptView(Attempt run,Task task,boolean retryAllowed) {
@@ -327,7 +326,7 @@ public class CollectionQueryService {
} }
private String safeMessage(String code,String error) { private String safeMessage(String code,String error) {
if("BUSINESS_SESSION_HTTP_LIMIT".equals(code))return "本次正式采集会话的请求预算已耗尽,根任务及关联子任务已停止,不能恢复原任务"; if("BUSINESS_SESSION_HTTP_LIMIT".equals(code))return "本次正式采集会话的请求预算已耗尽,根任务及关联子任务已停止,不能恢复原任务";
if("BUSINESS_SESSION_DEADLINE".equals(code))return "本次正式采集会话已到截止时间,根任务及关联子任务已停止,不能恢复原任务"; if("BUSINESS_SESSION_DEADLINE".equals(code))return "该任务曾因旧采集截止时间停止,需核对原检查点和账户保护状态后受控恢复";
if(error==null)return null; if(error==null)return null;
// Errors are already sanitized by the write side; avoid exposing arbitrary payloads here. // Errors are already sanitized by the write side; avoid exposing arbitrary payloads here.
return "采集尝试失败,请根据错误码查看受控诊断记录"; return "采集尝试失败,请根据错误码查看受控诊断记录";
@@ -354,7 +353,7 @@ public class CollectionQueryService {
if(protection.accountPauseReason()!=null&&!protection.accountPauseReason().isBlank())return "ACCOUNT_PAUSED"; if(protection.accountPauseReason()!=null&&!protection.accountPauseReason().isBlank())return "ACCOUNT_PAUSED";
if(protection.accountCooldownUntil()!=null&&clock.instant().isBefore(protection.accountCooldownUntil()))return "ACCOUNT_COOLDOWN"; if(protection.accountCooldownUntil()!=null&&clock.instant().isBefore(protection.accountCooldownUntil()))return "ACCOUNT_COOLDOWN";
if(!config.isExclusiveHandoffVerified()||!config.isPriorEgressClosed()||!config.isInflightDrained()||!config.isTransferredBudgetVerified() if(!config.isExclusiveHandoffVerified()||!config.isPriorEgressClosed()||!config.isInflightDrained()||!config.isTransferredBudgetVerified()
||(!continuousBusiness()&&(config.getExclusiveGrantExpiresAtUtc()==null||!clock.instant().isBefore(utc(config.getExclusiveGrantExpiresAtUtc())))))return "EXCLUSIVE_GRANT_NOT_VALID"; ||("VALIDATION".equals(config.getCollectionMode())&&(config.getExclusiveGrantExpiresAtUtc()==null||!clock.instant().isBefore(utc(config.getExclusiveGrantExpiresAtUtc())))))return "EXCLUSIVE_GRANT_NOT_VALID";
if("BUSINESS".equals(config.getCollectionMode()))return null; if("BUSINESS".equals(config.getCollectionMode()))return null;
if(config.getDailyRequestBudget()<=0)return "DAILY_BUDGET_NOT_CONFIGURED"; if(config.getDailyRequestBudget()<=0)return "DAILY_BUDGET_NOT_CONFIGURED";
if(protection.reservedHttpRequests()!=null&&clock.instant().atZone(BUSINESS_ZONE).toLocalDate().toString().equals(protection.budgetDate()) if(protection.reservedHttpRequests()!=null&&clock.instant().atZone(BUSINESS_ZONE).toLocalDate().toString().equals(protection.budgetDate())
@@ -3,21 +3,21 @@ package com.baiyue.lastleg.selfoperated.analytics.request;
import com.baiyue.lastleg.selfoperated.analytics.model.SelfopSupport; import com.baiyue.lastleg.selfoperated.analytics.model.SelfopSupport;
import java.time.LocalDateTime; import java.time.LocalDateTime;
/** BUSINESS session accounting and time ceiling, shared by the root and descendants. */ /** BUSINESS session accounting and explicit stops, shared by the root and descendants. */
public final class SelfopBusinessBudget { public final class SelfopBusinessBudget {
/** Recorded field only; BUSINESS has no HTTP-count ceiling. New sessions write zero. */ /** Recorded field only; BUSINESS has no HTTP-count ceiling. New sessions write zero. */
public int maxHttp; public int maxHttp;
public int http, confirmedHttp; public int http, confirmedHttp;
public boolean continuous; public boolean continuous;
/** Existing deadlines remain recorded evidence, never an execution limit. */
public LocalDateTime startedAtUtc, deadlineAtUtc; public LocalDateTime startedAtUtc, deadlineAtUtc;
public String stopReason; public String stopReason;
public static String freeze(LocalDateTime now, int seconds) { public static String freeze(LocalDateTime now) {
if (now == null || seconds < 1) if (now == null)
throw new IllegalArgumentException("BUSINESS_SESSION_BUDGET_INVALID"); throw new IllegalArgumentException("BUSINESS_SESSION_BUDGET_INVALID");
var budget = new SelfopBusinessBudget(); var budget = new SelfopBusinessBudget();
budget.startedAtUtc = now; budget.startedAtUtc = now;
budget.deadlineAtUtc = now.plusSeconds(seconds);
return SelfopSupport.json(budget); return SelfopSupport.json(budget);
} }
@@ -29,7 +29,7 @@ public final class SelfopBusinessBudget {
public void validate() { public void validate() {
if (http < 0 || confirmedHttp < 0 || confirmedHttp > http if (http < 0 || confirmedHttp < 0 || confirmedHttp > http
|| startedAtUtc == null || (continuous ? deadlineAtUtc != null : deadlineAtUtc == null || !deadlineAtUtc.isAfter(startedAtUtc)) || startedAtUtc == null || deadlineAtUtc != null && (continuous || !deadlineAtUtc.isAfter(startedAtUtc))
|| stopReason != null && !stopReason.matches("[A-Z0-9_]{1,64}")) || stopReason != null && !stopReason.matches("[A-Z0-9_]{1,64}"))
throw new EgressRejectedException("BUSINESS_SESSION_BUDGET_INVALID", false, null); throw new EgressRejectedException("BUSINESS_SESSION_BUDGET_INVALID", false, null);
} }
@@ -37,7 +37,6 @@ public final class SelfopBusinessBudget {
public String limit(LocalDateTime now) { public String limit(LocalDateTime now) {
validate(); validate();
if (stopReason != null) return stopReason; if (stopReason != null) return stopReason;
if (!continuous && !now.isBefore(deadlineAtUtc)) return "BUSINESS_SESSION_DEADLINE";
return null; return null;
} }
} }
@@ -42,9 +42,8 @@ public class SelfopEgressService {
if("VALIDATION".equals(context.executionPurpose())&&budget.reservedHttp>=budget.dailyLimit)throw new EgressRejectedException("ACCOUNT_DAILY_BUDGET",false,now.plusHours(8).toLocalDate().plusDays(1).atStartOfDay().minusHours(8)); if("VALIDATION".equals(context.executionPurpose())&&budget.reservedHttp>=budget.dailyLimit)throw new EgressRejectedException("ACCOUNT_DAILY_BUDGET",false,now.plusHours(8).toLocalDate().plusDays(1).atStartOfDay().minusHours(8));
var outbound=new EgressContext(context.sourceAccount(),context.executionPurpose(),String.valueOf(context.rootTaskId()),context.sessionId(),UUID.randomUUID().toString(),context.exclusiveGrantId()); var outbound=new EgressContext(context.sourceAccount(),context.executionPurpose(),String.valueOf(context.rootTaskId()),context.sessionId(),UUID.randomUUID().toString(),context.exclusiveGrantId());
var pending=new SelfopApiBudget.Pending();pending.context=outbound;pending.operation=operation;pending.internalRequestHash=requestHash;pending.taskId=context.taskId();pending.runId=context.runId();pending.expiresAtUtc=now.plusSeconds(Math.max(1,Math.min(120,config.getEgressLeaseSeconds()))); var pending=new SelfopApiBudget.Pending();pending.context=outbound;pending.operation=operation;pending.internalRequestHash=requestHash;pending.taskId=context.taskId();pending.runId=context.runId();pending.expiresAtUtc=now.plusSeconds(Math.max(1,Math.min(120,config.getEgressLeaseSeconds())));
if(!continuousBusiness(context.executionPurpose())&&pending.expiresAtUtc.isAfter(config.getExclusiveGrantExpiresAtUtc()))pending.expiresAtUtc=config.getExclusiveGrantExpiresAtUtc(); if("VALIDATION".equals(context.executionPurpose())&&pending.expiresAtUtc.isAfter(config.getExclusiveGrantExpiresAtUtc()))pending.expiresAtUtc=config.getExclusiveGrantExpiresAtUtc();
if(validation!=null&&pending.expiresAtUtc.isAfter(validation.deadlineAtUtc))pending.expiresAtUtc=validation.deadlineAtUtc; if(validation!=null&&pending.expiresAtUtc.isAfter(validation.deadlineAtUtc))pending.expiresAtUtc=validation.deadlineAtUtc;
if(business!=null&&business.deadlineAtUtc!=null&&pending.expiresAtUtc.isAfter(business.deadlineAtUtc))pending.expiresAtUtc=business.deadlineAtUtc;
budget.pending=pending;budget.reservedHttp++;budget.lastRequestAtUtc=now; budget.pending=pending;budget.reservedHttp++;budget.lastRequestAtUtc=now;
if(validation!=null){validation.consume(operation,warehouseCode,detailKey);root.setValidationBudgetJson(SelfopSupport.json(validation));mapper.saveRoot(root);} if(validation!=null){validation.consume(operation,warehouseCode,detailKey);root.setValidationBudgetJson(SelfopSupport.json(validation));mapper.saveRoot(root);}
if(business!=null){business.http++;saveBusiness(root,business);} if(business!=null){business.http++;saveBusiness(root,business);}
@@ -58,10 +57,9 @@ public class SelfopEgressService {
var task=mapper.lockTask(pending.taskId);var run=mapper.lockRun(pending.runId); var task=mapper.lockTask(pending.taskId);var run=mapper.lockRun(pending.runId);
if(hasBusinessSession(root)){ if(hasBusinessSession(root)){
var b=business(root);verifyBusinessTask(task,root); var b=business(root);verifyBusinessTask(task,root);
// Authorization preserves the root time/stop fence independently of request accounting. // Explicit root stops remain independent of request accounting and short permit expiry.
String limit=b.limit(now); String limit=b.limit(now);
if(limit!=null){stopBusiness(root,b,limit,now);throw reject(limit);} if(limit!=null){stopBusiness(root,b,limit,now);throw reject(limit);}
if(b.deadlineAtUtc!=null&&pending.expiresAtUtc.isAfter(b.deadlineAtUtc))pending.expiresAtUtc=b.deadlineAtUtc;
} }
verify(task,run,root,account,now); verify(task,run,root,account,now);
if(pending.consumed||pending.completed||!now.isBefore(pending.expiresAtUtc))throw reject("EGRESS_PERMIT_REPLAY_OR_EXPIRED"); if(pending.consumed||pending.completed||!now.isBefore(pending.expiresAtUtc))throw reject("EGRESS_PERMIT_REPLAY_OR_EXPIRED");
@@ -127,7 +125,7 @@ public class SelfopEgressService {
if("OFF".equals(config.getCollectionMode())||!Objects.equals(config.getCollectionMode(),purpose)||!("VALIDATION".equals(purpose)||"BUSINESS".equals(purpose)))throw reject("EGRESS_MODE_REJECTED"); if("OFF".equals(config.getCollectionMode())||!Objects.equals(config.getCollectionMode(),purpose)||!("VALIDATION".equals(purpose)||"BUSINESS".equals(purpose)))throw reject("EGRESS_MODE_REJECTED");
if("BUSINESS".equals(purpose)&&!("production".equals(config.getEnvironment())||"test".equals(config.getEnvironment())))throw reject("BUSINESS_ENVIRONMENT_REJECTED"); if("BUSINESS".equals(purpose)&&!("production".equals(config.getEnvironment())||"test".equals(config.getEnvironment())))throw reject("BUSINESS_ENVIRONMENT_REJECTED");
if(!Objects.equals(config.getSourceAccount(),account)||account==null||account.isBlank())throw reject("SOURCE_ACCOUNT_MISMATCH"); if(!Objects.equals(config.getSourceAccount(),account)||account==null||account.isBlank())throw reject("SOURCE_ACCOUNT_MISMATCH");
if(grant==null||grant.isBlank()||!Objects.equals(grant,config.getExclusiveGrantId())||(!continuousBusiness(purpose)&&(config.getExclusiveGrantExpiresAtUtc()==null||!now.isBefore(config.getExclusiveGrantExpiresAtUtc())))||!config.isExclusiveHandoffVerified()||!config.isPriorEgressClosed()||!config.isInflightDrained()||!config.isTransferredBudgetVerified())throw reject("EXCLUSIVE_HANDOFF_UNCONFIRMED"); if(grant==null||grant.isBlank()||!Objects.equals(grant,config.getExclusiveGrantId())||("VALIDATION".equals(purpose)&&(config.getExclusiveGrantExpiresAtUtc()==null||!now.isBefore(config.getExclusiveGrantExpiresAtUtc())))||!config.isExclusiveHandoffVerified()||!config.isPriorEgressClosed()||!config.isInflightDrained()||!config.isTransferredBudgetVerified())throw reject("EXCLUSIVE_HANDOFF_UNCONFIRMED");
if(("VALIDATION".equals(purpose)&&config.getDailyRequestBudget()<1)||config.getBusinessMinRequestIntervalMs()<1||config.getRequestBudget()<1||config.getTimeBudgetSeconds()<1||config.getMaxRateLimitFailures()<1||config.getCooldownSeconds()<1)throw reject("ACCOUNT_BUDGET_UNCONFIGURED"); if(("VALIDATION".equals(purpose)&&config.getDailyRequestBudget()<1)||config.getBusinessMinRequestIntervalMs()<1||config.getRequestBudget()<1||config.getTimeBudgetSeconds()<1||config.getMaxRateLimitFailures()<1||config.getCooldownSeconds()<1)throw reject("ACCOUNT_BUDGET_UNCONFIGURED");
} }
private SelfopOutboundDiscoveryScope account(String id){var a=mapper.lockAccount(id);if(a==null)throw reject("EGRESS_ACCOUNT_MISSING");return a;} private SelfopOutboundDiscoveryScope account(String id){var a=mapper.lockAccount(id);if(a==null)throw reject("EGRESS_ACCOUNT_MISSING");return a;}
@@ -93,7 +93,7 @@ public class SelfopTaskService {
public SelfopSyncTask create(SelfopSyncTask task) { public SelfopSyncTask create(SelfopSyncTask task) {
identity(task.getRequestKey(),128); var existing=tasks.selectOne(new QueryWrapper<SelfopSyncTask>() identity(task.getRequestKey(),128); var existing=tasks.selectOne(new QueryWrapper<SelfopSyncTask>()
.eq("source_account",task.getSourceAccount()).eq("request_key",task.getRequestKey())); .eq("source_account",task.getSourceAccount()).eq("request_key",task.getRequestKey()));
if(existing!=null){require(existing.getRequestHash().equals(task.getRequestHash()),"SELFOP_IDEMPOTENCY_CONFLICT");businessGuard(existing,!localBusinessWork(existing));return existing;} if(existing!=null){require(existing.getRequestHash().equals(task.getRequestHash()),"SELFOP_IDEMPOTENCY_CONFLICT");businessGuard(existing);return existing;}
if(task.getExecutionPurpose()==null)task.setExecutionPurpose(configuredPurpose(config)); if(task.getExecutionPurpose()==null)task.setExecutionPurpose(configuredPurpose(config));
requireExecution(config,task.getExecutionPurpose()); requireExecution(config,task.getExecutionPurpose());
var now=utcNow();task.setId(IdWorker.getId());task.setAttemptCount(0);task.setCreatedAt(now);task.setUpdatedAt(now); var now=utcNow();task.setId(IdWorker.getId());task.setAttemptCount(0);task.setCreatedAt(now);task.setUpdatedAt(now);
@@ -121,13 +121,13 @@ public class SelfopTaskService {
}else if(task.getRequestKey().startsWith("manual:")&&task.getUpstreamTaskId()==null) { }else if(task.getRequestKey().startsWith("manual:")&&task.getUpstreamTaskId()==null) {
require(controlledBusinessTask(task,tasks::selectById,continuousBusiness(config)),"SELFOP_TEST_MANUAL_TASK_REQUIRED"); require(controlledBusinessTask(task,tasks::selectById,continuousBusiness(config)),"SELFOP_TEST_MANUAL_TASK_REQUIRED");
task.setBusinessRootTaskId(task.getId()); task.setBusinessRootTaskId(task.getId());
task.setBusinessBudgetJson(SelfopBusinessBudget.freeze(now,config.getTimeBudgetSeconds())); task.setBusinessBudgetJson(SelfopBusinessBudget.freeze(now));
}else { }else {
var parent=businessParent(task); var parent=businessParent(task);
require(parent!=null&&Objects.equals(parent.getSourceAccount(),task.getSourceAccount())&&controlledBusinessTask(parent,tasks::selectById,continuousBusiness(config)),"SELFOP_TEST_MANUAL_TASK_REQUIRED"); require(parent!=null&&Objects.equals(parent.getSourceAccount(),task.getSourceAccount())&&controlledBusinessTask(parent,tasks::selectById,continuousBusiness(config)),"SELFOP_TEST_MANUAL_TASK_REQUIRED");
if("MANUAL_RETRY".equals(task.getTaskType()))require("SUCCEEDED".equals(task.getStatus())&&task.getRequestKey().startsWith("retry:")&&task.getUpstreamTaskId()!=null,"SELFOP_TEST_MANUAL_TASK_REQUIRED"); if("MANUAL_RETRY".equals(task.getTaskType()))require("SUCCEEDED".equals(task.getStatus())&&task.getRequestKey().startsWith("retry:")&&task.getUpstreamTaskId()!=null,"SELFOP_TEST_MANUAL_TASK_REQUIRED");
require("MANUAL_RETRY".equals(task.getTaskType())||controlledBusinessTask(task,tasks::selectById,continuousBusiness(config)),"SELFOP_TEST_MANUAL_TASK_REQUIRED"); require("MANUAL_RETRY".equals(task.getTaskType())||controlledBusinessTask(task,tasks::selectById,continuousBusiness(config)),"SELFOP_TEST_MANUAL_TASK_REQUIRED");
businessGuard(parent,!"WAREHOUSE_SYNC".equals(task.getTaskRole()));task.setBusinessRootTaskId(parent.getBusinessRootTaskId()); businessGuard(parent);task.setBusinessRootTaskId(parent.getBusinessRootTaskId());
} }
} }
task.setStatus(task.getStatus()==null?"READY":task.getStatus()); task.setStatus(task.getStatus()==null?"READY":task.getStatus());
@@ -138,13 +138,13 @@ public class SelfopTaskService {
} }
try{tasks.insert(task);}catch(DuplicateKeyException ignored){existing=tasks.selectOne(new QueryWrapper<SelfopSyncTask>() try{tasks.insert(task);}catch(DuplicateKeyException ignored){existing=tasks.selectOne(new QueryWrapper<SelfopSyncTask>()
.eq("source_account",task.getSourceAccount()).eq("request_key",task.getRequestKey()).last("FOR UPDATE")); .eq("source_account",task.getSourceAccount()).eq("request_key",task.getRequestKey()).last("FOR UPDATE"));
require(existing!=null&&existing.getRequestHash().equals(task.getRequestHash()),"SELFOP_IDEMPOTENCY_CONFLICT");businessGuard(existing,!localBusinessWork(existing));return existing;} require(existing!=null&&existing.getRequestHash().equals(task.getRequestHash()),"SELFOP_IDEMPOTENCY_CONFLICT");businessGuard(existing);return existing;}
return task; return task;
} }
@Transactional(noRollbackFor=BusinessSessionStoppedException.class) @Transactional(noRollbackFor=BusinessSessionStoppedException.class)
public SelfopSyncRun begin(long taskId,SelfopLeaseService.Lease lease) { public SelfopSyncRun begin(long taskId,SelfopLeaseService.Lease lease) {
lockBusinessAccount(taskId);leases.verify(lease); lockBusinessAccount(taskId);leases.verify(lease);
if("test".equals(config.getEnvironment())||continuousBusiness(config)){var preview=tasks.selectById(taskId);businessGuard(preview,!localBusinessWork(preview));} if("test".equals(config.getEnvironment())||continuousBusiness(config)){var preview=tasks.selectById(taskId);businessGuard(preview);}
var task=locks.lockTask(taskId);require(task!=null&&!"SUCCEEDED".equals(task.getStatus())&&!"DEAD".equals(task.getStatus()),"SELFOP_TASK_NOT_EXECUTABLE"); var task=locks.lockTask(taskId);require(task!=null&&!"SUCCEEDED".equals(task.getStatus())&&!"DEAD".equals(task.getStatus()),"SELFOP_TASK_NOT_EXECUTABLE");
checkTaskLease(task,lease); checkTaskLease(task,lease);
requireExecution(config,task.getExecutionPurpose()); requireExecution(config,task.getExecutionPurpose());
@@ -186,14 +186,14 @@ public class SelfopTaskService {
@Transactional(noRollbackFor=BusinessSessionStoppedException.class) @Transactional(noRollbackFor=BusinessSessionStoppedException.class)
public void yield(long taskId,long runId,SelfopLeaseService.Lease lease) { public void yield(long taskId,long runId,SelfopLeaseService.Lease lease) {
lockBusinessAccount(taskId);leases.verify(lease); lockBusinessAccount(taskId);leases.verify(lease);
guardBeforeTaskLock(taskId,true); guardBeforeTaskLock(taskId);
var now=utcNow();var task=locks.lockTask(taskId);var run=locks.lockRun(runId);checkRunLease(task,run,lease);requireBusiness(task);requireBusiness(run);SelfopProgressProjection.freezeRun(task,run);run.setStatus("YIELDED");run.setFinishedAtUtc(now);run.setUpdatedAt(now);runs.updateById(run); var now=utcNow();var task=locks.lockTask(taskId);var run=locks.lockRun(runId);checkRunLease(task,run,lease);requireBusiness(task);requireBusiness(run);SelfopProgressProjection.freezeRun(task,run);run.setStatus("YIELDED");run.setFinishedAtUtc(now);run.setUpdatedAt(now);runs.updateById(run);
task.setStatus("RETRY_WAIT");task.setNextRetryAtUtc(now.plusSeconds(1));task.setUpdatedAt(now);tasks.updateById(task); task.setStatus("RETRY_WAIT");task.setNextRetryAtUtc(now.plusSeconds(1));task.setUpdatedAt(now);tasks.updateById(task);
} }
@Transactional(noRollbackFor=BusinessSessionStoppedException.class) @Transactional(noRollbackFor=BusinessSessionStoppedException.class)
public void fail(long taskId,long runId,String code,boolean retryable,SelfopLeaseService.Lease lease) { public void fail(long taskId,long runId,String code,boolean retryable,SelfopLeaseService.Lease lease) {
lockBusinessAccount(taskId);leases.verify(lease); lockBusinessAccount(taskId);leases.verify(lease);
try{guardBeforeTaskLock(taskId,true);}catch(BusinessSessionStoppedException stopped){return;} try{guardBeforeTaskLock(taskId);}catch(BusinessSessionStoppedException stopped){return;}
var now=utcNow();var task=locks.lockTask(taskId);var run=locks.lockRun(runId);checkRunLease(task,run,lease);requireBusiness(task);requireBusiness(run);SelfopProgressProjection.freezeRun(task,run);run.setLastPhase(run.getStatus());run.setStatus("FAILED");run.setErrorCode(code); var now=utcNow();var task=locks.lockTask(taskId);var run=locks.lockRun(runId);checkRunLease(task,run,lease);requireBusiness(task);requireBusiness(run);SelfopProgressProjection.freezeRun(task,run);run.setLastPhase(run.getStatus());run.setStatus("FAILED");run.setErrorCode(code);
run.setErrorMessage("自营同步失败,需核对运行记录");run.setFinishedAtUtc(now);run.setUpdatedAt(now);runs.updateById(run); run.setErrorMessage("自营同步失败,需核对运行记录");run.setFinishedAtUtc(now);run.setUpdatedAt(now);runs.updateById(run);
// A yielded execution is not a failed request; finite failures are counted separately. // A yielded execution is not a failed request; finite failures are counted separately.
@@ -218,14 +218,14 @@ public class SelfopTaskService {
lockBusinessAccount(id); lockBusinessAccount(id);
var commandQuery=new QueryWrapper<SelfopSyncTask>().eq("source_account",preview.getSourceAccount()).eq("request_key","retry:"+requestKey); var commandQuery=new QueryWrapper<SelfopSyncTask>().eq("source_account",preview.getSourceAccount()).eq("request_key","retry:"+requestKey);
var prior=tasks.selectOne(commandQuery); var prior=tasks.selectOne(commandQuery);
if(prior!=null){require(prior.getRequestHash().equals(hash(List.of(id,requestKey,reason))),"SELFOP_IDEMPOTENCY_CONFLICT");businessGuard(preview,!localBusinessWork(preview));return preview;} if(prior!=null){require(prior.getRequestHash().equals(hash(List.of(id,requestKey,reason))),"SELFOP_IDEMPOTENCY_CONFLICT");businessGuard(preview);return preview;}
// Match executor lock ordering: account/warehouse scope first, task second. // Match executor lock ordering: account/warehouse scope first, task second.
String activeOwner;LocalDateTime activeUntil; String activeOwner;LocalDateTime activeUntil;
if("ACCOUNT_DISCOVERY".equals(preview.getTaskRole())){var scope=locks.lockDiscovery(preview.getDiscoveryScopeId());require(scope!=null,"SELFOP_SCOPE_MISSING");activeOwner=scope.getLeaseOwner();activeUntil=scope.getLeaseUntilUtc();} if("ACCOUNT_DISCOVERY".equals(preview.getTaskRole())){var scope=locks.lockDiscovery(preview.getDiscoveryScopeId());require(scope!=null,"SELFOP_SCOPE_MISSING");activeOwner=scope.getLeaseOwner();activeUntil=scope.getLeaseUntilUtc();}
else{var scope=locks.lockScope(preview.getScopeId());require(scope!=null,"SELFOP_SCOPE_MISSING");activeOwner=scope.getLeaseOwner();activeUntil=scope.getLeaseUntilUtc();} else{var scope=locks.lockScope(preview.getScopeId());require(scope!=null,"SELFOP_SCOPE_MISSING");activeOwner=scope.getLeaseOwner();activeUntil=scope.getLeaseUntilUtc();}
prior=tasks.selectOne(new QueryWrapper<SelfopSyncTask>().eq("source_account",preview.getSourceAccount()).eq("request_key","retry:"+requestKey).last("FOR UPDATE")); prior=tasks.selectOne(new QueryWrapper<SelfopSyncTask>().eq("source_account",preview.getSourceAccount()).eq("request_key","retry:"+requestKey).last("FOR UPDATE"));
if(prior!=null){require(prior.getRequestHash().equals(hash(List.of(id,requestKey,reason))),"SELFOP_IDEMPOTENCY_CONFLICT");return tasks.selectById(id);} if(prior!=null){require(prior.getRequestHash().equals(hash(List.of(id,requestKey,reason))),"SELFOP_IDEMPOTENCY_CONFLICT");return tasks.selectById(id);}
businessGuard(preview,!localBusinessWork(preview)); businessGuard(preview);
require(SelfopLeaseService.available(activeOwner,activeUntil,utcNow()),"SELFOP_RETRY_LEASE_BUSY"); require(SelfopLeaseService.available(activeOwner,activeUntil,utcNow()),"SELFOP_RETRY_LEASE_BUSY");
var original=locks.lockTask(id);require(original!=null,"SELFOP_TASK_MISSING"); var original=locks.lockTask(id);require(original!=null,"SELFOP_TASK_MISSING");
requireBusiness(original);requireExecution(config,original.getExecutionPurpose()); requireBusiness(original);requireExecution(config,original.getExecutionPurpose());
@@ -245,7 +245,7 @@ public class SelfopTaskService {
public List<SelfopSyncTask> due(int limit){return continuousBusiness(config)?locks.dueContinuous(utcNow(),limit,config.getSourceAccount()):locks.due(utcNow(),limit,config.getCollectionMode());} public List<SelfopSyncTask> due(int limit){return continuousBusiness(config)?locks.dueContinuous(utcNow(),limit,config.getSourceAccount()):locks.due(utcNow(),limit,config.getCollectionMode());}
@Transactional(noRollbackFor=BusinessSessionStoppedException.class) @Transactional(noRollbackFor=BusinessSessionStoppedException.class)
public void candidateFinished(long taskId,long runId,SelfopLeaseService.Lease lease) { public void candidateFinished(long taskId,long runId,SelfopLeaseService.Lease lease) {
lockBusinessAccount(taskId);leases.verify(lease);guardBeforeTaskLock(taskId,false);var task=locks.lockTask(taskId);var run=locks.lockRun(runId); lockBusinessAccount(taskId);leases.verify(lease);guardBeforeTaskLock(taskId);var task=locks.lockTask(taskId);var run=locks.lockRun(runId);
checkRunLease(task,run,lease); checkRunLease(task,run,lease);
require("READY".equals(run.getStatus()),"SELFOP_CANDIDATE_NOT_READY"); require("READY".equals(run.getStatus()),"SELFOP_CANDIDATE_NOT_READY");
run.setFinishedAtUtc(utcNow());run.setUpdatedAt(utcNow());runs.updateById(run); run.setFinishedAtUtc(utcNow());run.setUpdatedAt(utcNow());runs.updateById(run);
@@ -254,7 +254,7 @@ public class SelfopTaskService {
} }
@Transactional(noRollbackFor=BusinessSessionStoppedException.class) @Transactional(noRollbackFor=BusinessSessionStoppedException.class)
public void waitForBudget(long taskId,long runId,LocalDateTime until,SelfopLeaseService.Lease lease) { public void waitForBudget(long taskId,long runId,LocalDateTime until,SelfopLeaseService.Lease lease) {
lockBusinessAccount(taskId);leases.verify(lease);if("test".equals(config.getEnvironment())||continuousBusiness(config)){var preview=tasks.selectById(taskId);businessGuard(preview,!localBusinessWork(preview));}var task=locks.lockTask(taskId);var run=locks.lockRun(runId);checkRunLease(task,run,lease);requireBusiness(task);requireBusiness(run); lockBusinessAccount(taskId);leases.verify(lease);if("test".equals(config.getEnvironment())||continuousBusiness(config)){var preview=tasks.selectById(taskId);businessGuard(preview);}var task=locks.lockTask(taskId);var run=locks.lockRun(runId);checkRunLease(task,run,lease);requireBusiness(task);requireBusiness(run);
SelfopProgressProjection.freezeRun(task,run);run.setLastPhase(run.getStatus());run.setStatus("YIELDED");run.setFinishedAtUtc(utcNow());run.setUpdatedAt(utcNow());runs.updateById(run); SelfopProgressProjection.freezeRun(task,run);run.setLastPhase(run.getStatus());run.setStatus("YIELDED");run.setFinishedAtUtc(utcNow());run.setUpdatedAt(utcNow());runs.updateById(run);
task.setStatus("RETRY_WAIT");task.setNextRetryAtUtc(until);task.setUpdatedAt(utcNow());tasks.updateById(task); task.setStatus("RETRY_WAIT");task.setNextRetryAtUtc(until);task.setUpdatedAt(utcNow());tasks.updateById(task);
} }
@@ -264,17 +264,17 @@ public class SelfopTaskService {
public CollectionCheckpoint checkpoint(SelfopSyncTask observedTask,SelfopSyncRun observedRun,SelfopLeaseService.Lease lease) { public CollectionCheckpoint checkpoint(SelfopSyncTask observedTask,SelfopSyncRun observedRun,SelfopLeaseService.Lease lease) {
long taskId=observedTask.getId();lockBusinessAccount(taskId);leases.verify(lease); long taskId=observedTask.getId();lockBusinessAccount(taskId);leases.verify(lease);
if("test".equals(config.getEnvironment())||continuousBusiness(config)) { if("test".equals(config.getEnvironment())||continuousBusiness(config)) {
var preview=tasks.selectById(taskId);require(preview!=null,"SELFOP_TASK_MISSING");businessGuard(preview,false,false); var preview=tasks.selectById(taskId);require(preview!=null,"SELFOP_TASK_MISSING");businessGuard(preview,false);
} }
var task=locks.lockTask(taskId);var run=locks.lockRun(observedRun.getId());checkRunLease(task,run,lease); var task=locks.lockTask(taskId);var run=locks.lockRun(observedRun.getId());checkRunLease(task,run,lease);
requireBusiness(task);requireBusiness(run); requireBusiness(task);requireBusiness(run);
require("RUNNING".equals(task.getStatus())&&"FETCHING".equals(run.getStatus()),"SELFOP_COLLECTION_TERMINATED"); require("RUNNING".equals(task.getStatus())&&"FETCHING".equals(run.getStatus()),"SELFOP_COLLECTION_TERMINATED");
return new CollectionCheckpoint(task,run); return new CollectionCheckpoint(task,run);
} }
/** Only TEST manual collection has a session ceiling; production retains its resumable scheduler. */ /** Explicit session stops remain binding after removal of artificial BUSINESS deadlines. */
public static final class BusinessSessionStoppedException extends com.baiyue.common.core.exception.BizException { public static final class BusinessSessionStoppedException extends com.baiyue.common.core.exception.BizException {
private final String reason; private final String reason;
BusinessSessionStoppedException(String reason){super("本次测试采集会话已结束或预算无效,请核对记录后新建有界任务");this.reason=reason;} BusinessSessionStoppedException(String reason){super("本次采集会话已停止或计数无效,请核对停止原因与原检查点");this.reason=reason;}
public String reason(){return reason;} public String reason(){return reason;}
} }
private boolean testBusiness(SelfopSyncTask task){return task!=null&&("test".equals(config.getEnvironment())||continuousBusiness(config))&&"BUSINESS".equals(task.getExecutionPurpose());} private boolean testBusiness(SelfopSyncTask task){return task!=null&&("test".equals(config.getEnvironment())||continuousBusiness(config))&&"BUSINESS".equals(task.getExecutionPurpose());}
@@ -290,27 +290,12 @@ public class SelfopTaskService {
var account=accounts.selectOne(new QueryWrapper<SelfopOutboundDiscoveryScope>().eq("source_account",task.getSourceAccount())); var account=accounts.selectOne(new QueryWrapper<SelfopOutboundDiscoveryScope>().eq("source_account",task.getSourceAccount()));
require(account!=null,"SELFOP_SCOPE_MISSING");locks.lockDiscovery(account.getId()); require(account!=null,"SELFOP_SCOPE_MISSING");locks.lockDiscovery(account.getId());
} }
private void guardBeforeTaskLock(long id,boolean checkDeadline){ private void guardBeforeTaskLock(long id){
if(!"test".equals(config.getEnvironment())&&!continuousBusiness(config))return; if(!"test".equals(config.getEnvironment())&&!continuousBusiness(config))return;
var task=tasks.selectById(id);businessGuard(task,checkDeadline); var task=tasks.selectById(id);businessGuard(task);
} }
private boolean localBusinessWork(SelfopSyncTask task){ private void businessGuard(SelfopSyncTask task){businessGuard(task,true);}
if(!testBusiness(task)||!"WAREHOUSE_SYNC".equals(task.getTaskRole()))return false; private void businessGuard(SelfopSyncTask task,boolean terminate){
var scope=scopes.selectById(task.getScopeId());
if(scope==null||!"OUTBOUND".equals(scope.getDataFamily()))return false;
if("REBUILD".equals(task.getTaskType())||"APPLIED".equals(task.getApplyState()))return true;
if(task.getRoutingManifestJson()==null)return false;
Map<?,?> manifest;
try{manifest=read(task.getRoutingManifestJson(),Map.class);}catch(IllegalArgumentException invalid){return false;}
if(manifest==null||!Boolean.TRUE.equals(manifest.get("directoryComplete"))||!(manifest.get("objects") instanceof Map<?,?> objects))return false;
return objects.values().stream().allMatch(value->{
if(!(value instanceof Map<?,?> entry))return false;
Object id=entry.get("effectiveStageId");
return (id instanceof Long||id instanceof Integer)&&((Number)id).longValue()>0;
});
}
private void businessGuard(SelfopSyncTask task,boolean checkDeadline){businessGuard(task,checkDeadline,true);}
private void businessGuard(SelfopSyncTask task,boolean checkDeadline,boolean terminate){
if(!testBusiness(task))return; if(!testBusiness(task))return;
require(controlledBusinessTask(task,tasks::selectById,continuousBusiness(config)),"SELFOP_TEST_MANUAL_TASK_REQUIRED"); require(controlledBusinessTask(task,tasks::selectById,continuousBusiness(config)),"SELFOP_TEST_MANUAL_TASK_REQUIRED");
require(task.getBusinessRootTaskId()!=null,"BUSINESS_SESSION_ROOT_REQUIRED"); require(task.getBusinessRootTaskId()!=null,"BUSINESS_SESSION_ROOT_REQUIRED");
@@ -327,7 +312,6 @@ public class SelfopTaskService {
var budget=read(root.getBusinessBudgetJson(),SelfopBusinessBudget.class);budget.validate(); var budget=read(root.getBusinessBudgetJson(),SelfopBusinessBudget.class);budget.validate();
require(!budget.continuous||(continuousBusiness(config)&&continuousBusinessRoot(root)),"SELFOP_CONTINUOUS_COLLECTION_DISABLED"); require(!budget.continuous||(continuousBusiness(config)&&continuousBusinessRoot(root)),"SELFOP_CONTINUOUS_COLLECTION_DISABLED");
String reason=budget.stopReason; String reason=budget.stopReason;
if(reason==null&&checkDeadline&&!budget.continuous&&!utcNow().isBefore(budget.deadlineAtUtc))reason="BUSINESS_SESSION_DEADLINE";
if(reason==null&&"DEAD".equals(root.getStatus()))reason="BUSINESS_SESSION_STOPPED"; if(reason==null&&"DEAD".equals(root.getStatus()))reason="BUSINESS_SESSION_STOPPED";
if(reason==null)return; if(reason==null)return;
if(!terminate)throw new BusinessSessionStoppedException(reason); if(!terminate)throw new BusinessSessionStoppedException(reason);
@@ -137,7 +137,7 @@ class CollectionQueryServiceTest {
assertFalse(rows.isEmpty());for(var row:rows){assertTrue(row.executionEnabled());assertNull(row.disabledReason());assertNull(row.nextScheduledAt()); assertFalse(rows.isEmpty());for(var row:rows){assertTrue(row.executionEnabled());assertNull(row.disabledReason());assertNull(row.nextScheduledAt());
assertEquals("INVENTORY".equals(row.dataFamily()),row.publicationEnabled());} assertEquals("INVENTORY".equals(row.dataFamily()),row.publicationEnabled());}
config.setExclusiveGrantExpiresAtUtc(now.minusSeconds(1)); config.setExclusiveGrantExpiresAtUtc(now.minusSeconds(1));
for(var row:service.overview(q(null,null,null)).warehouses()){assertFalse(row.executionEnabled());assertEquals("EXCLUSIVE_GRANT_NOT_VALID",row.disabledReason());} for(var row:service.overview(q(null,null,null)).warehouses()){assertTrue(row.executionEnabled());assertNull(row.disabledReason());}
config.setEnvironment("development"); config.setEnvironment("development");
for(var row:service.overview(q(null,null,null)).warehouses()){assertFalse(row.executionEnabled());assertEquals("BUSINESS_ENVIRONMENT_REJECTED",row.disabledReason());} for(var row:service.overview(q(null,null,null)).warehouses()){assertFalse(row.executionEnabled());assertEquals("BUSINESS_ENVIRONMENT_REJECTED",row.disabledReason());}
} }
@@ -157,9 +157,9 @@ class CollectionQueryServiceTest {
assertEquals("EXCLUSIVE_GRANT_NOT_VALID",row.disabledReason()); assertEquals("EXCLUSIVE_GRANT_NOT_VALID",row.disabledReason());
} }
config.setPriorEgressClosed(true);config.setContinuousBusinessCollectionEnabled(false); config.setPriorEgressClosed(true);config.setContinuousBusinessCollectionEnabled(false);
assertFalse(service.overview(q(null,null,null)).warehouses().get(0).executionEnabled()); assertTrue(service.overview(q(null,null,null)).warehouses().get(0).executionEnabled());
} }
@Test void continuousBudgetRequiresEnabledModeAndNoDeadlineWhileBoundedBudgetsStillExpire() { @Test void plannerIdentityStillNeedsSwitchButManualRecordedDeadlineDoesNotExpire() {
enableBusiness("test");config.setContinuousBusinessCollectionEnabled(true); enableBusiness("test");config.setContinuousBusinessCollectionEnabled(true);
String continuous="{\"continuous\":true,\"startedAtUtc\":\""+now.minusDays(2)+"\",\"maxHttp\":0,\"http\":10,\"confirmedHttp\":10,\"deadlineAtUtc\":null,\"stopReason\":null}"; String continuous="{\"continuous\":true,\"startedAtUtc\":\""+now.minusDays(2)+"\",\"maxHttp\":0,\"http\":10,\"confirmedHttp\":10,\"deadlineAtUtc\":null,\"stopReason\":null}";
when(mapper.task("primary","A","LXCOL-TASK")).thenReturn(manualBusinessTask(continuous)); when(mapper.task("primary","A","LXCOL-TASK")).thenReturn(manualBusinessTask(continuous));
@@ -167,7 +167,9 @@ class CollectionQueryServiceTest {
when(mapper.task("primary","A","LXCOL-TASK")).thenReturn(manualBusinessTask(continuous.replace("\"deadlineAtUtc\":null","\"deadlineAtUtc\":\""+now.plusDays(1)+"\""))); when(mapper.task("primary","A","LXCOL-TASK")).thenReturn(manualBusinessTask(continuous.replace("\"deadlineAtUtc\":null","\"deadlineAtUtc\":\""+now.plusDays(1)+"\"")));
assertEquals("BUSINESS_SESSION_BUDGET_MISSING",service.taskDetail("LXCOL-TASK",q(null,null,null)).task().actionBlockReason()); assertEquals("BUSINESS_SESSION_BUDGET_MISSING",service.taskDetail("LXCOL-TASK",q(null,null,null)).task().actionBlockReason());
when(mapper.task("primary","A","LXCOL-TASK")).thenReturn(manualBusinessTask(businessBudget(now.minusMinutes(1).toString(),null,10))); when(mapper.task("primary","A","LXCOL-TASK")).thenReturn(manualBusinessTask(businessBudget(now.minusMinutes(1).toString(),null,10)));
assertEquals("BUSINESS_SESSION_DEADLINE",service.taskDetail("LXCOL-TASK",q(null,null,null)).task().actionBlockReason()); assertNull(service.taskDetail("LXCOL-TASK",q(null,null,null)).task().actionBlockReason());
when(mapper.task("primary","A","LXCOL-TASK")).thenReturn(manualBusinessTask(continuous.replace("\"continuous\":true","\"continuous\":false")));
assertEquals(List.of("RETRY"),service.taskDetail("LXCOL-TASK",q(null,null,null)).task().allowedActions());
config.setContinuousBusinessCollectionEnabled(false); config.setContinuousBusinessCollectionEnabled(false);
when(mapper.task("primary","A","LXCOL-TASK")).thenReturn(manualBusinessTask(continuous)); when(mapper.task("primary","A","LXCOL-TASK")).thenReturn(manualBusinessTask(continuous));
assertEquals("CONTINUOUS_COLLECTION_DISABLED",service.taskDetail("LXCOL-TASK",q(null,null,null)).task().actionBlockReason()); assertEquals("CONTINUOUS_COLLECTION_DISABLED",service.taskDetail("LXCOL-TASK",q(null,null,null)).task().actionBlockReason());
@@ -234,14 +236,14 @@ class CollectionQueryServiceTest {
enableBusiness("production");when(mapper.task("primary","A","LXCOL-TASK")).thenReturn(manualBusinessTask(null)); enableBusiness("production");when(mapper.task("primary","A","LXCOL-TASK")).thenReturn(manualBusinessTask(null));
assertEquals(List.of("RETRY"),service.taskDetail("LXCOL-TASK",q(null,null,null)).task().allowedActions()); assertEquals(List.of("RETRY"),service.taskDetail("LXCOL-TASK",q(null,null,null)).task().allowedActions());
} }
@Test void expiredBudgetOnlyAllowsProvenLocalWorkButStoppedBudgetNeverDoes() { @Test void recordedDeadlineDoesNotBlockRemoteOrLocalWorkButPersistedStopStillDoes() {
enableBusiness("test"); enableBusiness("test");
when(mapper.latestAttempts(eq("primary"),anyList())).thenReturn(List.of(run(1,"FAILED",null))); when(mapper.latestAttempts(eq("primary"),anyList())).thenReturn(List.of(run(1,"FAILED",null)));
for(boolean ready:List.of(false,true)) { for(boolean ready:List.of(false,true)) {
when(mapper.task("primary","A","LXCOL-TASK")).thenReturn(manualBusinessTask(businessBudget(now.toString(),null,5),ready)); when(mapper.task("primary","A","LXCOL-TASK")).thenReturn(manualBusinessTask(businessBudget(now.toString(),null,5),ready));
var task=service.taskDetail("LXCOL-TASK",q(null,null,null)).task(); var task=service.taskDetail("LXCOL-TASK",q(null,null,null)).task();
assertEquals(ready,task.latestAttempt().retryable()); assertTrue(task.latestAttempt().retryable());
assertEquals(ready?null:"BUSINESS_SESSION_DEADLINE",task.actionBlockReason()); assertNull(task.actionBlockReason());
} }
when(mapper.task("primary","A","LXCOL-TASK")).thenReturn(manualBusinessTask(businessBudget(now.toString(),"BUSINESS_SESSION_DEADLINE",5),true)); when(mapper.task("primary","A","LXCOL-TASK")).thenReturn(manualBusinessTask(businessBudget(now.toString(),"BUSINESS_SESSION_DEADLINE",5),true));
assertEquals("BUSINESS_SESSION_STOPPED",service.taskDetail("LXCOL-TASK",q(null,null,null)).task().actionBlockReason()); assertEquals("BUSINESS_SESSION_STOPPED",service.taskDetail("LXCOL-TASK",q(null,null,null)).task().actionBlockReason());
@@ -253,8 +255,8 @@ class CollectionQueryServiceTest {
now,null,now,null,"PARTIAL",null,null,code,null,now.minusMinutes(2),"A",1L,"hash"); now,null,now,null,"PARTIAL",null,null,code,null,now.minusMinutes(2),"A",1L,"hash");
when(mapper.runs(any(),eq(upper))).thenReturn(List.of(attempt)); when(mapper.runs(any(),eq(upper))).thenReturn(List.of(attempt));
String message=service.runs(q(null,null,null)).records().get(0).failureSummary(); String message=service.runs(q(null,null,null)).records().get(0).failureSummary();
assertTrue(message.contains(code.endsWith("HTTP_LIMIT")?"请求预算已耗尽":"已到截止时间")); assertTrue(message.contains(code.endsWith("HTTP_LIMIT")?"请求预算已耗尽":"旧采集截止时间"));
assertTrue(message.contains("不能恢复原任务")); assertTrue(message.contains(code.endsWith("HTTP_LIMIT")?"不能恢复原任务":"受控恢复"));
} }
} }
Attempt run(int attempt,String status,String snapshot) { Attempt run(int attempt,String status,String snapshot) {
@@ -28,7 +28,7 @@ class SelfopBusinessBudgetPersistenceTest {
} }
var tasks=session.getMapper(SelfopSyncTaskMapper.class); var tasks=session.getMapper(SelfopSyncTaskMapper.class);
var stale=new SelfopSyncTask();stale.setId(1L);stale.setStatus("RUNNING");stale.setBusinessRootTaskId(1L); var stale=new SelfopSyncTask();stale.setId(1L);stale.setStatus("RUNNING");stale.setBusinessRootTaskId(1L);
stale.setBusinessBudgetJson(SelfopBusinessBudget.freeze(LocalDateTime.of(2026,10,13,0,0),120)); stale.setBusinessBudgetJson(SelfopBusinessBudget.freeze(LocalDateTime.of(2026,10,13,0,0)));
assertEquals(1,tasks.insert(stale)); // NEVER updates must still allow initial freezing. assertEquals(1,tasks.insert(stale)); // NEVER updates must still allow initial freezing.
var charged=SelfopSupport.read(stale.getBusinessBudgetJson(),SelfopBusinessBudget.class); var charged=SelfopSupport.read(stale.getBusinessBudgetJson(),SelfopBusinessBudget.class);
charged.http=2;charged.confirmedHttp=1; charged.http=2;charged.confirmedHttp=1;
@@ -17,23 +17,25 @@ class SelfopBusinessBudgetTest {
} }
private final LocalDateTime start=LocalDateTime.of(2026,10,13,0,0); private final LocalDateTime start=LocalDateTime.of(2026,10,13,0,0);
@Test void serializedConsumptionAndDeadlineSurviveReload() { @Test void recordedFiniteDeadlineSurvivesReloadWithoutStoppingExistingBusiness() {
var budget=SelfopSupport.read(SelfopBusinessBudget.freeze(start,60),SelfopBusinessBudget.class); var budget=SelfopSupport.read(SelfopBusinessBudget.freeze(start),SelfopBusinessBudget.class);
budget.deadlineAtUtc=start.plusSeconds(60);
budget.http=2;budget.confirmedHttp=1; budget.http=2;budget.confirmedHttp=1;
var restored=SelfopSupport.read(SelfopSupport.json(budget),SelfopBusinessBudget.class); var restored=SelfopSupport.read(SelfopSupport.json(budget),SelfopBusinessBudget.class);
assertEquals(start.plusSeconds(60),restored.deadlineAtUtc); assertEquals(start.plusSeconds(60),restored.deadlineAtUtc);
assertEquals(2,restored.http);assertEquals(1,restored.confirmedHttp); assertEquals(2,restored.http);assertEquals(1,restored.confirmedHttp);
assertEquals(0,restored.maxHttp);assertNull(restored.limit(start.plusSeconds(1))); assertEquals(0,restored.maxHttp);assertNull(restored.limit(start.plusSeconds(1)));
assertEquals("BUSINESS_SESSION_DEADLINE",restored.limit(start.plusSeconds(60))); assertNull(restored.limit(start.plusSeconds(60)));assertNull(restored.limit(start.plusYears(1)));
assertFalse(restored.continuous);
restored.stopReason="BUSINESS_SESSION_HTTP_LIMIT"; restored.stopReason="BUSINESS_SESSION_HTTP_LIMIT";
assertEquals(restored.stopReason,restored.limit(start.plusDays(1))); assertEquals(restored.stopReason,restored.limit(start.plusDays(1)));
} }
@Test void unlimitedHttpRetainsStrictDeadlineAndAccounting() { @Test void newBusinessHasNoDeadlineAndRetainsStrictAccounting() {
assertThrows(IllegalArgumentException.class,()->SelfopBusinessBudget.freeze(null,60)); assertThrows(IllegalArgumentException.class,()->SelfopBusinessBudget.freeze(null));
assertThrows(IllegalArgumentException.class,()->SelfopBusinessBudget.freeze(start,0));
assertThrows(EgressRejectedException.class,()->new SelfopBusinessBudget().validate()); assertThrows(EgressRejectedException.class,()->new SelfopBusinessBudget().validate());
var budget=SelfopSupport.read(SelfopBusinessBudget.freeze(start,60),SelfopBusinessBudget.class); var budget=SelfopSupport.read(SelfopBusinessBudget.freeze(start),SelfopBusinessBudget.class);
assertNull(budget.deadlineAtUtc);assertFalse(budget.continuous);
budget.confirmedHttp=1; budget.confirmedHttp=1;
assertThrows(EgressRejectedException.class,budget::validate); assertThrows(EgressRejectedException.class,budget::validate);
budget.confirmedHttp=12001;budget.http=12001;budget.maxHttp=5000; budget.confirmedHttp=12001;budget.http=12001;budget.maxHttp=5000;
@@ -34,17 +34,18 @@ class SelfopEgressServiceTest {
cfg.setCollectionMode("BUSINESS");cfg.setEnvironment("unknown");assertThatThrownBy(()->reserve(Operation.WAREHOUSES,1,null,null)).hasMessage("BUSINESS_ENVIRONMENT_REJECTED"); cfg.setCollectionMode("BUSINESS");cfg.setEnvironment("unknown");assertThatThrownBy(()->reserve(Operation.WAREHOUSES,1,null,null)).hasMessage("BUSINESS_ENVIRONMENT_REJECTED");
cfg.setCollectionMode("VALIDATION");cfg.setPriorEgressClosed(false);assertThatThrownBy(()->reserve(Operation.WAREHOUSES,1,null,null)).hasMessage("EXCLUSIVE_HANDOFF_UNCONFIRMED");verifyNoInteractions(mapper); cfg.setCollectionMode("VALIDATION");cfg.setPriorEgressClosed(false);assertThatThrownBy(()->reserve(Operation.WAREHOUSES,1,null,null)).hasMessage("EXCLUSIVE_HANDOFF_UNCONFIRMED");verifyNoInteractions(mapper);
} }
@Test void testBusinessStillChargesPersistentAccountAndRequiresLiveExclusiveGrant(){ @Test void businessStillChargesPersistentAccountAndRequiresMatchingGrantIdentity(){
testBusiness(5,300); testBusiness(5,300);
var c=reserve(Operation.WAREHOUSES,1,null,null);success(c,Operation.WAREHOUSES); var c=reserve(Operation.WAREHOUSES,1,null,null);success(c,Operation.WAREHOUSES);
assertThat(budget().reservedHttp).isEqualTo(1);assertThat(budget().confirmedHttp).isEqualTo(1); assertThat(budget().reservedHttp).isEqualTo(1);assertThat(budget().confirmedHttp).isEqualTo(1);
assertThat(businessBudget().http).isEqualTo(1);assertThat(businessBudget().confirmedHttp).isEqualTo(1); assertThat(businessBudget().http).isEqualTo(1);assertThat(businessBudget().confirmedHttp).isEqualTo(1);
cfg.setExclusiveGrantExpiresAtUtc(now()); cfg.setExclusiveGrantExpiresAtUtc(now());
assertThatThrownBy(()->reserve(Operation.WAREHOUSES,1,null,null)).hasMessage("EXCLUSIVE_HANDOFF_UNCONFIRMED"); success(reserve(Operation.WAREHOUSES,1,null,null),Operation.WAREHOUSES);
assertThat(budget().reservedHttp).isEqualTo(1); cfg.setExclusiveGrantId("different");assertThatThrownBy(()->reserve(Operation.WAREHOUSES,1,null,null)).hasMessage("EXCLUSIVE_HANDOFF_UNCONFIRMED");
assertThat(budget().reservedHttp).isEqualTo(2);
} }
@Test void frozenPageSizeCannotBeExpanded(){assertThatThrownBy(()->reserve(Operation.WAREHOUSES,11,null,null)).hasMessage("VALIDATION_PAGE_SIZE");assertThat(budget().reservedHttp).isZero();} @Test void frozenPageSizeCannotBeExpanded(){assertThatThrownBy(()->reserve(Operation.WAREHOUSES,11,null,null)).hasMessage("VALIDATION_PAGE_SIZE");assertThat(budget().reservedHttp).isZero();}
@Test void continuousBusinessGrantExpiryDoesNotClipShortPendingOrPermitAndSwitchKeepsCounters(){ @Test void businessGrantExpiryDoesNotClipShortPendingOrPermitRegardlessOfPlannerSwitch(){
testBusiness(5,300);success(reserve(Operation.WAREHOUSES,1,null,null),Operation.WAREHOUSES); testBusiness(5,300);success(reserve(Operation.WAREHOUSES,1,null,null),Operation.WAREHOUSES);
cfg.setContinuousBusinessCollectionEnabled(true);cfg.setExclusiveGrantExpiresAtUtc(now().minusSeconds(1));cfg.setEgressLeaseSeconds(120); cfg.setContinuousBusinessCollectionEnabled(true);cfg.setExclusiveGrantExpiresAtUtc(now().minusSeconds(1));cfg.setEgressLeaseSeconds(120);
var c=reserve(Operation.WAREHOUSES,1,null,null); var c=reserve(Operation.WAREHOUSES,1,null,null);
@@ -54,7 +55,8 @@ class SelfopEgressServiceTest {
restarted.complete(new CompletionRequest(c,Operation.WAREHOUSES,PROVIDER,1,null,200,200,false,null)); restarted.complete(new CompletionRequest(c,Operation.WAREHOUSES,PROVIDER,1,null,200,200,false,null));
assertThat(budget().reservedHttp).isEqualTo(2);assertThat(budget().confirmedHttp).isEqualTo(2); assertThat(budget().reservedHttp).isEqualTo(2);assertThat(budget().confirmedHttp).isEqualTo(2);
cfg.setContinuousBusinessCollectionEnabled(false); cfg.setContinuousBusinessCollectionEnabled(false);
assertThatThrownBy(()->reserve(Operation.WAREHOUSES,1,null,null)).hasMessage("EXCLUSIVE_HANDOFF_UNCONFIRMED"); clock.advance(5);success(reserve(Operation.WAREHOUSES,1,null,null),Operation.WAREHOUSES);
assertThat(budget().reservedHttp).isEqualTo(3);assertThat(businessBudget().continuous).isFalse();
} }
@Test void continuousFlagNeverBypassesValidationWindowGrantIdentityOrHandoff(){ @Test void continuousFlagNeverBypassesValidationWindowGrantIdentityOrHandoff(){
cfg.setContinuousBusinessCollectionEnabled(true);cfg.setExclusiveGrantExpiresAtUtc(null); cfg.setContinuousBusinessCollectionEnabled(true);cfg.setExclusiveGrantExpiresAtUtc(null);
@@ -64,15 +66,15 @@ class SelfopEgressServiceTest {
cfg.setExclusiveGrantId("grant");cfg.setPriorEgressClosed(false); cfg.setExclusiveGrantId("grant");cfg.setPriorEgressClosed(false);
assertThatThrownBy(()->reserve(Operation.WAREHOUSES,1,null,null)).hasMessage("EXCLUSIVE_HANDOFF_UNCONFIRMED");verifyNoInteractions(mapper); assertThatThrownBy(()->reserve(Operation.WAREHOUSES,1,null,null)).hasMessage("EXCLUSIVE_HANDOFF_UNCONFIRMED");verifyNoInteractions(mapper);
} }
@Test void continuousGrantDoesNotRenewFiniteManualRootOrRelaxFence(){ @Test void oldManualDeadlineDoesNotClipPermitOrRelaxFence(){
testBusiness(5,10);cfg.setContinuousBusinessCollectionEnabled(true);cfg.setExclusiveGrantExpiresAtUtc(null); testBusiness(5,10);cfg.setContinuousBusinessCollectionEnabled(true);cfg.setExclusiveGrantExpiresAtUtc(null);
account.setFenceToken(8L); account.setFenceToken(8L);
assertThatThrownBy(()->reserve(Operation.WAREHOUSES,1,null,null)).hasMessage("EGRESS_FENCE_LOST"); assertThatThrownBy(()->reserve(Operation.WAREHOUSES,1,null,null)).hasMessage("EGRESS_FENCE_LOST");
account.setFenceToken(7L);var c=reserve(Operation.WAREHOUSES,1,null,null); account.setFenceToken(7L);var c=reserve(Operation.WAREHOUSES,1,null,null);
assertThat(budget().pending.expiresAtUtc).isEqualTo(businessBudget().deadlineAtUtc); assertThat(budget().pending.expiresAtUtc).isAfter(businessBudget().deadlineAtUtc);
clock.advance(10); clock.advance(10);
assertThatThrownBy(()->service.authorize(new AuthorizationRequest(c,Operation.WAREHOUSES,PROVIDER))).hasMessage("BUSINESS_SESSION_DEADLINE"); assertThat(service.authorize(new AuthorizationRequest(c,Operation.WAREHOUSES,PROVIDER)).expiresAt()).isEqualTo(clock.instant().plusSeconds(30));
assertThat(budget().pending.consumed).isFalse(); assertThat(budget().pending.consumed).isTrue();assertThat(businessBudget().continuous).isFalse();
} }
@Test void continuousBusinessAcrossMidnightRetainsUnknownPendingAndReviewAcrossRestart(){ @Test void continuousBusinessAcrossMidnightRetainsUnknownPendingAndReviewAcrossRestart(){
business();cfg.setContinuousBusinessCollectionEnabled(true);cfg.setExclusiveGrantExpiresAtUtc(null); business();cfg.setContinuousBusinessCollectionEnabled(true);cfg.setExclusiveGrantExpiresAtUtc(null);
@@ -188,7 +190,9 @@ class SelfopEgressServiceTest {
} }
} }
private void business(){cfg.setCollectionMode("BUSINESS");cfg.setEnvironment("production");task.setExecutionPurpose("BUSINESS");run.setExecutionPurpose("BUSINESS");} private void business(){cfg.setCollectionMode("BUSINESS");cfg.setEnvironment("production");task.setExecutionPurpose("BUSINESS");run.setExecutionPurpose("BUSINESS");}
private void testBusiness(int http,int seconds){business();cfg.setEnvironment("test");task.setBusinessRootTaskId(1L);task.setBusinessBudgetJson(SelfopBusinessBudget.freeze(now(),seconds));} private void testBusiness(int http,int seconds){business();cfg.setEnvironment("test");task.setBusinessRootTaskId(1L);
var recorded=new SelfopBusinessBudget();recorded.startedAtUtc=now();recorded.deadlineAtUtc=now().plusSeconds(seconds);
task.setBusinessBudgetJson(SelfopSupport.json(recorded));}
private SelfopBusinessBudget businessBudget(){return SelfopSupport.read(task.getBusinessBudgetJson(),SelfopBusinessBudget.class);} private SelfopBusinessBudget businessBudget(){return SelfopSupport.read(task.getBusinessBudgetJson(),SelfopBusinessBudget.class);}
private void continuousBusinessRoot(){ private void continuousBusinessRoot(){
business();cfg.setEnvironment("test");cfg.setContinuousBusinessCollectionEnabled(true); business();cfg.setEnvironment("test");cfg.setContinuousBusinessCollectionEnabled(true);
@@ -298,7 +302,7 @@ class SelfopEgressServiceTest {
verify(mapper,never()).terminateBusinessTasks(any(),any());verify(mapper,never()).terminateValidationRuns(any(),any(),any(),any()); verify(mapper,never()).terminateBusinessTasks(any(),any());verify(mapper,never()).terminateValidationRuns(any(),any(),any(),any());
} }
@Test void businessWithoutDailyQuotaStillUsesIntervalAndActual429Review(){ @Test void businessWithoutDailyQuotaStillUsesIntervalAndActual429Review(){
testBusiness(1,300);cfg.setDailyRequestBudget(0);cfg.setBusinessMinRequestIntervalMs(1200); testBusiness(1,1);clock.advance(2);cfg.setExclusiveGrantExpiresAtUtc(now().minusSeconds(1));cfg.setDailyRequestBudget(0);cfg.setBusinessMinRequestIntervalMs(1200);
var c=reserve(Operation.WAREHOUSES,1,null,null);service.authorize(new AuthorizationRequest(c,Operation.WAREHOUSES,PROVIDER)); var c=reserve(Operation.WAREHOUSES,1,null,null);service.authorize(new AuthorizationRequest(c,Operation.WAREHOUSES,PROVIDER));
service.complete(new CompletionRequest(c,Operation.WAREHOUSES,PROVIDER,1,null,200,200,false,null)); service.complete(new CompletionRequest(c,Operation.WAREHOUSES,PROVIDER,1,null,200,200,false,null));
assertThatThrownBy(()->reserve(Operation.OUTBOUND_QUERY,1,null,null)).hasMessage("ACCOUNT_INTERVAL_WAIT"); assertThatThrownBy(()->reserve(Operation.OUTBOUND_QUERY,1,null,null)).hasMessage("ACCOUNT_INTERVAL_WAIT");
@@ -306,28 +310,50 @@ class SelfopEgressServiceTest {
service.complete(new CompletionRequest(next,Operation.OUTBOUND_QUERY,PROVIDER,1,"LINGXING_WMS_HTTP_RATE_LIMITED",429,null,true,null)); service.complete(new CompletionRequest(next,Operation.OUTBOUND_QUERY,PROVIDER,1,"LINGXING_WMS_HTTP_RATE_LIMITED",429,null,true,null));
assertThat(budget().reservedHttp).isEqualTo(2);assertThat(budget().confirmedHttp).isEqualTo(2);assertThat(budget().rateLimitFailures).isEqualTo(1); assertThat(budget().reservedHttp).isEqualTo(2);assertThat(budget().confirmedHttp).isEqualTo(2);assertThat(budget().rateLimitFailures).isEqualTo(1);
assertThat(account.getApiPauseReason()).isEqualTo("RATE_LIMIT_COOLDOWN"); assertThat(account.getApiPauseReason()).isEqualTo("RATE_LIMIT_COOLDOWN");
clock.advance(cfg.getCooldownSeconds()+1);var afterCooldown=reserve(Operation.WAREHOUSES,1,null,null);
service.authorize(new AuthorizationRequest(afterCooldown,Operation.WAREHOUSES,PROVIDER));
service.complete(new CompletionRequest(afterCooldown,Operation.WAREHOUSES,PROVIDER,1,"LINGXING_WMS_AUTHORIZATION_REJECTED",401,null,true,null));
clock.advance(2);assertThatThrownBy(()->reserve(Operation.WAREHOUSES,1,null,null)).hasMessage("ACCOUNT_REQUIRES_REVIEW");
assertThat(account.getApiPauseReason()).isEqualTo("AUTHORIZATION_REQUIRES_REVIEW");
} }
@Test void testBusinessDeadlineClipsPermitAndAuthorizeRechecksBeforeHttp(){ @Test void oldManualDeadlineRemovalKeepsSingleFlightAndUnknownOutcomeProtection(){
testBusiness(5,10);var c=reserve(Operation.WAREHOUSES,1,null,null);assertThat(budget().pending.expiresAtUtc).isEqualTo(businessBudget().deadlineAtUtc); testBusiness(5,1);clock.advance(2);cfg.setExclusiveGrantExpiresAtUtc(null);
var c=reserve(Operation.WAREHOUSES,1,null,null);
assertThatThrownBy(()->reserve(Operation.WAREHOUSES,1,null,null)).hasMessage("ACCOUNT_REQUEST_INFLIGHT");
service.authorize(new AuthorizationRequest(c,Operation.WAREHOUSES,PROVIDER));clock.advance(31);
var restarted=new SelfopEgressService(mapper,cfg,clock);
assertThatThrownBy(()->restarted.reserve(context(),Operation.WAREHOUSES,1,null,null,INTERNAL)).hasMessage("ACCOUNT_COOLDOWN_WAIT");
assertThat(account.getApiPauseReason()).isEqualTo("REQUEST_OUTCOME_UNKNOWN");assertThat(budget().pending.abandoned).isTrue();
assertThat(businessBudget().http).isEqualTo(1);assertThat(businessBudget().confirmedHttp).isZero();
}
@Test void existingManualBusinessContinuesPastBothRecordedDeadlineAndGrantWithoutReset(){
testBusiness(5,10);cfg.setExclusiveGrantExpiresAtUtc(now().plusSeconds(10));cfg.setContinuousBusinessCollectionEnabled(false);
task.setRequestKey("manual:2026-10-06");task.setRequestedBy("operator");task.setRoutingManifestJson("{\"nextPage\":37,\"scan\":2}");
var original=businessBudget();original.http=8820;original.confirmedHttp=8819;task.setBusinessBudgetJson(SelfopSupport.json(original));
var c=reserve(Operation.WAREHOUSES,1,null,null);assertThat(budget().pending.expiresAtUtc).isAfter(original.deadlineAtUtc);
clock.advance(9);var permit=service.authorize(new AuthorizationRequest(c,Operation.WAREHOUSES,PROVIDER)); clock.advance(9);var permit=service.authorize(new AuthorizationRequest(c,Operation.WAREHOUSES,PROVIDER));
assertThat(permit.expiresAt()).isEqualTo(businessBudget().deadlineAtUtc.toInstant(ZoneOffset.UTC)); assertThat(permit.expiresAt()).isEqualTo(clock.instant().plusSeconds(30));
clock.advance(2);service.complete(new CompletionRequest(c,Operation.WAREHOUSES,PROVIDER,1,null,200,200,false,null)); clock.advance(2);service.complete(new CompletionRequest(c,Operation.WAREHOUSES,PROVIDER,1,null,200,200,false,null));
assertThat(businessBudget().confirmedHttp).isEqualTo(1); clock.advance(5);var restarted=new SelfopEgressService(mapper,cfg,clock);success(restarted.reserve(context(),Operation.WAREHOUSES,1,null,null,INTERNAL),Operation.WAREHOUSES);
assertThatThrownBy(()->reserve(Operation.WAREHOUSES,1,null,null)).hasMessage("BUSINESS_SESSION_DEADLINE"); assertThat(businessBudget().http).isEqualTo(8822);assertThat(businessBudget().confirmedHttp).isEqualTo(8821);
assertThat(businessBudget().stopReason).isEqualTo("BUSINESS_SESSION_DEADLINE"); assertThat(businessBudget().startedAtUtc).isEqualTo(original.startedAtUtc);assertThat(businessBudget().deadlineAtUtc).isEqualTo(original.deadlineAtUtc);
assertThat(businessBudget().continuous).isFalse();assertThat(businessBudget().stopReason).isNull();
assertThat(task.getId()).isEqualTo(1L);assertThat(task.getBusinessRootTaskId()).isEqualTo(1L);assertThat(task.getRequestKey()).isEqualTo("manual:2026-10-06");
assertThat(task.getRequestedBy()).isEqualTo("operator");assertThat(task.getRoutingManifestJson()).isEqualTo("{\"nextPage\":37,\"scan\":2}");
verify(mapper,never()).terminateBusinessTasks(any(),any());
} }
@Test void testBusinessExpiredOrStoppedPreparedAttemptCannotAuthorize(){ @Test void businessShortPendingExpiryStillRejectsAfterSessionDeadlineRemoved(){
testBusiness(5,10);var c=reserve(Operation.WAREHOUSES,1,null,null);clock.advance(10); testBusiness(5,10);cfg.setEgressLeaseSeconds(20);var c=reserve(Operation.WAREHOUSES,1,null,null);clock.advance(20);
assertThatThrownBy(()->service.authorize(new AuthorizationRequest(c,Operation.WAREHOUSES,PROVIDER))) assertThatThrownBy(()->service.authorize(new AuthorizationRequest(c,Operation.WAREHOUSES,PROVIDER)))
.hasMessage("BUSINESS_SESSION_DEADLINE").satisfies(e->assertThat(((EgressRejectedException)e).sampleDone()).isFalse()); .hasMessage("EGRESS_PERMIT_REPLAY_OR_EXPIRED").satisfies(e->assertThat(((EgressRejectedException)e).sampleDone()).isFalse());
assertThat(budget().pending.consumed).isFalse();assertThat(businessBudget().http).isEqualTo(1);assertThat(businessBudget().confirmedHttp).isZero(); assertThat(budget().pending.consumed).isFalse();assertThat(businessBudget().http).isEqualTo(1);assertThat(businessBudget().confirmedHttp).isZero();
assertThat(task.getStatus()).isEqualTo("DEAD"); assertThat(task.getStatus()).isEqualTo("RUNNING");
} }
@Test void testBusinessMissingBudgetOrWrongRootFailsClosedWithoutBackfill(){ @Test void testBusinessMissingBudgetOrWrongRootFailsClosedWithoutBackfill(){
testBusiness(5,300);task.setBusinessBudgetJson(null); testBusiness(5,300);task.setBusinessBudgetJson(null);
assertThatThrownBy(()->reserve(Operation.WAREHOUSES,1,null,null)).hasMessage("BUSINESS_SESSION_ROOT_INVALID"); assertThatThrownBy(()->reserve(Operation.WAREHOUSES,1,null,null)).hasMessage("BUSINESS_SESSION_ROOT_INVALID");
assertThat(task.getBusinessBudgetJson()).isNull();assertThat(budget().reservedHttp).isZero(); assertThat(task.getBusinessBudgetJson()).isNull();assertThat(budget().reservedHttp).isZero();
task.setBusinessBudgetJson(SelfopBusinessBudget.freeze(now(),300));task.setBusinessRootTaskId(999L); task.setBusinessBudgetJson(SelfopBusinessBudget.freeze(now()));task.setBusinessRootTaskId(999L);
assertThatThrownBy(()->reserve(Operation.WAREHOUSES,1,null,null)).hasMessage("BUSINESS_SESSION_ROOT_INVALID"); assertThatThrownBy(()->reserve(Operation.WAREHOUSES,1,null,null)).hasMessage("BUSINESS_SESSION_ROOT_INVALID");
assertThat(businessBudget().http).isZero(); assertThat(businessBudget().http).isZero();
} }
@@ -403,7 +403,7 @@ class SelfopTaskServiceTest {
service.create(root); service.create(root);
var budget=read(root.getBusinessBudgetJson(),com.baiyue.lastleg.selfoperated.analytics.request.SelfopBusinessBudget.class); var budget=read(root.getBusinessBudgetJson(),com.baiyue.lastleg.selfoperated.analytics.request.SelfopBusinessBudget.class);
assertThat(root.getBusinessRootTaskId()).isEqualTo(root.getId());assertThat(budget.maxHttp).isZero(); assertThat(root.getBusinessRootTaskId()).isEqualTo(root.getId());assertThat(budget.maxHttp).isZero();
assertThat(budget.deadlineAtUtc).isEqualTo(budget.startedAtUtc.plusSeconds(91)); assertThat(budget.deadlineAtUtc).isNull();assertThat(budget.continuous).isFalse();
var child=task("READY");child.setId(42L);child.setBusinessRootTaskId(root.getId()); var child=task("READY");child.setId(42L);child.setBusinessRootTaskId(root.getId());
assertThat(SelfopRequestContext.forTask(child,run("FETCHING"),config).rootTaskId()).isEqualTo(root.getId()); assertThat(SelfopRequestContext.forTask(child,run("FETCHING"),config).rootTaskId()).isEqualTo(root.getId());
config.setEnvironment("production"); config.setEnvironment("production");
@@ -431,19 +431,29 @@ class SelfopTaskServiceTest {
assertThat(read(root.getBusinessBudgetJson(),com.baiyue.lastleg.selfoperated.analytics.request.SelfopBusinessBudget.class).http).isEqualTo(7); assertThat(read(root.getBusinessBudgetJson(),com.baiyue.lastleg.selfoperated.analytics.request.SelfopBusinessBudget.class).http).isEqualTo(7);
verify(runs).insert(next); verify(runs).insert(next);
} }
@Test void expiredFetchCannotBeginAndManualRetryCannotRestartItsSession() { @Test void existingManualDiscoveryPastDeadlineBeginsFromPersistedCheckpointAndIdentity() {
var root=boundedRoot(2,utcNow().minusSeconds(1)); var root=boundedRoot(2,utcNow().minusSeconds(1));
assertThatThrownBy(()->service.begin(1L,lease)).isInstanceOf(SelfopTaskService.BusinessSessionStoppedException.class); root.setTaskRole("ACCOUNT_DISCOVERY");root.setScopeId(null);root.setWarehouseCode(null);root.setDiscoveryScopeId(99L);
assertThatThrownBy(()->service.retry(1L,"new-key","operator","retry")) root.setRequestKey("manual:2026-10-06");root.setTaskType("BACKFILL");root.setAttemptCount(3);
.isInstanceOf(SelfopTaskService.BusinessSessionStoppedException.class); root.setQueryFromUtc(LocalDateTime.of(2026,10,5,16,0));root.setQueryUntilUtc(LocalDateTime.of(2026,10,6,16,0));
verify(runs,never()).insert(any(SelfopSyncRun.class));verify(tasks,never()).insert(any(SelfopSyncTask.class)); var manifest=new SelfopRoutingManifest();manifest.setNextPage(37);manifest.setScan(2);manifest.setPages(100);manifest.setTotal(1000L);
assertThat(read(root.getBusinessBudgetJson(),com.baiyue.lastleg.selfoperated.analytics.request.SelfopBusinessBudget.class).http).isEqualTo(2); root.setRoutingManifestJson(json(manifest));String checkpoint=root.getRoutingManifestJson(),budget=root.getBusinessBudgetJson();
var account=new SelfopOutboundDiscoveryScope();account.setId(99L);when(locks.lockDiscovery(99L)).thenReturn(account);
var next=service.begin(1L,new SelfopLeaseService.Lease(99L,"owner",7L,true));
assertThat(next.getAttemptNo()).isEqualTo(4);assertThat(next.getTaskId()).isEqualTo(1L);assertThat(next.getStatus()).isEqualTo("FETCHING");
assertThat(root.getRoutingManifestJson()).isEqualTo(checkpoint);assertThat(root.getBusinessBudgetJson()).isEqualTo(budget);
assertThat(root.getBusinessRootTaskId()).isEqualTo(1L);assertThat(root.getRequestKey()).isEqualTo("manual:2026-10-06");
assertThat(root.getRequestedBy()).isEqualTo("operator");assertThat(root.getQueryFromUtc()).isEqualTo(LocalDateTime.of(2026,10,5,16,0));
verify(runs).insert(next);verify(tasks,never()).insert(any(SelfopSyncTask.class));
} }
@Test void expiredYieldAndBudgetWaitEndInsteadOfSchedulingAnotherDispatch() { @Test void pastRecordedDeadlineYieldAndBudgetWaitKeepSchedulingSameRoot() {
boundedRoot(2,utcNow().minusSeconds(1)); var root=boundedRoot(2,utcNow().minusSeconds(1));root.setStatus("RUNNING");
assertThatThrownBy(()->service.yield(1L,20L,lease)).isInstanceOf(SelfopTaskService.BusinessSessionStoppedException.class); var active=run("FETCHING");when(locks.lockRun(20L)).thenReturn(active);String budget=root.getBusinessBudgetJson();
assertThatThrownBy(()->service.waitForBudget(1L,20L,utcNow().plusDays(1),lease)).isInstanceOf(SelfopTaskService.BusinessSessionStoppedException.class); service.yield(1L,20L,lease);assertThat(root.getStatus()).isEqualTo("RETRY_WAIT");assertThat(active.getStatus()).isEqualTo("YIELDED");
verify(tasks,never()).updateById(any(SelfopSyncTask.class));verify(runs,never()).updateById(any(SelfopSyncRun.class)); root.setStatus("RUNNING");active.setStatus("FETCHING");var until=utcNow().plusMinutes(5);
service.waitForBudget(1L,20L,until,lease);
assertThat(root.getNextRetryAtUtc()).isEqualTo(until);assertThat(root.getBusinessBudgetJson()).isEqualTo(budget);
verify(tasks,never()).update(any(),any());
} }
@Test void continuousFailureStopsInsteadOfBecomingAnotherScheduledRetry(){ @Test void continuousFailureStopsInsteadOfBecomingAnotherScheduledRetry(){
var root=boundedRoot(0,utcNow().plusMinutes(3));config.setContinuousBusinessCollectionEnabled(true); var root=boundedRoot(0,utcNow().plusMinutes(3));config.setContinuousBusinessCollectionEnabled(true);
@@ -467,7 +477,7 @@ class SelfopTaskServiceTest {
assertThat(read(root.getBusinessBudgetJson(),com.baiyue.lastleg.selfoperated.analytics.request.SelfopBusinessBudget.class).stopReason).isEqualTo("BUSINESS_SESSION_HTTP_LIMIT"); assertThat(read(root.getBusinessBudgetJson(),com.baiyue.lastleg.selfoperated.analytics.request.SelfopBusinessBudget.class).stopReason).isEqualTo("BUSINESS_SESSION_HTTP_LIMIT");
} }
@Test void manualRetryAuditRetainsOriginalRootAndCannotFreezeAnotherAllowance() { @Test void manualRetryAuditRetainsOriginalRootAndCannotFreezeAnotherAllowance() {
var root=boundedRoot(3,utcNow().plusMinutes(3));root.setStatus("RETRY_WAIT");String frozen=root.getBusinessBudgetJson(); var root=boundedRoot(3,utcNow().minusSeconds(1));root.setStatus("RETRY_WAIT");String frozen=root.getBusinessBudgetJson();
config.setRequestBudget(100);config.setTimeBudgetSeconds(900); config.setRequestBudget(100);config.setTimeBudgetSeconds(900);
assertThat(service.retry(1L,"retry-bounded","operator","retry failed attempt")).isSameAs(root); assertThat(service.retry(1L,"retry-bounded","operator","retry failed attempt")).isSameAs(root);
var saved=org.mockito.ArgumentCaptor.forClass(SelfopSyncTask.class);verify(tasks).insert(saved.capture()); var saved=org.mockito.ArgumentCaptor.forClass(SelfopSyncTask.class);verify(tasks).insert(saved.capture());
@@ -489,8 +499,10 @@ class SelfopTaskServiceTest {
order.verify(locks,times(2)).lockTask(1L); order.verify(locks,times(2)).lockTask(1L);
} }
@Test void incompleteManifestCannotUseLocalWorkExceptionAfterDeadline() { @Test void incompleteManifestNeverBypassesExplicitFailureStop() {
var root=boundedRoot(1,utcNow().minusSeconds(1));when(scopes.selectById(10L)).thenReturn(scope("OUTBOUND")); var root=boundedRoot(1,utcNow().minusSeconds(1));when(scopes.selectById(10L)).thenReturn(scope("OUTBOUND"));
var budget=read(root.getBusinessBudgetJson(),com.baiyue.lastleg.selfoperated.analytics.request.SelfopBusinessBudget.class);
budget.stopReason="AUTHORIZATION_REJECTED";root.setBusinessBudgetJson(json(budget));
for(String manifest:List.of("{\"directoryComplete\":true}","{\"directoryComplete\":true,\"objects\":[]}", for(String manifest:List.of("{\"directoryComplete\":true}","{\"directoryComplete\":true,\"objects\":[]}",
"{\"directoryComplete\":true,\"objects\":{\"one\":{\"effectiveStageId\":\" \"}}}")) { "{\"directoryComplete\":true,\"objects\":{\"one\":{\"effectiveStageId\":\" \"}}}")) {
root.setRoutingManifestJson(manifest); root.setRoutingManifestJson(manifest);