Files
baiyue-firstleg/scripts/order-submit-dependency-fault-executor.mjs
T

252 lines
18 KiB
JavaScript

// Explicit TEST-only runtime controller. Importing this module performs no HTTP.
import { readFileSync } from 'node:fs'
import { open, mkdir } from 'node:fs/promises'
import { createHash } from 'node:crypto'
import { resolve, dirname } from 'node:path'
import { pathToFileURL } from 'node:url'
import { assertOrderDependencyFaultObservation, orderPersistenceFacts } from './admin-test-env-modules/orders.mjs'
const runtime = readFileSync(new URL('./order-submit-dependency-fault-runtime.py', import.meta.url), 'utf8')
const proxySource = readFileSync(new URL('./order-submit-dependency-fault-proxy.py', import.meta.url)).toString('base64')
const commonSource = readFileSync(new URL('./test-event-fault-runtime.py', import.meta.url)).toString('base64')
const encode = value => Buffer.from(value).toString('base64')
const sha = value => createHash('sha256').update(JSON.stringify(value)).digest('hex')
const assert = (ok, message) => { if (!ok) throw Error(message) }
const pause = ms => new Promise(resolve => setTimeout(resolve, ms))
const safeError = error => /^(?:[A-Z][A-Z_0-9]{1,100}|Dependency fault evidence: [A-Za-z /:-]+)$/.test(error?.message || '')
? error.message : 'ORDER_DEPENDENCY_VERIFICATION_FAILED'
export function scopeFor(plan) {
assert(/^[A-Za-z0-9_-]{1,80}$/.test(plan?.marker || '') && /^[1-9]\d{0,19}$/.test(plan?.customerId || ''), 'Invalid TEST plan scope')
for (const role of ['merchant', 'platform']) {
const command = plan.commands?.[role]
assert(command?.method === 'POST' && /^[1-9]\d{0,19}$/.test(command.orderId || '')
&& command.path === `/api/${role}/v1/orders/first-leg/${command.orderId}/submit${role === 'platform' ? `?tenantId=${plan.customerId}` : ''}`
&& Number.isSafeInteger(command.body?.expectedVersion) && command.body.expectedVersion >= 0
&& Object.keys(command.body).length === 1 && /^[A-Za-z0-9_-]{1,128}$/.test(command.headers?.['Idempotency-Key'] || '')
&& Object.keys(command.headers).length === 1
&& /^[0-9a-f]{64}$/.test(command.initialOrderSha256 || ''), 'Invalid frozen submission command')
}
assert(plan.commands.merchant.orderId !== plan.commands.platform.orderId, 'Two distinct new drafts required')
assert(plan.runtimeChangesExecuted === false, 'Only a reviewed unexecuted plan is accepted')
assert(Array.isArray(plan.phases) && plan.phases.length === 2, 'Exact two dependency phases required')
for (const [dependency, property, path, messages] of [
['IAM','baiyue.clients.iam.url',`/internal/iam/tenants/${plan.customerId}/order-eligibility`,['IAM eligibility service is unavailable','IAM 下单资格响应无效']],
['FINANCE','baiyue.clients.finance.url',`/internal/finance/v1/wallet/accounts/${plan.customerId}`,['Finance account service is unavailable','Finance account response is invalid']],
]) {
const rows = plan.phases.filter(row => row.dependency === dependency)
assert(rows.length === 1 && rows[0].urlProperty === property && rows[0].targetPath === path && rows[0].targetMethod === 'GET'
&& rows[0].faultHttp === 503 && rows[0].expectedHttp === 200 && rows[0].expectedCode === 'S0503'
&& JSON.stringify(rows[0].expectedMessages) === JSON.stringify(messages), 'Dependency failure contract may not be weakened')
}
return { marker: plan.marker, customerId: plan.customerId,
orderIds: Object.fromEntries(['merchant', 'platform'].map(role => [role, plan.commands[role].orderId])), planSha256: sha(plan) }
}
export function renderOrderFaultAction(action, scope, agent, extra = {}) {
assert(['preflight','prepare','activate','health','mode','status','records','restore','reconcile','verify-restored','finish'].includes(action), 'Unknown runtime action')
assert(/^[A-Za-z0-9_.-]{1,128}$/.test(agent || '') && /^[A-Za-z0-9_-]{1,80}$/.test(scope.marker), 'Invalid agent/scope')
if (['prepare','preflight'].includes(action)) {
const routing = extra.routing
assert(routing && Object.keys(routing).every(key => ['configurationReviewed','sourceSha256','upstreams','nacos'].includes(key))
&& routing.configurationReviewed === true && routing.nacos && Object.keys(routing.nacos).every(key => ['baseUrl','loginPath','configPath'].includes(key))
&& routing.upstreams && Object.keys(routing.upstreams).sort().join(',') === 'FINANCE,IAM', 'Reviewed routing schema only; do not embed credentials')
for (const address of [routing.nacos.baseUrl,...Object.values(routing.upstreams)]) {
const url = new URL(address)
assert(['http:','https:'].includes(url.protocol) && !url.username && !url.password && !url.search && !url.hash
&& ['', '/'].includes(url.pathname), 'Routing origins cannot contain credentials/path/query')
}
if (action === 'prepare') assert(/^sha256:[0-9a-f]{64}$/.test(extra.expectedImageId || '')
&& /^[0-9a-f]{64}$/.test(routing.sourceSha256 || ''), 'Reviewed image and Nacos source fingerprint required')
}
const data = encode(JSON.stringify({ ...extra, action, scope, commonSource, ...(action === 'prepare' ? { proxySource } : {}) }))
const expression = value => '[' + encode(value).match(/.{1,4000}/g).map(part => `'${part}'`).join(',') + "].join('')"
const groovy = `def p=new ProcessBuilder(['python3','-c',new String(java.util.Base64.decoder.decode(${expression(runtime)}),'UTF-8'),'${data}']).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(${expression(groovy)}),'UTF-8'),c.channel))`
}
export function createOrderFaultRuntime(env, scope, fetcher = globalThis.fetch) {
assert(env.CHONGCHONG_ENV === 'test', 'CHONGCHONG_ENV=test required')
assert(env.JENKINS_URL && env.JENKINS_USER && env.JENKINS_API_TOKEN && env.FBA_CORE_JENKINS_AGENT, 'Explicit TEST Jenkins credentials and agent required')
return async (action, extra = {}) => {
let response
try {
response = await fetcher(new URL('scriptText', env.JENKINS_URL.replace(/\/?$/, '/')), {
method: 'POST', redirect: 'error', signal: AbortSignal.timeout(55000),
headers: { Authorization: 'Basic ' + encode(env.JENKINS_USER + ':' + env.JENKINS_API_TOKEN), 'Content-Type': 'application/x-www-form-urlencoded' },
body: new URLSearchParams({ script: renderOrderFaultAction(action, scope, env.FBA_CORE_JENKINS_AGENT, extra) }),
})
} catch { throw Error(`RUNTIME_${action.toUpperCase().replaceAll('-', '_')}_OUTCOME_UNKNOWN_USE_STATUS`) }
assert(response.ok, `RUNTIME_HTTP_${response.status}_NO_AUTO_RETRY`)
const rows = (await response.text()).split(/\r?\n/).filter(row => row.startsWith('ORDER_DEPENDENCY_JSON='))
assert(rows.length === 1, 'RUNTIME_RESPONSE_AMBIGUOUS_NO_AUTO_RETRY')
const result = JSON.parse(rows[0].slice('ORDER_DEPENDENCY_JSON='.length))
assert(result?.ok === true, /^[A-Z_0-9]{1,90}$/.test(result?.error || '') ? result.error : 'RUNTIME_ACTION_FAILED')
return result.data
}
}
export async function runOrderFaultWindow({ plan, routing, expectedImageId, action, snapshot, submit, recovery, wait = pause }) {
const scope = scopeFor(plan), report = { scope, success: false, observations: [], restoration: null }
let prepareAttempted = false, failure
const health = async () => {
for (let i = 0; i < 180; i++) {
const value = await action('health')
if (value.healthy === true) return
await wait(1000)
}
throw Error('ORDER_HEALTH_TIMEOUT')
}
try {
for (const role of ['merchant','platform']) {
const before = await snapshot(role)
assert(createHash('sha256').update(orderPersistenceFacts(before.detail)).digest('hex') === plan.commands[role].initialOrderSha256
&& before.detail.order.status === 'DRAFT' && String(before.detail.order.tenantId) === scope.customerId
&& String(before.detail.order.id) === scope.orderIds[role]
&& ['cargoes','parties','internalNotes','statusHistory'].every(field => Array.isArray(before.detail[field]))
&& before.inboundPageComplete === true, 'PREPARED_DRAFT_CHANGED')
}
prepareAttempted = true
report.prepared = await action('prepare', { routing, expectedImageId })
for (let i = 0; i < 15; i++) {
if ((await action('status')).proxy?.mode === 'PASS') break
assert(i < 14, 'PROXY_STARTUP_NOT_READY'); await wait(1000)
}
await action('activate'); await health()
for (const dependency of ['IAM','FINANCE']) {
for (const role of ['merchant','platform']) {
const before = await snapshot(role), command = plan.commands[role]
const request = { method: command.method, path: command.path, body: structuredClone(command.body),
headers: { ...command.headers, 'X-Request-Id': `dep-${plan.marker}-${dependency}-${role}` } }
let response
await action('mode', { mode: dependency, ttl: 60 })
try { response = await submit(role, request) }
finally { await action('mode', { mode: 'PASS' }) }
const after = await snapshot(role)
const proxyRecords = (await action('records')).records.filter(row => row.dependency === dependency)
report.observations.push(assertOrderDependencyFaultObservation(plan, dependency, role, { request, response, before, after, proxyRecords }))
}
}
} catch (error) { failure = error }
finally {
if (prepareAttempted) {
try {
await action('restore') // Never retry a mutation after an ambiguous outcome.
await health()
report.restoredRuntime = await action('verify-restored')
assert(report.restoredRuntime.originalImageId === expectedImageId
&& report.restoredRuntime.currentImageId === expectedImageId
&& report.restoredRuntime.originalFingerprint === report.restoredRuntime.currentFingerprint
&& report.restoredRuntime.nacosSourcesUnchanged === true
&& report.restoredRuntime.upstreamReads?.IAM === true && report.restoredRuntime.upstreamReads?.FINANCE === true,
'RESTORED_IDENTITY_OR_ROUTES_NOT_PROVED')
const businessRecovery = await recovery()
report.businessRecovery = businessRecovery
report.restoration = await action('finish', { businessRecovery })
assert(report.restoration.restored && report.restoration.proxyStopped && report.restoration.orderRoutesVerified, 'RESTORATION_NOT_COMPLETE')
} catch (error) { report.recoveryError = safeError(error); failure ||= error }
}
}
report.success = !failure && report.observations.length === 4 && Boolean(report.restoration?.restored)
if (failure) report.error = safeError(failure)
return report
}
function businessClient(env, plan, fetcher = globalThis.fetch) {
const sessions = {}
const call = async (method, path, body, headers = {}, token) => {
const response = await fetcher(new URL(path, env.ADMIN_TEST_GATEWAY), { method, redirect:'error', signal:AbortSignal.timeout(20000),
headers:{'Content-Type':'application/json', ...headers, ...(token ? {Authorization:`Bearer ${token}`} : {})}, ...(body === undefined ? {} : {body:JSON.stringify(body)}) })
return {http:response.status,body:await response.json()}
}
const ok = value => { assert(value.http === 200 && String(value.body?.code) === '200', 'BUSINESS_READ_OR_RECOVERY_FAILED'); return value.body.data }
const snapshot = async role => {
const command = plan.commands[role]
const detailPath = `/api/${role}/v1/orders/first-leg/${command.orderId}${role === 'platform' ? `?tenantId=${plan.customerId}` : ''}`
const detail = ok(await call('GET',detailPath,undefined,{},sessions[role].accessToken))
assert(String(detail?.order?.id)===command.orderId && String(detail.order.tenantId)===plan.customerId
&& ['cargoes','parties','internalNotes','statusHistory'].every(field=>Array.isArray(detail[field])), 'ORDER_SNAPSHOT_INCOMPLETE_OR_FOREIGN')
const taskResponse = await call('GET',`/api/platform/v1/first-leg/tasks/by-order/${command.orderId}?tenantId=${plan.customerId}`,undefined,{},sessions.platform.accessToken)
assert(taskResponse.http === 200 && ['200','A0404'].includes(String(taskResponse.body?.code)), 'TASK_SNAPSHOT_FAILED')
const task = String(taskResponse.body.code) === '200' ? taskResponse.body.data ?? null : null
const inbounds = []; let total
for (let page=1;;page++) {
const value=ok(await call('GET',`/api/platform/inbounds?tenantId=${plan.customerId}&orderNo=${encodeURIComponent(command.orderNo)}&page=${page}&size=200`,undefined,{},sessions.platform.accessToken))
assert(Array.isArray(value.records) && Number(value.page)===page && Number.isSafeInteger(Number(value.total)), 'INBOUND_PAGE_INVALID')
total ??= Number(value.total); assert(Number(value.total)===total && value.records.length===Math.min(200,Math.max(0,total-inbounds.length)), 'INBOUND_PAGE_INCOMPLETE')
for(const row of value.records)assert(row.orderNo===command.orderNo && String(row.tenantId)===plan.customerId && !inbounds.some(before=>String(before.id)===String(row.id)), 'INBOUND_SCOPE_OR_DUPLICATE')
inbounds.push(...value.records);if(inbounds.length===total)break
}
return {detail,task,inbounds,inboundPageComplete:true}
}
const submit = (role, request) => call(request.method,request.path,request.body,request.headers,sessions[role].accessToken)
return { snapshot, submit,
async login() {
for(const [role,prefix] of [['platform','ADMIN_TEST'],['merchant','MERCHANT_TEST']]) {
assert(env[prefix+'_USERNAME'] && env[prefix+'_PASSWORD'], 'EXPLICIT_BUSINESS_CREDENTIALS_REQUIRED')
sessions[role]=ok(await call('POST','/api/auth/login',{loginId:env[prefix+'_USERNAME'],password:env[prefix+'_PASSWORD']}))
}
const profile=ok(await call('GET','/api/merchant/v1/iam/tenant/profile',undefined,{},sessions.merchant.accessToken))
assert(String(profile.id)===plan.customerId && profile.status==='ACTIVE','REGISTERED_CUSTOMER_SESSION_MISMATCH')
},
async recovery() {
const evidence={}
for(const role of ['merchant','platform']) {
const command=plan.commands[role]
ok(await submit(role,command))
let before=await snapshot(role)
for(let i=0;i<10;i++) { await pause(500);const next=await snapshot(role);if(sha(before)===sha(next)){before=next;break}before=next }
assert(before.detail.order.status==='SUBMITTED' && before.detail.statusHistory.filter(row=>row.action==='SUBMITTED').length===1
&& before.inbounds.length<=1, 'RECOVERY_SINGLE_SUBMISSION_NOT_PROVED')
ok(await submit(role,command));const after=await snapshot(role)
assert(sha(before)===sha(after),'RECOVERY_REPLAY_DRIFT')
evidence[role]={orderId:command.orderId,status:'SUBMITTED',sameKeyReplayVerified:true,snapshotSha256:sha(after)}
}
return evidence
},
async logout() {
let success=true
for(const session of Object.values(sessions))try {ok(await call('POST','/api/auth/logout',{refreshToken:session.refreshToken},{},session.accessToken))}catch{success=false}
return success
},
}
}
async function main(argv) {
const [mode,...rest]=argv, args={}
for(let i=0;i<rest.length;i+=2){assert(rest[i]?.startsWith('--') && rest[i+1] && !rest[i+1].startsWith('--'),'Use explicit --name value arguments');args[rest[i].slice(2)]=rest[i+1]}
assert(['plan','preflight','run','status','reconcile','restore'].includes(mode),'Use plan/preflight/run/status/reconcile/restore')
assert(args.plan,'--plan required');const plan=JSON.parse(readFileSync(args.plan,'utf8')),scope=scopeFor(plan)
if(mode==='plan'){console.log(JSON.stringify({scope,actions:['prepare','activate','IAM x 2','FINANCE x 2','finally restore','verify original routes','finish proxy'],requiresExclusiveOrderWindow:true},null,2));return}
assert(args['confirm-test-runtime']==='yes','--confirm-test-runtime yes required')
const action=createOrderFaultRuntime(process.env,scope)
if(mode==='preflight'){assert(args.routing,'--routing required');console.log(JSON.stringify(await action('preflight',{routing:JSON.parse(readFileSync(args.routing,'utf8'))}),null,2));return}
if(['status','reconcile'].includes(mode)){console.log(JSON.stringify(await action(mode),null,2));return}
assert(args.report && process.env.ADMIN_TEST_GATEWAY,'--report and ADMIN_TEST_GATEWAY required')
await mkdir(dirname(resolve(args.report)),{recursive:true})
const reportFile=await open(resolve(args.report),'wx',0o600) // Never overwrite a previous run or recovery report.
const business=businessClient(process.env,plan);let result
try {
await business.login()
if(mode==='run') {
assert(args.routing && /^sha256:[0-9a-f]{64}$/.test(args['expected-image']||''),'--routing and immutable --expected-image required')
result=await runOrderFaultWindow({plan,routing:JSON.parse(readFileSync(args.routing,'utf8')),expectedImageId:args['expected-image'],action,...business})
} else {
await action('restore')
for(let i=0;i<180;i++){if((await action('health')).healthy)break;assert(i<179,'RESTORE_HEALTH_TIMEOUT');await pause(1000)}
const runtime=await action('verify-restored'),businessRecovery=await business.recovery()
result={success:true,runtime,businessRecovery,restoration:await action('finish',{businessRecovery})}
}
} catch(error){result={success:false,error:/^[A-Z_0-9 :.-]+$/.test(error.message)?error.message:'EXECUTION_FAILED_USE_RUNTIME_STATUS'}}
finally {
const sessionsClosed=await business.logout();result ??={success:false};result.scope=scope;result.sessionsClosed=sessionsClosed
if(!sessionsClosed)result.success=false
try {await reportFile.writeFile(JSON.stringify(result,null,2)+'\n');await reportFile.sync()}
finally {await reportFile.close()}
}
console.log(JSON.stringify({success:result.success,report:resolve(args.report),restored:result.restoration?.restored===true}))
if(!result.success)process.exitCode=1
}
if(process.argv[1] && import.meta.url===pathToFileURL(resolve(process.argv[1])).href)main(process.argv.slice(2)).catch(error=>{console.error(/^[A-Z_0-9]{1,90}$/.test(error.message)?error.message:'ORDER_DEPENDENCY_ARGUMENT_OR_CONFIGURATION_ERROR');process.exitCode=1})