生产更新:同步已提交的海外仓分析及业务修复(2026-10-16) #27
@@ -4103,6 +4103,7 @@ CREATE TABLE ll_selfop_sync_stage (
|
||||
UNIQUE KEY uk_selfop_stage_object (stage_batch_id,object_type,object_key_hash),
|
||||
KEY idx_selfop_stage_history (object_type,object_key_hash,stage_batch_id),
|
||||
KEY idx_selfop_stage_observation (object_type,object_key_hash,validation_status,observation_sequence),
|
||||
KEY idx_selfop_stage_fact_observation (observation_sequence,warehouse_code,object_type,validation_status,run_id,observed_at_utc),
|
||||
KEY idx_selfop_stage_route (stage_batch_id,warehouse_code,validation_status),
|
||||
CONSTRAINT chk_selfop_sync_stage CHECK (first_page_no > 0 AND (object_type <> 'OUTBOUND_ORDER' OR validation_status <> 'VALID' OR observation_sequence IS NOT NULL))
|
||||
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_bin COMMENT='待验证事实与修订证据';
|
||||
|
||||
@@ -3816,6 +3816,7 @@ CREATE TABLE ll_selfop_sync_stage (
|
||||
UNIQUE KEY uk_selfop_stage_object (stage_batch_id,object_type,object_key_hash),
|
||||
KEY idx_selfop_stage_history (object_type,object_key_hash,stage_batch_id),
|
||||
KEY idx_selfop_stage_observation (object_type,object_key_hash,validation_status,observation_sequence),
|
||||
KEY idx_selfop_stage_fact_observation (observation_sequence,warehouse_code,object_type,validation_status,run_id,observed_at_utc),
|
||||
KEY idx_selfop_stage_route (stage_batch_id,warehouse_code,validation_status),
|
||||
CONSTRAINT chk_selfop_sync_stage CHECK (first_page_no > 0 AND (object_type <> 'OUTBOUND_ORDER' OR validation_status <> 'VALID' OR observation_sequence IS NOT NULL))
|
||||
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_bin COMMENT='待验证事实与修订证据';
|
||||
|
||||
+10
@@ -94,6 +94,16 @@ class SelfoperatedAnalyticsSchemaContractTest {
|
||||
assertTrue(ddl.get("ll_selfop_inventory_snapshot").contains("snapshot_run_id,source_account,customer_code,warehouse_code,sku_code,product_quality"));
|
||||
}
|
||||
|
||||
@Test
|
||||
void factObservationLookupHasEqualityPrefixAndCoversProfileEvidence() throws Exception {
|
||||
var ddl = tables(Files.readString(Path.of("src/main/resources/db/lastleg.sql")));
|
||||
var stage = ddl.get("ll_selfop_sync_stage");
|
||||
assertTrue(stage.contains("KEY idx_selfop_stage_fact_observation (observation_sequence,warehouse_code,object_type,validation_status,run_id,observed_at_utc)"));
|
||||
// Observation lookup and business-key history serve different consumers; both are required.
|
||||
assertTrue(stage.contains("KEY idx_selfop_stage_history (object_type,object_key_hash,stage_batch_id)"));
|
||||
assertTrue(stage.contains("UNIQUE KEY uk_selfop_stage_object (stage_batch_id,object_type,object_key_hash)"));
|
||||
}
|
||||
|
||||
@Test
|
||||
void readAndSyncPermissionsAreIndependentWithThreeAuthorizedAdminMenus() throws Exception {
|
||||
var seed = Files.readString(Path.of("../iam/src/main/resources/db/selfoperated-analytics-permissions.sql"));
|
||||
|
||||
+17
@@ -19,6 +19,7 @@ public class OutboundAggregateMapperTest {
|
||||
sql("CREATE TABLE ll_selfop_outbound_order (id BIGINT PRIMARY KEY,source_account VARCHAR(64),customer_code VARCHAR(64),warehouse_code VARCHAR(64),sales_eligible INT,outbound_at_utc TIMESTAMP,business_date DATE,source_timezone VARCHAR(64),business_timezone VARCHAR(64),last_observation_sequence BIGINT)");
|
||||
sql("CREATE TABLE ll_selfop_outbound_item (id BIGINT PRIMARY KEY,order_id BIGINT,sku_code VARCHAR(128),product_name VARCHAR(255),quantity_unit VARCHAR(32),actual_quantity DECIMAL(20,6))");
|
||||
sql("CREATE TABLE ll_selfop_sync_stage (id BIGINT PRIMARY KEY,run_id BIGINT,warehouse_code VARCHAR(64),object_type VARCHAR(32),validation_status VARCHAR(24),observation_sequence BIGINT,observed_at_utc TIMESTAMP)");
|
||||
sql("CREATE INDEX idx_selfop_stage_fact_observation ON ll_selfop_sync_stage(observation_sequence,warehouse_code,object_type,validation_status,run_id,observed_at_utc)");
|
||||
sql("CREATE TABLE ll_selfop_sync_run (id BIGINT PRIMARY KEY,source_account VARCHAR(64),task_id BIGINT,execution_purpose VARCHAR(24))");
|
||||
sql("CREATE TABLE ll_selfop_sync_task (id BIGINT PRIMARY KEY,execution_purpose VARCHAR(24),validation_root_task_id BIGINT,task_role VARCHAR(32) DEFAULT 'WAREHOUSE_SYNC',scope_id BIGINT,scope_hash VARCHAR(64),apply_state VARCHAR(24),query_from_utc TIMESTAMP,query_until_utc TIMESTAMP,routing_manifest_json VARCHAR)");
|
||||
sql("CREATE ALIAS JSON_EXTRACT FOR 'com.baiyue.lastleg.selfoperated.analytics.outbound.OutboundAggregateMapperTest.jsonExtract'");
|
||||
@@ -62,6 +63,22 @@ public class OutboundAggregateMapperTest {
|
||||
assertEquals(1,mapper.invalidFacts("demo","A","Asia/Shanghai","Asia/Shanghai","PCS"));
|
||||
assertTrue(mapper.profileBatch("demo","A",null,null).isEmpty());
|
||||
}
|
||||
@Test void indexedObservationJoinRequiresMatchingWarehouseTypeValidityAndAccount() throws Exception {
|
||||
order(1,"demo","C1","A","SKU",1,"2026-10-01 00:00:00",2);
|
||||
order(2,"other","C1","A","SKU",1,"2026-10-01 00:00:00",9);
|
||||
sql("DELETE FROM ll_selfop_sync_stage WHERE id=1");
|
||||
sql("UPDATE ll_selfop_sync_stage SET observation_sequence=1 WHERE id=2");
|
||||
sql("INSERT INTO ll_selfop_sync_stage VALUES (3,1,'OTHER','OUTBOUND_ORDER','VALID',1,'2026-10-12 01:00:00'),"
|
||||
+"(4,1,'A','PRODUCT_STOCK','VALID',1,'2026-10-12 01:00:00'),"
|
||||
+"(5,1,'A','OUTBOUND_ORDER','INVALID',1,'2026-10-12 01:00:00')");
|
||||
assertEquals(1,mapper.invalidFacts("demo","A","Asia/Shanghai","Asia/Shanghai","PCS"));
|
||||
assertTrue(mapper.profileBatch("demo","A",null,null).isEmpty());
|
||||
sql("INSERT INTO ll_selfop_sync_stage VALUES (6,1,'A','OUTBOUND_ORDER','VALID',1,'2026-10-12 02:00:00')");
|
||||
assertEquals(0,mapper.invalidFacts("demo","A","Asia/Shanghai","Asia/Shanghai","PCS"));
|
||||
var profiles=mapper.profileBatch("demo","A",null,null);assertEquals(1,profiles.size());
|
||||
assertEquals(1L,profiles.get(0).getEvidenceRunId());assertEquals(1L,profiles.get(0).getLastSeenRunId());
|
||||
assertEquals(LocalDateTime.of(2026,10,12,2,0),profiles.get(0).getFirstSeenAtUtc());
|
||||
}
|
||||
@Test void databaseResultPagesAndMultiRowInsertsAreBounded() throws Exception {
|
||||
// Insert synthetic local orders without any provider I/O.
|
||||
try(Connection c=data.getConnection();PreparedStatement statement=c.prepareStatement("INSERT INTO ll_selfop_outbound_order VALUES (?,'demo','C1','A',1,'2026-10-01 00:00:00','2026-10-01','Asia/Shanghai','Asia/Shanghai',?)");
|
||||
|
||||
+216
@@ -0,0 +1,216 @@
|
||||
package com.baiyue.lastleg.selfoperated.analytics.outbound;
|
||||
|
||||
import com.baomidou.mybatisplus.core.MybatisConfiguration;
|
||||
import com.baomidou.mybatisplus.core.toolkit.IdWorker;
|
||||
import com.baomidou.mybatisplus.extension.spring.MybatisSqlSessionFactoryBean;
|
||||
import com.baiyue.lastleg.selfoperated.analytics.mapper.*;
|
||||
import com.baiyue.lastleg.selfoperated.analytics.model.*;
|
||||
import com.fasterxml.jackson.databind.ObjectMapper;
|
||||
import org.aopalliance.intercept.MethodInterceptor;
|
||||
import org.apache.ibatis.executor.Executor;
|
||||
import org.apache.ibatis.logging.nologging.NoLoggingImpl;
|
||||
import org.apache.ibatis.mapping.MappedStatement;
|
||||
import org.apache.ibatis.plugin.*;
|
||||
import org.junit.jupiter.api.Test;
|
||||
import org.junit.jupiter.api.condition.EnabledIfEnvironmentVariable;
|
||||
import org.mybatis.spring.SqlSessionTemplate;
|
||||
import org.springframework.aop.framework.ProxyFactory;
|
||||
import org.springframework.jdbc.core.JdbcTemplate;
|
||||
import org.springframework.jdbc.datasource.*;
|
||||
import org.springframework.transaction.annotation.AnnotationTransactionAttributeSource;
|
||||
import org.springframework.transaction.interceptor.TransactionInterceptor;
|
||||
import org.springframework.transaction.support.*;
|
||||
import java.math.BigDecimal;
|
||||
import java.nio.file.*;
|
||||
import java.time.*;
|
||||
import java.util.*;
|
||||
import static com.baiyue.lastleg.selfoperated.analytics.model.SelfopSupport.*;
|
||||
import static org.junit.jupiter.api.Assertions.*;
|
||||
|
||||
/** Explicitly enabled TEST database harness. No Boot context, collector, HTTP client or publication service. */
|
||||
@EnabledIfEnvironmentVariable(named="SELFOP_TEST_MYSQL_CANDIDATE_BUILD",matches="true")
|
||||
class OutboundCandidateBuildMySqlAcceptanceTest {
|
||||
private final Map<String,Object> report=new LinkedHashMap<>();
|
||||
private final List<Map<String,Object>> commits=new ArrayList<>();
|
||||
private final ObjectMapper jsonMapper=new ObjectMapper().findAndRegisterModules();
|
||||
private final String token=UUID.randomUUID().toString().replace("-","");
|
||||
private final String account="SELFOP-BUILD-"+token,warehouse="SYNTH-BUILD",owner="build-harness-"+token;
|
||||
private JdbcTemplate jdbc;
|
||||
private TransactionTemplate transactions;
|
||||
private long scopeId,taskId,runId,originTask,originRun;
|
||||
private Path output,progressFile;
|
||||
private final TransactionEvidence transactionEvidence=new TransactionEvidence();
|
||||
|
||||
@Test void realCandidateBuildOnIndependentSyntheticFacts() throws Exception {
|
||||
assertEquals("test",required("CHONGCHONG_ENV").toLowerCase(Locale.ROOT));
|
||||
String label=required("SELFOP_BUILD_LABEL");assertTrue(label.matches("r[1-9][0-9]*"));
|
||||
output=Path.of("target","selfop-candidate-build-"+label+".json");
|
||||
progressFile=Path.of(output+".progress.json");
|
||||
Path reservation=Path.of(output+".reserved"),started=Path.of(output+".started.json");
|
||||
assertFalse(Files.exists(output));assertFalse(Files.exists(started));assertFalse(Files.exists(progressFile));
|
||||
Files.writeString(reservation,"Reserved TEST candidate build acceptance\n",StandardOpenOption.CREATE_NEW);
|
||||
int skus=bounded("SELFOP_BUILD_SKUS",10000,10000,100000);
|
||||
int days=bounded("SELFOP_BUILD_DAYS",1,1,30);
|
||||
assertTrue((long)skus*days<=1000000,"At most one million synthetic orders");
|
||||
report.put("label",label);report.put("environment","TEST");report.put("synthetic",true);
|
||||
report.put("skuCount",skus);report.put("days",days);report.put("sourceHttpCalls",0);
|
||||
report.put("harnessConfiguredExternalRequestBudget",0);report.put("collectorAssembled",false);report.put("externalClientAssembled",false);
|
||||
report.put("publishCalled",false);report.put("harnessConfiguredCollectionMode","OFF");
|
||||
report.put("environmentCollectionModeAndAccountBudgetEvidence","ROOT_HTTP_BEFORE_AND_AFTER_REQUIRED");
|
||||
report.put("sourceHttpCallsEvidence","NO_EXTERNAL_CLIENT_ASSEMBLED_IN_HARNESS");
|
||||
report.put("transactions",commits);report.put("success",false);
|
||||
report.put("syntheticAccount",account);
|
||||
report.put("mappedStatementTimingMeaning","CLIENT_MYBATIS_EXECUTOR_CALL_INCLUDING_JDBC_AND_MAPPING_NOT_DATABASE_SERVER_TIME; BUILD_ONLY_EXCLUDES_SEED_AND_VERIFICATION");
|
||||
try {
|
||||
String host=required("CHONGCHONG_DB_HOST"),port=required("CHONGCHONG_DB_PORT");
|
||||
assertTrue(host.matches("[A-Za-z0-9.-]+")&&port.matches("[0-9]{1,5}"));
|
||||
var ds=new DriverManagerDataSource("jdbc:mysql://"+host+":"+port+"/baiyue_lastleg?useSSL=false&allowPublicKeyRetrieval=true&connectionTimeZone=UTC&forceConnectionTimeZoneToSession=true&characterEncoding=UTF-8&rewriteBatchedStatements=true&connectTimeout=10000&socketTimeout=120000",
|
||||
required("CHONGCHONG_DB_USERNAME"),required("CHONGCHONG_DB_PASSWORD"));
|
||||
ds.setDriverClassName("com.mysql.cj.jdbc.Driver");
|
||||
try(var c=ds.getConnection()){assertEquals("MySQL",c.getMetaData().getDatabaseProductName());assertEquals(8,c.getMetaData().getDatabaseMajorVersion());}
|
||||
jdbc=new JdbcTemplate(ds);var manager=new DataSourceTransactionManager(ds);transactions=new TransactionTemplate(manager);
|
||||
var config=new MybatisConfiguration();config.setMapUnderscoreToCamelCase(true);config.setLogImpl(NoLoggingImpl.class);config.addInterceptor(transactionEvidence);
|
||||
for(var type:List.of(OutboundLockMapper.class,OutboundAggregateMapper.class,SelfopSyncScopeMapper.class,SelfopSyncTaskMapper.class,
|
||||
SelfopSyncRunMapper.class,SelfopSyncStageMapper.class,SelfopOutboundOrderMapper.class,SelfopOutboundItemMapper.class,
|
||||
SelfopSkuDailySalesMapper.class,SelfopSkuProfileMapper.class))config.addMapper(type);
|
||||
var factory=new MybatisSqlSessionFactoryBean();factory.setDataSource(ds);factory.setConfiguration(config);
|
||||
var session=new SqlSessionTemplate(Objects.requireNonNull(factory.getObject()));
|
||||
var locks=session.getMapper(OutboundLockMapper.class);var tasks=session.getMapper(SelfopSyncTaskMapper.class);
|
||||
var runs=session.getMapper(SelfopSyncRunMapper.class);var scopes=session.getMapper(SelfopSyncScopeMapper.class);
|
||||
var facts=new OutboundFactService(locks,tasks,runs,scopes,session.getMapper(SelfopSyncStageMapper.class),
|
||||
session.getMapper(SelfopOutboundOrderMapper.class),session.getMapper(SelfopOutboundItemMapper.class));
|
||||
var properties=new SelfopAnalyticsProperties();assertEquals("OFF",properties.getCollectionMode());assertFalse(properties.isExecutionEnabled());
|
||||
// Lease remains bounded; no sleeps or artificial delays are used to manufacture build duration.
|
||||
properties.setLeaseSeconds(600);
|
||||
var target=new OutboundCandidateTransaction(locks,facts,runs,tasks,scopes,session.getMapper(SelfopSkuDailySalesMapper.class),
|
||||
session.getMapper(SelfopSkuProfileMapper.class),session.getMapper(OutboundAggregateMapper.class),properties);
|
||||
var proxy=new ProxyFactory(target);proxy.setProxyTargetClass(true);
|
||||
proxy.addAdvice((MethodInterceptor)invocation->{
|
||||
Instant start=Instant.now();long nanos=System.nanoTime();Object result=invocation.proceed();
|
||||
assertFalse(TransactionSynchronizationManager.isActualTransactionActive(),"Facade must observe committed method transaction");
|
||||
Map<String,Object> entry=new LinkedHashMap<>();entry.put("method",invocation.getMethod().getName());
|
||||
entry.put("startedAt",start.toString());entry.put("committedAt",Instant.now().toString());entry.put("elapsedMs",(System.nanoTime()-nanos)/1000000);
|
||||
if(result instanceof OutboundCandidateTransaction.Checkpoint cp){entry.put("phase",cp.phase());entry.put("dailyRows",cp.dailyRows());entry.put("profileRows",cp.profileRows());entry.put("committedBatches",cp.committedBatches());}
|
||||
commits.add(entry);report.put("lastCommittedMethod",invocation.getMethod().getName());
|
||||
report.put("phase",result instanceof OutboundCandidateTransaction.Checkpoint cp?cp.phase():"DONE");
|
||||
writeProgress();return result;
|
||||
});
|
||||
proxy.addAdvice(new TransactionInterceptor(manager,new AnnotationTransactionAttributeSource()));
|
||||
report.put("seedStart",Instant.now().toString());seed(skus,days);report.put("seedFinish",Instant.now().toString());
|
||||
Object pointerBefore=jdbc.queryForMap("SELECT published_run_id,published_fact_revision FROM ll_selfop_sync_scope WHERE id=?",scopeId);
|
||||
report.put("seedCompletedAt",Instant.now().toString());
|
||||
Files.writeString(started,jsonMapper.writeValueAsString(Map.of("label",label,"seedComplete",true,"at",Instant.now().toString())),StandardOpenOption.CREATE_NEW);
|
||||
Instant start=Instant.now();report.put("buildStartedAt",start.toString());
|
||||
report.put("phase","BUILD_START");writeProgress();
|
||||
transactionEvidence.active=true;
|
||||
try {new OutboundCandidateBuilder((OutboundCandidateTransaction)proxy.getProxy()).build(scopeId,runId,owner,1L);}
|
||||
finally {transactionEvidence.active=false;captureStatementMetrics();}
|
||||
report.put("buildFinishedAt",Instant.now().toString());report.put("buildElapsedMs",Duration.between(start,Instant.now()).toMillis());
|
||||
var saved=runs.selectById(runId);assertEquals("READY",saved.getStatus());assertNull(saved.getPublishedAtUtc());assertEquals("COMPLETE",saved.getCoverageStatus());
|
||||
assertNotNull(saved.getCompleteDayFrom());assertNotNull(saved.getCompleteDayUntil());
|
||||
assertEquals(days,java.time.temporal.ChronoUnit.DAYS.between(saved.getCompleteDayFrom(),saved.getCompleteDayUntil()));
|
||||
var progress=read(tasks.selectById(taskId).getProgressJson(),Map.class);
|
||||
var cp=read(json(progress.get("candidateBuild")),OutboundCandidateTransaction.Checkpoint.class);
|
||||
assertEquals("DONE",cp.phase());assertEquals((long)skus*days,cp.dailyRows());assertEquals(skus,cp.profileRows());
|
||||
assertEquals(cp.dailyRows(),count("ll_selfop_sku_daily_sales","sales_run_id",runId));
|
||||
assertEquals(cp.profileRows(),count("ll_selfop_sku_profile","profile_run_id",runId));
|
||||
assertEquals((skus*(long)days+499)/500+(skus+499)/500,cp.committedBatches());
|
||||
assertEquals(pointerBefore,jdbc.queryForMap("SELECT published_run_id,published_fact_revision FROM ll_selfop_sync_scope WHERE id=?",scopeId));
|
||||
assertEquals(0L,jdbc.queryForObject("SELECT COUNT(*) FROM ll_selfop_outbound_discovery_scope WHERE source_account=?",Long.class,account));
|
||||
assertTrue(transactionEvidence.inserts>0);assertFalse(transactionEvidence.nonTransactional);
|
||||
BigDecimal quantity=jdbc.queryForObject("SELECT SUM(outbound_quantity) FROM ll_selfop_sku_daily_sales WHERE sales_run_id=?",BigDecimal.class,runId);
|
||||
assertEquals(0,BigDecimal.valueOf((long)skus*days).compareTo(quantity));
|
||||
report.put("dailyRows",cp.dailyRows());report.put("profileRows",cp.profileRows());report.put("committedBatches",cp.committedBatches());
|
||||
report.put("aggregateInsertStatements",transactionEvidence.inserts);report.put("allAggregateWritesTransactional",true);
|
||||
report.put("publishedPointerUnchanged",true);report.put("publishedPointerInitiallyNull",true);report.put("candidateStatus",saved.getStatus());report.put("success",true);
|
||||
} catch(Throwable failure) {
|
||||
report.put("failureType",failure.getClass().getSimpleName()); // Never serialize SQL exceptions or credentials.
|
||||
for(Throwable cause=failure;cause!=null;cause=cause.getCause()) {
|
||||
if(cause instanceof java.sql.SQLException sqlFailure) {
|
||||
String state=sqlFailure.getSQLState();
|
||||
if(state!=null&&state.matches("[A-Z0-9]{5}"))report.put("sqlState",state);
|
||||
report.put("sqlVendorCode",sqlFailure.getErrorCode());break;
|
||||
}
|
||||
}
|
||||
throw new AssertionError("Candidate build harness failed: "+failure.getClass().getSimpleName());
|
||||
} finally {
|
||||
captureStatementMetrics();
|
||||
if(jdbc!=null&&taskId!=0) {
|
||||
try { transactions.executeWithoutResult(s->{
|
||||
jdbc.update("UPDATE ll_selfop_sync_task SET status='DEAD',next_retry_at_utc=NULL WHERE id=? AND source_account=?",taskId,account);
|
||||
jdbc.update("UPDATE ll_selfop_sync_scope SET lease_owner=NULL,lease_until_utc=NULL WHERE id=? AND source_account=? AND lease_owner=?",scopeId,account,owner);
|
||||
});report.put("draftDisabled",true); } catch(Exception ignored){report.put("draftDisabled",false);report.put("success",false);}
|
||||
}
|
||||
report.put("harnessState",Boolean.TRUE.equals(report.get("success"))?"FINISHED":"FAILED");writeProgress();
|
||||
Files.writeString(output,jsonMapper.writerWithDefaultPrettyPrinter().writeValueAsString(report),StandardOpenOption.CREATE_NEW);
|
||||
assertEquals(true,report.get("success"),"Inspect sanitized candidate build report");
|
||||
}
|
||||
}
|
||||
|
||||
private void seed(int skus,int days) {
|
||||
scopeId=IdWorker.getId();taskId=IdWorker.getId();runId=IdWorker.getId();originTask=IdWorker.getId();originRun=IdWorker.getId();
|
||||
report.put("scopeId",Long.toString(scopeId));report.put("taskId",Long.toString(taskId));report.put("runId",Long.toString(runId));
|
||||
LocalDate until=LocalDate.now(ZoneId.of("Asia/Shanghai")),from=until.minusDays(days);
|
||||
LocalDateTime now=utcNow().minusSeconds(1);String scopeJson=json(Map.of("quantityUnit","PCS","businessTimezone","Asia/Shanghai","sourceTimezone","Asia/Shanghai","fixtureOrigin","SYNTHETIC_BUILD_ACCEPTANCE"));String hash=hash(scopeJson);
|
||||
transactions.executeWithoutResult(s->{
|
||||
insert("ll_selfop_sync_scope",fields("id",scopeId,"created_at",now,"updated_at",now,"source_account",account,"warehouse_code",warehouse,"data_family","OUTBOUND","scope_json",scopeJson,"scope_hash",hash,"business_timezone","Asia/Shanghai","source_timezone","Asia/Shanghai","schedule_cron","0 0 0 1 1 *","stale_after_seconds",93600,"review_request_budget",1,"review_time_budget_seconds",1,"review_target_days",7,"fact_revision",1,"published_fact_revision",0,"fence_token",1,"lease_owner",owner,"lease_until_utc",utcNow().plusHours(1)));
|
||||
for(long id:List.of(originTask,taskId))insert("ll_selfop_sync_task",fields("id",id,"created_at",now,"updated_at",now,"task_no","SYNTH-BUILD-"+id,"execution_purpose","BUSINESS","source_account",account,"warehouse_code",warehouse,"task_role","WAREHOUSE_SYNC","scope_id",scopeId,"request_key","build-"+id,"request_hash",hash,"task_type","REBUILD","query_scope_json",scopeJson,"scope_hash",hash,"status",id==taskId?"RUNNING":"SUCCEEDED","apply_state","APPLIED","attempt_count",1,"query_from_utc",utc(from),"query_until_utc",utc(until),"routing_manifest_json","{\"directoryComplete\":true}"));
|
||||
for(long id:List.of(originRun,runId))insert("ll_selfop_sync_run",fields("id",id,"created_at",now,"updated_at",now,"task_id",id==runId?taskId:originTask,"execution_purpose","BUSINESS","scope_id",scopeId,"source_account",account,"warehouse_code",warehouse,"run_role","WAREHOUSE_SYNC","attempt_no",1,"scope_hash",hash,"scope_snapshot_json",scopeJson,"fence_token",1,"status","FACTS_APPLIED","last_phase","BUILD","coverage_status","UNKNOWN"));
|
||||
});
|
||||
for(int offset=0;offset<skus*days;offset+=500) {
|
||||
List<Map<String,Object>> orders=new ArrayList<>(),items=new ArrayList<>(),stages=new ArrayList<>();
|
||||
for(int n=offset;n<Math.min(offset+500,skus*days);n++) {
|
||||
String sku=String.format(Locale.ROOT,"SKU-%06d",n%skus),customer=String.format(Locale.ROOT,"OWNER-%03d",n%skus%100),delivery="ORDER-"+n;
|
||||
LocalDate date=from.plusDays(n/skus);long order=IdWorker.getId(),stage=IdWorker.getId();
|
||||
var payload=new SelfopPayloads.Order(account,customer,warehouse,delivery,null,100,null,date+" 12:00:00",List.of(new SelfopPayloads.Item(sku,"Synthetic candidate build",BigDecimal.ONE,"PCS")));
|
||||
String content=json(payload),digest=hash(content);
|
||||
orders.add(fields("id",order,"created_at",now,"updated_at",now,"source_account",account,"customer_code",customer,"warehouse_code",warehouse,"delivery_no",delivery,"source_status",100,"source_outbound_time",payload.sourceOutboundTime(),"source_timezone","Asia/Shanghai","outbound_at_utc",utc(date).plusHours(12),"business_date",date,"business_timezone","Asia/Shanghai","sales_eligible",1,"revision_no",1,"content_hash",digest,"last_run_id",originRun,"last_observation_sequence",stage));
|
||||
items.add(fields("id",IdWorker.getId(),"created_at",now,"updated_at",now,"order_id",order,"sku_code",sku,"product_name","Synthetic candidate build","actual_quantity",BigDecimal.ONE,"quantity_unit","PCS"));
|
||||
stages.add(fields("id",stage,"created_at",now,"updated_at",now,"run_id",originRun,"stage_batch_id",originTask,"last_fetch_run_id",originRun,"observation_sequence",stage,"warehouse_code",warehouse,"object_type","OUTBOUND_ORDER","object_key_hash",hash(account+"|"+customer+"|"+delivery),"first_page_no",1,"observed_at_utc",now,"content_hash",digest,"payload_json",content,"validation_status","VALID"));
|
||||
}
|
||||
transactions.executeWithoutResult(s->{batch("ll_selfop_sync_stage",stages);batch("ll_selfop_outbound_order",orders);batch("ll_selfop_outbound_item",items);});
|
||||
}
|
||||
// Seeding can take time; acquire a fresh fixture-only lease immediately before the real build.
|
||||
jdbc.update("UPDATE ll_selfop_sync_scope SET lease_until_utc=? WHERE id=? AND source_account=? AND lease_owner=?",utcNow().plusMinutes(10),scopeId,account,owner);
|
||||
}
|
||||
private void insert(String table,Map<String,Object> row){jdbc.update(insertSql(table,row),row.values().toArray());}
|
||||
private void batch(String table,List<Map<String,Object>> rows){jdbc.batchUpdate(insertSql(table,rows.get(0)),rows.stream().map(r->r.values().toArray()).toList());}
|
||||
private static String insertSql(String table,Map<String,Object> row){return "INSERT INTO "+table+" ("+String.join(",",row.keySet())+") VALUES ("+String.join(",",Collections.nCopies(row.size(),"?"))+")";}
|
||||
private static Map<String,Object> fields(Object... pairs){var result=new LinkedHashMap<String,Object>();for(int i=0;i<pairs.length;i+=2)result.put((String)pairs[i],pairs[i+1]);return result;}
|
||||
private long count(String table,String field,long id){return jdbc.queryForObject("SELECT COUNT(*) FROM "+table+" WHERE "+field+"=?",Long.class,id);}
|
||||
private static LocalDateTime utc(LocalDate date){return date.atStartOfDay(ZoneId.of("Asia/Shanghai")).withZoneSameInstant(ZoneOffset.UTC).toLocalDateTime();}
|
||||
private static String required(String name){String v=System.getenv(name);if(v==null||v.isBlank())throw new IllegalStateException("Missing TEST harness variable: "+name);return v;}
|
||||
private static int bounded(String key,int fallback,int min,int max){String value=System.getenv(key);int n=value==null?fallback:Integer.parseInt(value);assertTrue(n>=min&&n<=max,key+" outside bounded range");return n;}
|
||||
private void captureStatementMetrics(){
|
||||
report.put("buildMappedStatementCalls",transactionEvidence.calls);
|
||||
report.put("buildMappedStatements",transactionEvidence.snapshot());
|
||||
}
|
||||
private void writeProgress()throws java.io.IOException {
|
||||
captureStatementMetrics();report.put("progressUpdatedAt",Instant.now().toString());
|
||||
Path temporary=Path.of(progressFile+".tmp");
|
||||
Files.writeString(temporary,jsonMapper.writerWithDefaultPrettyPrinter().writeValueAsString(report),StandardOpenOption.CREATE_NEW);
|
||||
try {Files.move(temporary,progressFile,StandardCopyOption.ATOMIC_MOVE,StandardCopyOption.REPLACE_EXISTING);}
|
||||
catch(AtomicMoveNotSupportedException unsupported){Files.move(temporary,progressFile,StandardCopyOption.REPLACE_EXISTING);}
|
||||
}
|
||||
@Intercepts({@Signature(type=Executor.class,method="update",args={MappedStatement.class,Object.class}),
|
||||
@Signature(type=Executor.class,method="query",args={MappedStatement.class,Object.class,org.apache.ibatis.session.RowBounds.class,org.apache.ibatis.session.ResultHandler.class})})
|
||||
public static class TransactionEvidence implements Interceptor {
|
||||
long inserts,calls;boolean nonTransactional,active;
|
||||
private final Map<String,long[]> metrics=new LinkedHashMap<>();
|
||||
Map<String,Object> snapshot(){
|
||||
Map<String,Object> result=new LinkedHashMap<>();metrics.forEach((id,m)->result.put(id,Map.of("calls",m[0],"failedCalls",m[1],"totalClientElapsedMs",m[2]/1000000.0,"maxClientElapsedMs",m[3]/1000000.0)));
|
||||
return result;
|
||||
}
|
||||
@Override public Object intercept(Invocation invocation)throws Throwable {
|
||||
if(!active)return invocation.proceed();
|
||||
String id=((MappedStatement)invocation.getArgs()[0]).getId();
|
||||
if(id.equals(OutboundAggregateMapper.class.getName()+".insertDaily")||id.equals(OutboundAggregateMapper.class.getName()+".insertProfiles")) {
|
||||
inserts++;nonTransactional|=!TransactionSynchronizationManager.isActualTransactionActive();
|
||||
}
|
||||
long start=System.nanoTime();boolean succeeded=false;calls++;
|
||||
try {Object value=invocation.proceed();succeeded=true;return value;}
|
||||
finally {long elapsed=System.nanoTime()-start;long[] metric=metrics.computeIfAbsent(id,key->new long[4]);
|
||||
metric[0]++;if(!succeeded)metric[1]++;metric[2]+=elapsed;metric[3]=Math.max(metric[3],elapsed);}
|
||||
}
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user