diff --git a/deploy/sql/海外区域统一增量-MySQL8.sql b/deploy/sql/海外区域统一增量-MySQL8.sql index f55d4aed9..4a505e045 100644 --- a/deploy/sql/海外区域统一增量-MySQL8.sql +++ b/deploy/sql/海外区域统一增量-MySQL8.sql @@ -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='待验证事实与修订证据'; diff --git a/lastleg-service/src/main/resources/db/lastleg.sql b/lastleg-service/src/main/resources/db/lastleg.sql index fe085941e..7e8fb178a 100644 --- a/lastleg-service/src/main/resources/db/lastleg.sql +++ b/lastleg-service/src/main/resources/db/lastleg.sql @@ -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='待验证事实与修订证据'; diff --git a/lastleg-service/src/test/java/com/baiyue/lastleg/schema/SelfoperatedAnalyticsSchemaContractTest.java b/lastleg-service/src/test/java/com/baiyue/lastleg/schema/SelfoperatedAnalyticsSchemaContractTest.java index 9cb6904e9..3f2da3628 100644 --- a/lastleg-service/src/test/java/com/baiyue/lastleg/schema/SelfoperatedAnalyticsSchemaContractTest.java +++ b/lastleg-service/src/test/java/com/baiyue/lastleg/schema/SelfoperatedAnalyticsSchemaContractTest.java @@ -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")); diff --git a/lastleg-service/src/test/java/com/baiyue/lastleg/selfoperated/analytics/outbound/OutboundAggregateMapperTest.java b/lastleg-service/src/test/java/com/baiyue/lastleg/selfoperated/analytics/outbound/OutboundAggregateMapperTest.java index aade9be52..ea22d433f 100644 --- a/lastleg-service/src/test/java/com/baiyue/lastleg/selfoperated/analytics/outbound/OutboundAggregateMapperTest.java +++ b/lastleg-service/src/test/java/com/baiyue/lastleg/selfoperated/analytics/outbound/OutboundAggregateMapperTest.java @@ -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',?)"); diff --git a/lastleg-service/src/test/java/com/baiyue/lastleg/selfoperated/analytics/outbound/OutboundCandidateBuildMySqlAcceptanceTest.java b/lastleg-service/src/test/java/com/baiyue/lastleg/selfoperated/analytics/outbound/OutboundCandidateBuildMySqlAcceptanceTest.java new file mode 100644 index 000000000..01f7a3034 --- /dev/null +++ b/lastleg-service/src/test/java/com/baiyue/lastleg/selfoperated/analytics/outbound/OutboundCandidateBuildMySqlAcceptanceTest.java @@ -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 report=new LinkedHashMap<>(); + private final List> 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 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> orders=new ArrayList<>(),items=new ArrayList<>(),stages=new ArrayList<>(); + for(int n=offset;n{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 row){jdbc.update(insertSql(table,row),row.values().toArray());} + private void batch(String table,List> rows){jdbc.batchUpdate(insertSql(table,rows.get(0)),rows.stream().map(r->r.values().toArray()).toList());} + private static String insertSql(String table,Map row){return "INSERT INTO "+table+" ("+String.join(",",row.keySet())+") VALUES ("+String.join(",",Collections.nCopies(row.size(),"?"))+")";} + private static Map fields(Object... pairs){var result=new LinkedHashMap();for(int i=0;i=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 metrics=new LinkedHashMap<>(); + Map snapshot(){ + Map 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);} + } + } +}