feat(dispatch): unify whole vehicle scheduling, source ownership and staged execution
This commit is contained in:
@@ -1,4 +1,4 @@
|
||||
import { mkdirSync, writeFileSync, openSync, closeSync, unlinkSync, readFileSync } from 'node:fs'
|
||||
import { mkdirSync, writeFileSync, openSync, closeSync, unlinkSync, readFileSync, fsyncSync } from 'node:fs'
|
||||
import { execFileSync } from 'node:child_process'
|
||||
import { dirname, resolve, join } from 'node:path'
|
||||
import { fileURLToPath, pathToFileURL } from 'node:url'
|
||||
@@ -34,7 +34,7 @@ export function parseCrossCustomerArguments(args) {
|
||||
}
|
||||
export function crossCustomerPlan(config) {
|
||||
return { environment: 'TEST', execute: config.execute, runId: config.runId, warehouseId: config.warehouseId || null, destinationWarehouseId: config.destinationWarehouseId || null,
|
||||
output: `reports/fleet-cross-${config.runId}`, stages: ['two new customers', 'two ordinary transfer receipts', 'two dispatches on one trip',
|
||||
output: `reports/fleet-cross-${config.runId}`, stages: ['two new customers', 'two ordinary transfer receipts', 'one native cross-customer dispatch and its unique automatic trip',
|
||||
'unready customer gate rejection', 'shared departure and replay', 'independent receipts', 'trip completion and separate resource release'],
|
||||
limits: 'Prerequisite rejection is not mid-transaction rollback. No deployment, configuration change, SQL writes or automatic rerun. Retain business samples; disable temporary PDA in finally.' }
|
||||
}
|
||||
@@ -45,17 +45,19 @@ export async function runCrossCustomer(config, dependencies = {}) {
|
||||
|
||||
/** Not a CLI mode: called only by the explicit GRAY pool entry with its verified pool scenario. */
|
||||
export async function runFleetPoolGrayCrossCustomer(config, dependencies = {}) {
|
||||
if (Object.keys(dependencies).some(key => !['env', 'outputRoot', 'verify', 'contextFactory'].includes(key)))
|
||||
if (Object.keys(dependencies).some(key => !['env', 'outputRoot', 'verify', 'contextFactory', 'verificationScope'].includes(key)))
|
||||
throw Error('Gray pool cannot override evidence readers or fixture lifecycle')
|
||||
if (dependencies.verificationScope != null && dependencies.verificationScope !== 'CAPACITY_POOL') throw Error('Gray pool requires explicit CAPACITY_POOL verification scope')
|
||||
if (config.execute && (typeof dependencies.verify !== 'function' || typeof dependencies.contextFactory !== 'function'))
|
||||
throw Error('Gray pool requires its dedicated scenario and journaled context')
|
||||
return runCrossCustomerCore(config, dependencies, true)
|
||||
return runCrossCustomerCore(config, { ...dependencies, verificationScope: 'CAPACITY_POOL' }, true)
|
||||
}
|
||||
|
||||
async function runCrossCustomerCore(config, { env = process.env, outputRoot = ROOT, readers, prepare, verify = verifyFleetCrossCustomer,
|
||||
finish = finishFleetTrip, contextFactory = options => new AcceptanceRun(options), pickerAccess = resolvePickerAccess, pickerCreate = createPicker, pickerDisable = disablePicker } = {}, gray = false) {
|
||||
verificationScope = 'CROSS_CUSTOMER', finish = finishFleetTrip, contextFactory = options => new AcceptanceRun(options), pickerAccess = resolvePickerAccess, pickerCreate = createPicker, pickerDisable = disablePicker } = {}, gray = false) {
|
||||
if (!['CROSS_CUSTOMER', 'CAPACITY_POOL', 'SAME_ROUTE_COST'].includes(verificationScope) || gray && verificationScope !== 'CAPACITY_POOL') throw Error('Explicit supported verification scope required')
|
||||
const environment = gray ? 'GRAY' : 'TEST', reportName = `${gray ? 'fleet-pool-gray' : 'fleet-cross'}-${config.runId}`
|
||||
const plan = { ...crossCustomerPlan(config), environment, output: `reports/${reportName}`,
|
||||
const plan = { ...crossCustomerPlan(config), verificationScope, environment, output: `reports/${reportName}`,
|
||||
...(gray ? { scope: 'FLEET_CAPACITY_POOL_GRAY', stages: ['current pool fixture and full pool HTTP matrix', 'trip completion and separate resource release'] } : {}) }
|
||||
if (!config.execute) return plan
|
||||
if (config.environment !== environment || !(gray ? ['gray', 'prod'].includes(env.CHONGCHONG_ENV) : env.CHONGCHONG_ENV === 'test') || !/^[A-Za-z0-9][A-Za-z0-9_-]{0,47}$/.test(config.runId)
|
||||
@@ -71,10 +73,14 @@ async function runCrossCustomerCore(config, { env = process.env, outputRoot = RO
|
||||
const reports = resolve(outputRoot, 'reports'), output = join(reports, reportName)
|
||||
mkdirSync(reports, { recursive: true })
|
||||
const lockPath = join(reports, '.fleet-acceptance.lock'), lock = openSync(lockPath, 'wx')
|
||||
let context, picker, createdPicker, initialized = false, report, evidenceIndex = 0, requestIndex = 0
|
||||
let context, picker, createdPicker, initialized = false, report, evidenceIndex = 0, requestIndex = 0, stopped = null
|
||||
const sessions = new Map()
|
||||
const secrets = Object.entries(env).filter(([name]) => /PASSWORD|TOKEN/.test(name)).map(([, value]) => value)
|
||||
const save = (name, value) => writeFileSync(join(output, name), JSON.stringify(redactEvidence(value, secrets), null, 2) + '\n')
|
||||
const save = (name, value) => {
|
||||
const descriptor = openSync(join(output, name), 'w')
|
||||
try { writeFileSync(descriptor, JSON.stringify(redactEvidence(value, secrets), null, 2) + '\n'); fsyncSync(descriptor) }
|
||||
finally { closeSync(descriptor) }
|
||||
}
|
||||
try {
|
||||
writeFileSync(lock, JSON.stringify({ pid: process.pid, runId: config.runId }))
|
||||
mkdirSync(output); initialized = true; mkdirSync(join(output, 'evidence')); mkdirSync(join(output, 'requests'))
|
||||
@@ -88,13 +94,50 @@ async function runCrossCustomerCore(config, { env = process.env, outputRoot = RO
|
||||
const options = gray ? createFleetPoolGrayOptions(transport) : validateExecutionOptions({
|
||||
...parseArguments(['--modules', 'merchant-qualification-fixture', '--quiet', 'true']), ...transport, environment: 'test' })
|
||||
context = contextFactory(options)
|
||||
const previousCheckpoint = context.checkpoint
|
||||
context.checkpoint = async () => {
|
||||
try {
|
||||
if (previousCheckpoint) await previousCheckpoint.call(context)
|
||||
save('progress.json', { marker: context.marker, steps: context.steps, createdResources: context.createdResources })
|
||||
} catch (error) {
|
||||
stopped ||= new Error('Checkpoint persistence failed; no further business request may be sent')
|
||||
context.createdResources.fleetWriteBoundary = { status: 'STOPPED', outcome: 'PERSISTENCE_FAILED', reason: 'CHECKPOINT_FAILED' }
|
||||
throw stopped
|
||||
}
|
||||
}
|
||||
for (const method of ['request', 'requestWithBody']) {
|
||||
const original = context[method]
|
||||
context[method] = async function (...args) {
|
||||
const number = ++requestIndex, [verb, path, body] = args
|
||||
const record = { number, method: verb, path, stage: 'INTENT', body: body instanceof FormData ? '[multipart file]' : body }
|
||||
save(`requests/${number}.json`, record)
|
||||
const response = await original.apply(this, args)
|
||||
const [verb, path, body] = args
|
||||
const identityCleanup = verb === 'POST' && path === '/api/auth/logout' || verb === 'PUT'
|
||||
&& [picker?.membershipId, createdPicker?.membershipId].filter(Boolean).some(value => path === `/api/platform/v1/iam/users/${value}/status`)
|
||||
&& body?.status === 'DISABLED'
|
||||
if (stopped && !identityCleanup) throw stopped
|
||||
const number = ++requestIndex
|
||||
const write = !['GET', 'HEAD', 'OPTIONS'].includes(String(verb).toUpperCase())
|
||||
let evidenceBody = body
|
||||
if (body instanceof FormData) {
|
||||
evidenceBody = { multipart: [] }
|
||||
for (const [field, value] of body.entries()) evidenceBody.multipart.push(typeof value === 'string' ? { field, value }
|
||||
: { field, name: value.name, type: value.type, size: value.size, sha256: createHash('sha256').update(Buffer.from(await value.arrayBuffer())).digest('hex') })
|
||||
}
|
||||
const record = { number, method: verb, path, stage: 'INTENT', headers: args[4], body: evidenceBody }
|
||||
let response, sent = false
|
||||
try {
|
||||
save(`requests/${number}.json`, record)
|
||||
await context.checkpoint()
|
||||
sent = true
|
||||
response = await original.apply(this, args)
|
||||
} catch (error) {
|
||||
if (write) {
|
||||
stopped ||= new Error(`${sent ? 'Unknown write outcome' : 'Write checkpoint failed before send'}: ${verb} ${path}; preserve original request, do not resend`)
|
||||
record.stage = sent ? 'UNKNOWN_RESULT' : 'NOT_SENT'
|
||||
context.createdResources.fleetWriteBoundary = { status: 'STOPPED', method: verb, path, outcome: record.stage }
|
||||
save(`requests/${number}.json`, { ...record, error: String(error?.message || error) })
|
||||
await context.checkpoint().catch(() => {})
|
||||
}
|
||||
throw error
|
||||
}
|
||||
if (verb === 'POST' && path === '/api/platform/v1/iam/users' && ok(response)
|
||||
&& String(body?.overseasWarehouseId) === config.warehouseId && body?.username?.startsWith('pda_overseas_')) {
|
||||
createdPicker = { membershipId: response.body?.data?.membershipId }
|
||||
@@ -104,8 +147,19 @@ async function runCrossCustomerCore(config, { env = process.env, outputRoot = RO
|
||||
sessions.set(tokens.accessToken, tokens.refreshToken)
|
||||
secrets.push(tokens.accessToken, tokens.refreshToken)
|
||||
}
|
||||
const code = String(response?.body?.code || ''), native = /^\/api\/(?:platform|pda)\/v1\/last-leg\/dispatches(?:\/|$)/.test(path)
|
||||
const definite = Number.isInteger(response?.http) && response.http >= 200 && response.http < 500 && response.requestId && response.traceId
|
||||
&& (response.http === 200 && code === '200' || /^[A-Z]\d{4}$/.test(code) && (response.http === 200 || response.http >= 400))
|
||||
&& !(native && code === '200' && response.body.data == null)
|
||||
if (write && !definite) {
|
||||
stopped ||= new Error(`Unknown write outcome: ${verb} ${path}; preserve original request, do not resend`)
|
||||
context.createdResources.fleetWriteBoundary = { status: 'STOPPED', method: verb, path, outcome: 'UNKNOWN_RESULT', requestId: response?.requestId, traceId: response?.traceId }
|
||||
save(`requests/${number}.json`, { ...record, stage: 'UNKNOWN_RESULT', response })
|
||||
await context.checkpoint()
|
||||
throw stopped
|
||||
}
|
||||
save(`requests/${number}.json`, { ...record, stage: 'RESPONSE', response })
|
||||
save('progress.json', { marker: context.marker, steps: context.steps, createdResources: context.createdResources })
|
||||
await context.checkpoint()
|
||||
return response
|
||||
}
|
||||
}
|
||||
@@ -125,10 +179,23 @@ async function runCrossCustomerCore(config, { env = process.env, outputRoot = RO
|
||||
}
|
||||
const fixture = await prepare(context, { execute: true, warehouseId: config.warehouseId, destinationWarehouseId: config.destinationWarehouseId, pdaToken: picker.token, readInventory })
|
||||
save('fixture.json', fixture)
|
||||
if (fixture?.marker !== context.marker || !fixture.dispatchNo || !fixture.tripNo || String(fixture.warehouseId) !== config.warehouseId
|
||||
|| fixture.members?.length !== 2 || new Set(fixture.members.map(member => member.customerId)).size !== 2
|
||||
|| fixture.members.some(member => member.marker !== context.marker || member.dispatchNo !== fixture.dispatchNo || !member.customerId || member.receiptVerified !== true))
|
||||
throw Error('Current fixture requires one exact native main and two owned customer receipts')
|
||||
const trips = context.createdResources.fleetTrips?.filter(row => row.tripNo === fixture.tripNo && row.dispatchNo === fixture.dispatchNo)
|
||||
if (trips?.length !== 1 || fixture.tripProof !== trips[0] || String(trips[0].warehouseId) !== config.warehouseId)
|
||||
throw Error('Original fixture native main/trip proof missing or ambiguous for resource completion')
|
||||
const trip = trips[0]
|
||||
await context.checkpoint()
|
||||
const verification = await verify(context, fixture, { execute: true, readInventory })
|
||||
if (verification?.status !== 'VERIFIED' || verification.tripNo !== fixture.tripNo) throw Error('Current fixture verification incomplete')
|
||||
const trip = context.createdResources.fleetTrips?.find(row => row.tripNo === fixture.tripNo)
|
||||
if (!trip) throw Error('Original fixture trip missing for resource completion')
|
||||
const proofKey = { CROSS_CUSTOMER: 'fleetCrossCustomer', CAPACITY_POOL: 'fleetCapacityPool', SAME_ROUTE_COST: 'fleetSameRouteCost' }[verificationScope]
|
||||
const authoritative = context.createdResources[proofKey]
|
||||
if (verification !== authoritative || verification?.status !== 'VERIFIED' || verification.marker !== context.marker
|
||||
|| verification.dispatchNo !== fixture.dispatchNo || verification.tripNo !== fixture.tripNo)
|
||||
throw Error('Current fixture verification incomplete or not from the explicitly selected scope')
|
||||
if (fixture.tripProof !== trip || trip.dispatchNo !== fixture.dispatchNo || trip.tripNo !== fixture.tripNo) throw Error('Native main resource proof changed during verification')
|
||||
await context.checkpoint()
|
||||
const beforeClosure = []
|
||||
for (const member of fixture.members) beforeClosure.push(await readInventory({ marker: context.marker, customerId: member.customerId, warehouseId: config.warehouseId }))
|
||||
await finish(context, 'fleet-cross-customer-closure', trip)
|
||||
@@ -146,8 +213,10 @@ async function runCrossCustomerCore(config, { env = process.env, outputRoot = RO
|
||||
if (context) {
|
||||
let sessionsClosed = true
|
||||
for (const [accessToken, refreshToken] of sessions) {
|
||||
const response = await context.request('POST', '/api/auth/logout', { refreshToken }, accessToken)
|
||||
if (!ok(response) && response.http !== 401 && String(response.body?.code) !== 'A0401') sessionsClosed = false
|
||||
try {
|
||||
const response = await context.request('POST', '/api/auth/logout', { refreshToken }, accessToken)
|
||||
if (!ok(response) && response.http !== 401 && String(response.body?.code) !== 'A0401') sessionsClosed = false
|
||||
} catch { sessionsClosed = false }
|
||||
}
|
||||
report.sessionsClosed = sessionsClosed
|
||||
report.summary = buildAcceptanceSummary(context.steps, context.modules, context.coverageGaps)
|
||||
@@ -155,7 +224,7 @@ async function runCrossCustomerCore(config, { env = process.env, outputRoot = RO
|
||||
report.marker = context.marker; report.createdResources = context.createdResources; report.steps = context.steps
|
||||
}
|
||||
report ||= { success: false }
|
||||
report.environment = environment; report.runtimeProfile = env.CHONGCHONG_ENV
|
||||
report.verificationScope = verificationScope; report.environment = environment; report.runtimeProfile = env.CHONGCHONG_ENV
|
||||
if (gray) report.scope = 'FLEET_CAPACITY_POOL_GRAY'
|
||||
report.requestCount = requestIndex; report.inventorySnapshots = evidenceIndex; report.goalComplete = false
|
||||
try { if (initialized) save('report.json', report) } finally { closeSync(lock); unlinkSync(lockPath) }
|
||||
|
||||
Reference in New Issue
Block a user