// Explicit TEST-only row-lock fault; importing this file performs no I/O. import fs from 'node:fs' import { createHash } from 'node:crypto' import { resolve } from 'node:path' import { pathToFileURL } from 'node:url' import { parseCrossCustomerArguments, runCrossCustomer } from './fleet-cross-customer-test-env.mjs' import { verifyFleetCrossCustomer } from './admin-test-env-modules/fleet-cross-customer-acceptance.mjs' const demand=(v,message)=>{if(!v)throw Error(message)} const literal=v=>`CONVERT(0x${Buffer.from(v).toString('hex')} USING utf8mb4) COLLATE utf8mb4_bin` const encoded=v=>'['+Buffer.from(v).toString('base64').match(/.{1,4000}/g).map(x=>`'${x}'`).join(',')+"].join('')" const id=v=>String(v??'') const rr=['SET NAMES utf8mb4 COLLATE utf8mb4_bin;','SET SESSION TRANSACTION ISOLATION LEVEL REPEATABLE READ;','START TRANSACTION WITH CONSISTENT SNAPSHOT, READ ONLY;'] const pause=ms=>new Promise(done=>setTimeout(done,ms)) const canonical=v=>JSON.stringify(sortFields(v)) function sortFields(v){return Array.isArray(v)?v.map(sortFields):v&&typeof v==='object'?Object.fromEntries(Object.keys(v).sort().map(k=>[k,sortFields(v[k])])):v} export function validateRollbackScope(s) { demand(s&&/^[A-Za-z0-9_-]{1,64}$/.test(s.marker||'')&&/^[A-Za-z0-9_-]{1,64}$/.test(s.tripNo||''),'ROLLBACK_SCOPE_REQUIRED') for(const key of ['firstCustomerId','secondCustomerId','firstDispatchId','secondDispatchId','firstOutboundId','secondPalletId']) demand(typeof s[key]==='string'&&/^[1-9][0-9]{0,19}$/.test(s[key]),'ROLLBACK_IDENTIFIER_INVALID') demand(s.firstCustomerId!==s.secondCustomerId&&s.firstDispatchId!==s.secondDispatchId&&/^sha256:[a-f0-9]{64}$/.test(s.expectedImage||''),'ROLLBACK_DISTINCT_CUSTOMERS_AND_RUNTIME_REQUIRED') demand(/^[A-Za-z0-9_-]{1,64}$/.test(s.firstDispatchNo||''),'ROLLBACK_DISPATCH_NUMBER_INVALID') return s } export function rollbackSql(scope,mode) { const s=validateRollbackScope(scope),second=`customer_id=${literal(s.secondCustomerId)} AND dispatch_order_id=${s.secondDispatchId} AND warehouse_pallet_id=${s.secondPalletId} AND status='ACTIVE'` if(mode==='LEASE')return `SET SESSION innodb_lock_wait_timeout=5;START TRANSACTION;SELECT JSON_OBJECT('lockId',CAST(id AS CHAR),'customerId',customer_id,'connectionId',CONNECTION_ID()) FROM ll_dispatch_pallet_lock WHERE ${second} FOR UPDATE;SELECT JSON_OBJECT('marker','ROLLBACK_${s.marker}','elapsed',SLEEP(65));ROLLBACK;SELECT JSON_OBJECT('released',JSON_EXTRACT('true','$'));` const lease=`SELECT JSON_OBJECT('collection','lease','connectionId',t.trx_mysql_thread_id,'lockId',l.LOCK_DATA) FROM performance_schema.data_locks l JOIN information_schema.innodb_trx t ON t.trx_id=l.ENGINE_TRANSACTION_ID WHERE l.OBJECT_SCHEMA=DATABASE() AND l.OBJECT_NAME='ll_dispatch_pallet_lock' AND l.INDEX_NAME='PRIMARY' AND l.LOCK_STATUS='GRANTED' AND l.LOCK_DATA IN (SELECT CAST(id AS CHAR) FROM ll_dispatch_pallet_lock WHERE ${second}) AND t.trx_query LIKE ${literal('%ROLLBACK_'+s.marker+'%')};` const waits=`SELECT JSON_OBJECT('collection','wait','requester',r.trx_mysql_thread_id,'blocker',b.trx_mysql_thread_id,'lockId',l.LOCK_DATA) FROM performance_schema.data_lock_waits w JOIN performance_schema.data_locks l ON l.ENGINE_LOCK_ID=w.REQUESTING_ENGINE_LOCK_ID JOIN information_schema.innodb_trx r ON r.trx_id=w.REQUESTING_ENGINE_TRANSACTION_ID JOIN information_schema.innodb_trx b ON b.trx_id=w.BLOCKING_ENGINE_TRANSACTION_ID WHERE l.OBJECT_SCHEMA=DATABASE() AND l.OBJECT_NAME='ll_dispatch_pallet_lock' AND l.INDEX_NAME='PRIMARY' AND l.LOCK_DATA IN (SELECT CAST(id AS CHAR) FROM ll_dispatch_pallet_lock WHERE ${second});` if(mode==='LOCKS')return [...rr,lease,waits,'COMMIT;'].join('\n') if(mode==='OBSERVE')return ['SET NAMES utf8mb4 COLLATE utf8mb4_bin;','SET SESSION TRANSACTION ISOLATION LEVEL READ UNCOMMITTED;','START TRANSACTION READ ONLY;',lease,waits, `SELECT JSON_OBJECT('collection','firstDispatch','id',CAST(id AS CHAR),'status',status,'shipped',shipped_package_count) FROM ll_dispatch_order WHERE id=${s.firstDispatchId} AND customer_id=${literal(s.firstCustomerId)};`, `SELECT JSON_OBJECT('collection','firstOutbound','id',CAST(id AS CHAR),'status',status) FROM ll_outbound_order WHERE id=${s.firstOutboundId} AND customer_id=${literal(s.firstCustomerId)};`, `SELECT JSON_OBJECT('collection','firstReplay','count',COUNT(*)) FROM ll_dispatch_pda_operation WHERE customer_id=${literal(s.firstCustomerId)} AND operation_type='DEPART' AND idempotency_key=${literal(s.tripNo+':'+s.firstDispatchNo)};`,'COMMIT;'].join('\n') if(mode==='RELEASED')return [...rr,lease,waits,'COMMIT;'].join('\n') throw Error('ROLLBACK_MODE_INVALID') } export function buildRollbackRemoteScript(scope,mode,agent) { demand(['RUNTIME','LEASE','LOCKS','OBSERVE','RELEASED'].includes(mode)&&/^[A-Za-z0-9_.-]{1,128}$/.test(agent||''),'ROLLBACK_REMOTE_MODE_REQUIRED') const sql=mode==='RUNTIME'?"SELECT JSON_OBJECT('lockWaitSeconds',@@GLOBAL.innodb_lock_wait_timeout);":rollbackSql(scope,mode) const imageGuard=mode==='RUNTIME'?'':` or c['Image']!='${scope.expectedImage}'` const python=`import json,os,subprocess,urllib.parse,re try: c=json.loads(subprocess.check_output(['docker','inspect','firstleg-test-lastleg'],text=True,stderr=subprocess.DEVNULL))[0] e=dict(x.split('=',1) for x in c['Config']['Env']) if c['Name']!='/firstleg-test-lastleg' or c['Config'].get('Labels',{}).get('com.docker.compose.project')!='firstleg-test' or e.get('DEPLOY_ENV')!='test' or c['State'].get('Health',{}).get('Status')!='healthy'${imageGuard}:raise RuntimeError('TEST_RUNTIME_CHANGED') u=urllib.parse.urlsplit(e['LASTLEG_DB_URL'].removeprefix('jdbc:'));host='127.0.0.1' if u.hostname=='host.docker.internal' else u.hostname q=subprocess.run(['mysql','--host='+host,'--port='+str(u.port or 3306),'--user='+e['LASTLEG_DB_USERNAME'],'--batch','--raw','--skip-column-names',u.path.lstrip('/')],input=${JSON.stringify(sql)},capture_output=True,text=True,env=dict(os.environ,MYSQL_PWD=e['LASTLEG_DB_PASSWORD']),timeout=${mode==='LEASE'?75:25}) if q.returncode: code=re.search(r'ERROR ([0-9]{3,5})',q.stderr);print('TRIP_ROLLBACK_JSON='+json.dumps({'ok':False,'error':'MYSQL_'+(code.group(1) if code else 'UNKNOWN')})) else:print('TRIP_ROLLBACK_JSON='+json.dumps({'ok':True,'imageId':c['Image'],'rows':[json.loads(x) for x in q.stdout.splitlines() if x.strip()]})) except Exception as error: print('TRIP_ROLLBACK_JSON='+json.dumps({'ok':False,'error':str(error) if re.fullmatch('[A-Z_]{1,64}',str(error)) else 'REMOTE_ROLLBACK_FAILED'})) ` const groovy=`def p=new ProcessBuilder(['python3','-c',new String(java.util.Base64.decoder.decode(${encoded(python)}),'UTF-8')]).redirectErrorStream(true).start();def out=p.inputStream.getText('UTF-8');p.waitFor();print(out)` return `def c=jenkins.model.Jenkins.instance.getComputer('${agent}');if(c==null||c.channel==null)throw new RuntimeException('TEST agent unavailable');print(hudson.util.RemotingDiagnostics.executeGroovy(new String(java.util.Base64.decoder.decode(${encoded(groovy)}),'UTF-8'),c.channel))` } export function assertMidTransactionObservation(rows) { const one=name=>rows.filter(x=>x.collection===name) const lease=one('lease'),wait=one('wait'),dispatch=one('firstDispatch'),outbound=one('firstOutbound'),replay=one('firstReplay') demand(lease.length===1&&wait.length===1&&wait[0].blocker===lease[0].connectionId&&wait[0].lockId===lease[0].lockId&&wait[0].requester!==wait[0].blocker,'OWNED_LATE_ROW_WAIT_NOT_PROVED') demand(dispatch.length===1&&dispatch[0].status==='DEPARTED'&&dispatch[0].shipped===1&&outbound.length===1&&outbound[0].status==='OUTBOUNDED'&&replay.length===1&&replay[0].count===1,'FIRST_CUSTOMER_UNCOMMITTED_WRITE_NOT_PROVED') return {leaseConnectionId:lease[0].connectionId,requestingConnectionId:wait[0].requester,lockId:lease[0].lockId,dirtyObservationOnly:true} } export function requestRollbackDeparture(context,path,body,token) { // The normal JSON request stops at 30 seconds. This deliberate 50-second // database timeout needs the existing 60-second, single-attempt request path. return context.requestWithBody('POST',path,body,token) } export function assertDatabaseLockConflict(response) { demand(response?.http===200&&String(response.body?.code)==='A0409' &&response.body?.message==='资源正在处理中,请稍后重试' &&response.requestId&&response.traceId,'EXPECTED_DATABASE_LOCK_CONFLICT_REQUIRED') } async function remote(env,scope,mode) { const script=buildRollbackRemoteScript(scope,mode,env.FBA_CORE_JENKINS_AGENT) const response=await fetch(new URL('scriptText',env.JENKINS_URL.replace(/\/?$/,'/')),{method:'POST',redirect:'error',signal:AbortSignal.timeout(mode==='LEASE'?95000:45000),headers:{Authorization:'Basic '+Buffer.from(env.JENKINS_USER+':'+env.JENKINS_API_TOKEN).toString('base64'),'Content-Type':'application/x-www-form-urlencoded'},body:new URLSearchParams({script})}) demand(response.ok,'ROLLBACK_EVIDENCE_HTTP_FAILED') const lines=(await response.text()).split(/\r?\n/).filter(x=>x.startsWith('TRIP_ROLLBACK_JSON='));demand(lines.length===1,'ROLLBACK_REMOTE_RESULT_AMBIGUOUS') const result=JSON.parse(lines[0].slice('TRIP_ROLLBACK_JSON='.length));demand(result.ok===true,/^[A-Z_0-9]+$/.test(result.error||'')?result.error:'ROLLBACK_REMOTE_FAILED') return result } export function createDepartureRollbackVerifier(env,config,runtime) { return async({context,fixture,ready,departPath,departBody,snapshot})=>{ const output=resolve('reports',`fleet-cross-${config.runId}`,'rollback.json'),fd=fs.openSync(output,'wx',0o600) const report={verified:false,scope:'OWNED_TWO_CUSTOMER_DEPARTURE_ROLLBACK',startedAt:new Date().toISOString(),samples:[],command:{path:departPath,body:departBody},runtime} const save=()=>{fs.ftruncateSync(fd,0);fs.writeSync(fd,JSON.stringify(report,null,2),0,'utf8');fs.fsyncSync(fd)} let leasePromise,pending try { demand(ready.trip.dispatches?.length===2&&fixture.members.length===2,'TWO_CUSTOMER_READY_SCOPE_REQUIRED') const [first,second]=ready.trip.dispatches.map(binding=>fixture.members.find(x=>x.dispatchNo===binding.dispatchNo)) demand(first&&second&&first.customerId!==second.customerId&&first.dispatchNo===fixture.members[0].dispatchNo,'CURRENT_DEPARTURE_ORDER_REQUIRED') const firstState=ready.members[fixture.members.indexOf(first)],secondState=ready.members[fixture.members.indexOf(second)] const scope=report.target=validateRollbackScope({marker:context.marker,tripNo:fixture.tripNo,expectedImage:runtime.imageId, firstCustomerId:first.customerId,secondCustomerId:second.customerId,firstDispatchId:id(firstState.dispatch.id),secondDispatchId:id(secondState.dispatch.id), firstOutboundId:id(firstState.outbound.outbound.id),secondPalletId:second.palletId,firstDispatchNo:first.dispatchNo}) save();leasePromise=remote(env,scope,'LEASE').then(value=>({value}),error=>({error:error.message})) for(let i=0;i<12;i++) { const rows=(await remote(env,scope,'LOCKS')).rows;report.samples.push({phase:'LEASE',rows});save() if(rows.filter(x=>x.collection==='lease').length===1)break demand(i!==11,'LIVE_LEASE_NOT_OBSERVED');await pause(250) } pending=requestRollbackDeparture(context,departPath,departBody,fixture.pdaToken).then(value=>({value}),error=>({error:error.message})) for(let i=0;i<20;i++) { const rows=(await remote(env,scope,'OBSERVE')).rows;report.samples.push({phase:'MID_TRANSACTION',rows});save() if(rows.some(x=>x.collection==='wait')) {report.observation=assertMidTransactionObservation(rows);save();break} demand(i!==19,'MID_TRANSACTION_NOT_OBSERVED');await pause(250) } report.response=await pending;save() report.lease=await leasePromise;save() demand(!report.lease.error&&report.lease.value.rows.at(-1)?.released===true,'LEASE_RELEASE_NOT_PROVED') const released=report.released=(await remote(env,scope,'RELEASED')).rows demand(released.length===0,'OWNED_ROW_LOCK_OR_WAITER_REMAINS') const restored=await snapshot('ROLLBACK_AFTER_FAULT') report.consistentSnapshot={beforeSha256:createHash('sha256').update(canonical(ready)).digest('hex'),afterSha256:createHash('sha256').update(canonical(restored)).digest('hex')} demand(report.consistentSnapshot.beforeSha256===report.consistentSnapshot.afterSha256,'FULL_TWO_CUSTOMER_ROLLBACK_SNAPSHOT_CHANGED') demand(!report.response.error&&report.response.value?.http!==0,'DEPARTURE_OUTCOME_UNKNOWN_RETAIN_ORIGINAL_COMMAND') assertDatabaseLockConflict(report.response.value) // The caller also checks restoration before sending the same original command. report.verified=true;return report }catch(error){report.error=error.message;throw error} finally { if(pending&&!report.response)report.response=await pending if(leasePromise&&!report.lease)report.lease=await leasePromise report.finishedAt=new Date().toISOString();save();fs.closeSync(fd) } } } export async function runTripRollback(config,{env=process.env}={}) { if(!config.execute)return {...config,scope:'OWNED_TWO_CUSTOMER_DEPARTURE_ROLLBACK',fault:'65-second SELECT FOR UPDATE lease on the second customer pallet-lock row; first customer writes observed as uncommitted, default 50-second lock timeout rolls back the entire original request; then same original command and full normal chain',businessSqlWrites:false,deployment:false} demand(env.CHONGCHONG_ENV==='test'&&['JENKINS_URL','JENKINS_USER','JENKINS_API_TOKEN','FBA_CORE_JENKINS_AGENT'].every(key=>env[key]),'EXPLICIT_TEST_EVIDENCE_PROFILE_REQUIRED') const runtime=await remote(env,{},'RUNTIME');demand(runtime.rows.length===1&&runtime.rows[0].lockWaitSeconds===50,'DEFAULT_50_SECOND_LOCK_TIMEOUT_REQUIRED') const sources=['scripts/fleet-trip-rollback-test-env.mjs','scripts/admin-test-env-modules/fleet-cross-customer-acceptance.mjs'].map(path=>({path,sha256:createHash('sha256').update(fs.readFileSync(path)).digest('hex')})) return runCrossCustomer(config,{env,verify:async(context,fixture,options)=>{ fs.writeFileSync(resolve('reports',`fleet-cross-${config.runId}`,'rollback-source.json'),JSON.stringify({runtime,sources},null,2),{flag:'wx'}) return verifyFleetCrossCustomer(context,fixture,{...options,verifyDepartureRollback:createDepartureRollbackVerifier(env,config,runtime)}) }}) } if(process.argv[1]&&import.meta.url===pathToFileURL(resolve(process.argv[1])).href) { try {const config=parseCrossCustomerArguments(process.argv.slice(2)),result=await runTripRollback(config);console.log(JSON.stringify(config.execute?{success:result.success,error:result.error,summary:result.summary,requests:result.requestCount,goalComplete:false}:result,null,2));if(config.execute&&!result.success)process.exitCode=1} catch(error){console.error(/^[A-Z_0-9]+$/.test(error.message)?error.message:'ROLLBACK_ENTRY_FAILED');process.exitCode=1} }