From 51bf4e5558b8f294b37ba5c3644af37fb7172c92 Mon Sep 17 00:00:00 2001 From: ouyangxq <775153912@qq.com> Date: Sat, 3 Oct 2026 08:13:04 +0800 Subject: [PATCH] test(fba): verify signed delivery ACK loss and original event recovery --- scripts/README.md | 6 +- .../phase5-root-fulfillment-contract.test.mjs | 3 +- .../phase5-root-fulfillment.mjs | 52 ++++++- scripts/event-delivery-fault-proxy.py | 10 +- scripts/event-delivery-fault-proxy.test.mjs | 44 +++++- scripts/fba-core-regression-test-env.mjs | 66 +++++++-- scripts/fba-core-regression-test-env.test.mjs | 39 +++++ scripts/test-delivery-fault-proof.mjs | 76 ++++++++++ scripts/test-event-fault-executor.mjs | 26 ++-- scripts/test-event-fault-executor.test.mjs | 137 +++++++++++++++++- scripts/test-event-fault-runtime.py | 97 +++++++++---- 11 files changed, 499 insertions(+), 57 deletions(-) create mode 100644 scripts/test-delivery-fault-proof.mjs diff --git a/scripts/README.md b/scripts/README.md index b0ea465c0..d2e1cd427 100644 --- a/scripts/README.md +++ b/scripts/README.md @@ -1,5 +1,9 @@ # 后端脚本管理与全量 HTTP 验收索引 +`test-delivery-fault-proof.mjs` 是签收回写故障的内部证据校验器，无CLI、网络或写入。按精确客户/派送/来源入仓/来源订单/根订单核对原事件及载荷摘要、13次投递与代理记录、唯一收件箱和人工重试审计，并比较重试前后的Firstleg与Order事实。首片签收仍有有效余量时允许根单继续等待；不把根单必须立即COMPLETED当成成功条件。 + +F40 专项可用 `--signature-mode ROOT_FIRST|CHILD_FIRST|ROOT_SIGNED_REMAINDER_CANCELLED` 只运行显式指定的一种顺序，省略仍串行执行三种；报告和成功范围仅覆盖所列模式。`--verify-delivery-event-recovery` 必须同时显式指定一种顺序，且环境为TEST；仅在该模式第一张真实FBA派送签收前开启定向ACK丢失，不能使用普通海外转运的签收替代国内回写。故障窗口会重建同镜像Lastleg并临时调整事件目标和退避配置，禁止与其他部署或共享环境写入并行；原事件DEAD恢复后必须验证原镜像/配置/网络身份、健康及代理退出。入口事先留存精确故障scope，finally兜底恢复，缺恢复证据或恢复校验失败一律失败。完整库存取证接入既有只读器，报告保留于该批次 `evidence/`，不能把某一模式或事件通过解读为全部反向事件完成。 + `fleet-cross-customer-test-env.mjs` 是最小两客户共享车次的串行 TEST 专项入口，默认 `--plan` 零网络。显式运行参数为 `--execute --environment TEST --run-id <新批次> --warehouse-id <实际海外收货仓ID> --destination-warehouse-id `，凭据仅从显式 `node --env-file="" --env-file=""` 与 `FBA_CORE_JENKINS_AGENT` 提供。通过普通转运来货、PDA实收和客户确认创建两个新客户各一托一箱，分别预排并关联同车；一单未拣齐时验证全车拒绝及完整库存快照不变，再统一发车、原键重放、逐客户签收，最后正式结案并分别释放车辆/司机，核对结案前后库存零变化。完整库存由只读取证器获取，不写业务SQL，不部署或改共享配置。报告位于 `reports/fleet-cross-<新批次>/`，每次请求发送前记录意图；失败保留原现场，不自动重跑或续跑，已存在批次拒绝覆盖。finally停用本轮PDA并退出独立会话；业务样本保留。前置拒绝不代表事务中途回滚，后者另验。对应入口保护合同为 `fleet-cross-customer-test-env.test.mjs`；内部生成与验收函数见模块README。 头程订单提交使用首次原始详情的 `expectedVersion` 与原 `Idempotency-Key`。生产单票的 `SUBMIT_INTENT` 必须先持久化 `submitCommand` 再发送，显式续跑不得使用当前版本替换原命令，缺少原命令的提交意图及后续检查点停止并等待人工核对。源 Excel 造数同样在提交前保存订单条目的 `submitCommand`；中断或失败条目保留原检查点且不自动重跑，不重建订单或换版本继续。 @@ -22,7 +26,7 @@ Manifest 使用 `version:1, scope, revision, requirements:[{id,domain,level,chec 离线回归：`node --test scripts/acceptance-evidence-audit.test.mjs`。命令失败返回1，参数/文件错误返回2；报告写入调用者指定的新路径。 -`event-delivery-fault-proxy.py` 是 TEST 环境定向事件 ACK 丢失工具，不创建或重放事件。运行者显式传入 `--upstream` 完整消费者地址、`--event-type`，并只选 `--event-id`、`--loading-plan-id` 或 `--container-booking-id`；柜次选择精确匹配 payload.sourceContainerBookingId，用于海外卸柜回写，不接受混合选择条件。`EVENT_FAULT_CONTROL_TOKEN` 从外部环境注入，保护固定本机监听的控制端口。默认 PASS；经鉴权切为 FAULT 后，仅在选定事件已真实转发且收到匹配的 APPLIED/DUPLICATE ACK 时向发送方返回 503，其他事件直通。同一选择范围内锁定首个事件后不改选其他事件 ID，原 ID 改摘要或载荷直接拒绝。日志只保留事件标识、摘要和结果，不保留正文或凭据。它会影响实际投递，仅限明确授权的测试环境；生产者配置、重试、DEAD 原事件人工恢复及最终恢复直连由运行者负责，工具退出不自动恢复服务配置。`--log` 可显式指定脱敏证据文件，失败保留原事件与日志，禁止改摘要或借其他事件重跑。`event-delivery-fault-proxy.test.mjs` 使用本机替身服务验证行为，不访问共享环境；工具离线通过不能替代真实重试验收。 +`event-delivery-fault-proxy.py` 是 TEST 环境定向事件 ACK 丢失工具，不创建或重放事件。运行者显式传入 `--upstream` 完整消费者地址、`--event-type`，并只选 `--event-id`、`--loading-plan-id`、`--container-booking-id` 或 `--delivery-id`；派送选择器只读取正式 `payload.deliveryOrderId`，不接受 `deliveryId` 载荷别名；柜次选择精确匹配 payload.sourceContainerBookingId，用于海外卸柜回写，不接受混合选择条件。`EVENT_FAULT_CONTROL_TOKEN` 从外部环境注入，保护固定本机监听的控制端口。默认 PASS；经鉴权切为 FAULT 后，仅在选定事件已真实转发且收到匹配的 APPLIED/DUPLICATE ACK 时向发送方返回 503，其他事件直通。同一选择范围内锁定首个事件后不改选其他事件 ID，原 ID 改摘要或载荷直接拒绝。日志只保留事件标识、摘要和结果，不保留正文或凭据。它会影响实际投递，仅限明确授权的测试环境；生产者配置、重试、DEAD 原事件人工恢复及最终恢复直连由运行者负责，工具退出不自动恢复服务配置。`--log` 可显式指定脱敏证据文件，失败保留原事件与日志，禁止改摘要或借其他事件重跑。`event-delivery-fault-proxy.test.mjs` 使用本机替身服务验证行为，不访问共享环境；工具离线通过不能替代真实重试验收。 `admin-test-env-acceptance.mjs` 与 `client-test-env-acceptance.mjs` 仅接受统一 TEST Gateway 和 Auth 会话；移除环境文件中的 `ADMIN_OVERSEAS_GATEWAY`，不再传 `--overseas-gateway` 或非 test 环境。平台、客户、PDA 身份仍分别登录并验证权限及客户隔离。海外资源接口使用 `/api/platform/v1/overseas-warehouse-resources/**`，仓库主档仍归 Warehouse。跨服务模块的重复 GET 不证明事件重放；真实重试/幂等证据缺失会报告部署阻断，详见模块 README 和下方事件人工重试入口。离线合同不会执行真实 HTTP；环境执行仍需本轮写入授权。 diff --git a/scripts/admin-test-env-modules/phase5-root-fulfillment-contract.test.mjs b/scripts/admin-test-env-modules/phase5-root-fulfillment-contract.test.mjs index 95ebf2c33..faf894e84 100644 --- a/scripts/admin-test-env-modules/phase5-root-fulfillment-contract.test.mjs +++ b/scripts/admin-test-env-modules/phase5-root-fulfillment-contract.test.mjs @@ -290,8 +290,9 @@ test('complete orchestrator calls actual source chain and blocks missing Outbox for(const name of ['orders','inbound-measurement','phase3-customs','phase3-outbound-dispatch', 'phase3-carrier-schedule-queries'])assert.ok(source.includes(`'${name}'`)) for(const name of ['loadOneFragment','verifyPalletLabels','awaitProducerConsumption', - 'convertReceivingContainers','prepareTerminalDrayage','runReceiving','pickAll','saveReceiptAndComplete']) + 'convertReceivingContainers','prepareTerminalDrayage','runReceiving','pickAll','withDeliveryEventRecovery']) assert.match(source,new RegExp(`await ${name}\\(`)) + assert.match(source,/\(\) => saveReceiptAndComplete\(context, deliveries\[0\], customerId, index\)/) assert.match(source,/rootOrderId,\s*sourceOrderIds:\[rootOrderId,childOrderId\]/) assert.match(source,/HTTP_VERIFIED_OUTBOX_PENDING/) assert.match(source,/finally \{ preserveReceivingEvidence\(proof,scope\); proof.currentStage = 'PDA_IDENTITY_CLEANUP'; await disablePicker/) diff --git a/scripts/admin-test-env-modules/phase5-root-fulfillment.mjs b/scripts/admin-test-env-modules/phase5-root-fulfillment.mjs index 440cf455f..5d16542fa 100644 --- a/scripts/admin-test-env-modules/phase5-root-fulfillment.mjs +++ b/scripts/admin-test-env-modules/phase5-root-fulfillment.mjs @@ -5,6 +5,8 @@ import { stableLoadedCartonMarks } from './phase3-loading.mjs' import { awaitProducerConsumption, convertReceivingContainers, runReceiving, resolveSharedWarehouse, resolvePickerAccess, createPicker, disablePicker } from './phase4-overseas-pda.mjs' import { pickAll, verifyDepartureReadiness, saveReceiptAndComplete } from './phase5-overseas-fulfillment.mjs' +import { isDeepStrictEqual } from 'node:util' +import { assertTestDeliveryFaultProof } from '../test-delivery-fault-proof.mjs' import { prepareTerminalDrayage } from './phase5-terminal-drayage.mjs' import { prepareTaxCustoms, verifyTaxDocumentsAfterCompletion } from './phase5-root-tax-customs.mjs' import { prepareFleetTrip, confirmFleetLoading, finishFleetTrip } from './fleet-trip-fulfillment.mjs' @@ -432,7 +434,10 @@ async function sendFragment(context, producer, picker, merchantToken, index) { query(`/api/platform/v1/last-leg/delivery-orders/${deliveries[0].deliveryNo}/status`,{customerId}), {status:'DELIVERING',expectedVersion:Number(beforeDelivery.order.version),remark:context.marker}) check(context,'F40 派送处于配送中后保存签收',delivering.deliveryStatus === 'DELIVERING',`delivery=${deliveries[0].deliveryNo}`) - const completed = await saveReceiptAndComplete(context, deliveries[0], customerId, index) + const completed = await withDeliveryEventRecovery(context, { + marker: context.context.f40BaseMarker, customerId: id(customerId), deliveryId: id(beforeDelivery.order.id), + sourceInboundId: id(producer.sourceInboundId), sourceOrderId: id(producer.sourceOrderId), rootOrderId: id(producer.rootOrderId), + }, id(picker.warehouseId), index, () => saveReceiptAndComplete(context, deliveries[0], customerId, index)) check(context, 'F40 实际签收资料驱动派送完成', completed?.order?.status === 'DELIVERED', `delivery=${deliveries[0].deliveryNo}`) check(context,'F40 签收完成记录真实签收时间', Boolean(completed.order.signedAt), `delivery=${deliveries[0].deliveryNo}`) const tripClosure = await finishFleetTrip(context, MODULE, tripProof) @@ -440,6 +445,51 @@ async function sendFragment(context, producer, picker, merchantToken, index) { receiptVersion: completed.order.version, signedAt: completed.order.signedAt, receiptSnapshot:completed, tripNo: tripProof.tripNo, tripClosure } } +/** Only the first actual FBA signature is faulted; caller performs the normal receipt command. */ +export async function withDeliveryEventRecovery(context, faultScope, warehouseId, index, sign) { + const executor = context.context.f40DeliveryEventFaultExecutor + if (!executor || index !== 0) return sign() + const proof = context.createdResources.phase5RootFulfillment + if (proof.deliveryEventRecovery) throw Error('F40 delivery recovery already attempted') + const reader = context.context.f40ReadOnlyInventoryEvidence + if (typeof reader !== 'function') throw Error('F40 complete inventory evidence reader required') + const record = proof.deliveryEventRecovery = { scope: faultScope, status: 'PREPARING', restored: false } + const inventoryScope = { marker: faultScope.marker, customerId: faultScope.customerId, warehouseId } + try { + record.inventoryBefore = await reader(inventoryScope) + record.preparationAttempted = true + await executor.prepare(faultScope) + record.status = 'SIGNING' + const result = await sign() + record.status = 'RECOVERING' + record.proof = await executor.readProof(faultScope) + assertTestDeliveryFaultProof(record.proof, faultScope) + if (record.proof.restored !== true) throw Error('F40 delivery producer restoration missing') + record.inventoryAfter = await reader(inventoryScope) + if (record.inventoryBefore?.scope?.coverage !== 'CUSTOMER_WAREHOUSE_COMPLETE' + || record.inventoryBefore.scope.customerId !== faultScope.customerId + || record.inventoryBefore.scope.warehouseId !== warehouseId + || !['balances','positions','ledger'].every(key => Array.isArray(record.inventoryBefore[key])) + || !isDeepStrictEqual(record.inventoryBefore, record.inventoryAfter)) throw Error('F40 full inventory drift during delivery recovery') + record.inventoryUnchanged = true + record.status = 'VERIFIED' + return result + } catch (error) { + record.status = 'FAILED' + throw error + } finally { + if (record.preparationAttempted) try { + record.restoration = await executor.cleanup() + record.restored = record.restoration?.restored === true && record.restoration?.healthy === true && record.restoration?.proxyStopped === true + if (!record.restored) throw Error('F40 delivery restoration incomplete') + } catch (error) { + record.status = 'RESTORATION_FAILED' + record.restored = false + throw error + } + } +} + /** Explicit TEST orchestrator, called by the parent acceptance context; never synthesizes an event. */ export async function run(context) { try { return await runRootFulfillment(context) } diff --git a/scripts/event-delivery-fault-proxy.py b/scripts/event-delivery-fault-proxy.py index 7bd8fa203..28d18ce37 100644 --- a/scripts/event-delivery-fault-proxy.py +++ b/scripts/event-delivery-fault-proxy.py @@ -32,11 +32,12 @@ def encoded(value): class State: - def __init__(self, event_type, event_id, loading_plan_id, log_path=None, container_booking_id=None): + def __init__(self, event_type, event_id, loading_plan_id, log_path=None, container_booking_id=None, delivery_id=None): self.event_type = event_type self.selector_id = event_id self.plan_id = loading_plan_id self.booking_id = container_booking_id + self.delivery_id = delivery_id self.log_path = log_path self.lock = threading.Lock() self.mode = "PASS" @@ -70,6 +71,8 @@ class State: selected = event_id == self.selector_id elif self.booking_id: selected = isinstance(payload, dict) and str(payload.get("sourceContainerBookingId")) == self.booking_id + elif self.delivery_id: + selected = isinstance(payload, dict) and str(payload.get("deliveryOrderId")) == self.delivery_id else: selected = isinstance(payload, dict) and str(payload.get("loadingPlanId")) == self.plan_id qualifies = event.get("eventType") == self.event_type and selected @@ -260,6 +263,7 @@ def main(): selector.add_argument("--event-id") selector.add_argument("--loading-plan-id") selector.add_argument("--container-booking-id") + selector.add_argument("--delivery-id") parser.add_argument("--bind", default="127.0.0.1") parser.add_argument("--port", type=int, default=0) parser.add_argument("--control-port", type=int, default=0) @@ -278,9 +282,11 @@ def main(): parser.error("loading-plan-id must be a positive decimal identifier") if args.container_booking_id and not re.fullmatch(r"[1-9][0-9]{0,19}", args.container_booking_id): parser.error("container-booking-id must be a positive decimal identifier") + if args.delivery_id and not re.fullmatch(r"[1-9][0-9]{0,19}", args.delivery_id): + parser.error("delivery-id must be a positive decimal identifier") if not re.fullmatch(r"[A-Za-z][A-Za-z0-9]{0,79}", args.event_type) or not 0 < args.timeout <= 120: parser.error("invalid event type or timeout") - state = State(args.event_type, args.event_id, args.loading_plan_id, args.log, args.container_booking_id) + state = State(args.event_type, args.event_id, args.loading_plan_id, args.log, args.container_booking_id, args.delivery_id) data = ThreadingHTTPServer((args.bind, args.port), data_handler(state, args.upstream, args.timeout)) control = ThreadingHTTPServer(("127.0.0.1", args.control_port), control_handler(state, token)) for server in (data, control): diff --git a/scripts/event-delivery-fault-proxy.test.mjs b/scripts/event-delivery-fault-proxy.test.mjs index 5d1dde42b..bd149224b 100644 --- a/scripts/event-delivery-fault-proxy.test.mjs +++ b/scripts/event-delivery-fault-proxy.test.mjs @@ -3,7 +3,7 @@ import { spawn, spawnSync } from 'node:child_process' import { createServer } from 'node:http' import { mkdtemp, readFile, rm } from 'node:fs/promises' import { tmpdir } from 'node:os' -import { join } from 'node:path' +import { join, dirname, basename, resolve } from 'node:path' import { fileURLToPath } from 'node:url' import test from 'node:test' @@ -78,6 +78,8 @@ async function fixture(selector = ['--loading-plan-id', '2100000000000000001'], child.kill(); await exited } await close(consumer) + assert.equal(dirname(resolve(directory)), resolve(tmpdir())) + assert.ok(basename(directory).startsWith('event-fault-offline-')) await rm(directory, { recursive: true }) }, } @@ -218,3 +220,43 @@ test('booking selector rejects malformed identities and mixed selection before s assert.ok(!result.stdout.includes('"ready"')) } }) + +test('delivery selector isolates the first exact delivery event and preserves its ACK-loss recovery', async t => { + const deliveryId = '2100000000000000088' + const proxy = await fixture(['--delivery-id', deliveryId], 'OverseasDeliveryCompleted') + t.after(() => proxy.stop()) + await proxy.control('/mode', { mode: 'FAULT' }) + const signed = { ...event(), eventType: 'OverseasDeliveryCompleted', + payload: { deliveryOrderId: deliveryId, sourceInboundId: '2100000000000000099' } } + for (const unrelated of [ + { ...signed, eventId: otherId, payload: { deliveryOrderId: '2100000000000000089' } }, + { ...signed, eventId: otherId, payload: { deliveryId } }, + { ...signed, eventId: otherId, payload: { loadingPlanId: deliveryId } }, + { ...signed, eventId: otherId, eventType: 'OverseasContainerUnloaded' }, + ]) assert.equal((await proxy.send(unrelated)).status, 200) + assert.equal((await proxy.send(signed)).status, 503) + assert.equal((await proxy.send(signed)).status, 503) + assert.equal((await proxy.send({ ...signed, eventId: samePlanOtherId })).status, 200) + const calls = proxy.calls.length + assert.equal((await proxy.send({ ...signed, payload: { deliveryOrderId: '2100000000000000089' } })).status, 409) + assert.equal(proxy.calls.length, calls) + await proxy.control('/mode', { mode: 'PASS' }) + const recovered = await proxy.send(signed) + assert.equal(recovered.status, 200) + assert.equal(JSON.parse(recovered.raw).data.outcome, 'DUPLICATE') + const status = await (await proxy.control('/status')).json() + assert.equal(status.eventId, targetId) + assert.deepEqual(status.records.filter(row => row.errorCode !== 'TARGET_EVENT_HASH_CONFLICT').map(row => row.outcome), ['APPLIED', 'DUPLICATE', 'DUPLICATE']) + assert.ok(!(await readFile(proxy.log, 'utf8')).includes('sourceInboundId')) +}) + +test('delivery selection rejects malformed IDs or a second selector before opening a listener', () => { + for (const args of [['--delivery-id', '0'], ['--delivery-id', '../88'], ['--delivery-id', '88', '--event-id', targetId], + ['--delivery-id', '88', '--container-booking-id', '66'], ['--delivery-id', '88', '--loading-plan-id', '77']]) { + const result = spawnSync(process.env.PYTHON || 'python', [script, '--upstream', 'http://127.0.0.1:1/internal/events', + '--event-type', 'OverseasDeliveryCompleted', ...args], { encoding: 'utf8', timeout: 5000, + env: { ...process.env, EVENT_FAULT_CONTROL_TOKEN: controlToken } }) + assert.equal(result.status, 2) + assert.ok(!result.stdout.includes('"ready"')) + } +}) diff --git a/scripts/fba-core-regression-test-env.mjs b/scripts/fba-core-regression-test-env.mjs index 454f9abec..1f6229a6f 100644 --- a/scripts/fba-core-regression-test-env.mjs +++ b/scripts/fba-core-regression-test-env.mjs @@ -15,18 +15,26 @@ export function parseCoreArguments(argv) { const values={},flags=new Set() for(let i=0;i1)throw Error('Choose plan or execute') + if(flags.has('--plan')&&flags.has('--execute'))throw Error('Choose plan or execute') if(values['--environment'] && values['--environment']!=='TEST')throw Error('Only explicit TEST is supported') const execute=flags.has('--execute'),runId=values['--run-id']||(!execute?'plan':null) if(!runId||!/^[A-Za-z0-9][A-Za-z0-9_-]{0,47}$/.test(runId))throw Error('Fresh run-id is required') if(execute&&values['--environment']!=='TEST')throw Error('Execution requires --environment TEST') const taxMode=values['--tax-mode']||'BUY_DECLARATION' if(!F40_TAX_MODES.includes(taxMode))throw Error('Unsupported F40 tax mode') - return {execute,environment:'TEST',runId,taxMode,outputDir:values['--output-dir']||`reports/fba-core-${runId}`} + const signatureMode=values['--signature-mode'] + if(signatureMode&&!F40_MODES.includes(signatureMode))throw Error('Unsupported F40 signature mode') + const verifyDeliveryEventRecovery=flags.has('--verify-delivery-event-recovery') + if(verifyDeliveryEventRecovery&&!signatureMode)throw Error('Delivery fault requires one explicit signature mode') + return {execute,environment:'TEST',runId,taxMode,signatureMode,verifyDeliveryEventRecovery,outputDir:values['--output-dir']||`reports/fba-core-${runId}`} +} +function selectedModes(config) { + if(config.signatureMode&&!F40_MODES.includes(config.signatureMode))throw Error('Unsupported F40 signature mode') + return config.signatureMode?[config.signatureMode]:F40_MODES } function taxScope(taxMode) { if(!F40_TAX_MODES.includes(taxMode))throw Error('Unsupported F40 tax mode') @@ -36,13 +44,16 @@ function taxScope(taxMode) { } export function executionPlan(config) { return {environment:'TEST',execution:config.execute?'EXECUTE':'PLAN',runId:config.runId,outputDir:config.outputDir, - taxMode:config.taxMode,scope:taxScope(config.taxMode), - modes:F40_MODES,module:'phase5-root-fulfillment',requiredEnvironmentKeys:REQUIRED_ENV_KEYS, + taxMode:config.taxMode,scope:taxScope(config.taxMode),verifyDeliveryEventRecovery:config.verifyDeliveryEventRecovery===true, + modes:selectedModes(config),module:'phase5-root-fulfillment',requiredEnvironmentKeys:REQUIRED_ENV_KEYS, optionalEnvironmentKeys:['CHONGCHONG_MERCHANT_USERNAME','CHONGCHONG_MERCHANT_PASSWORD'], - behavior:'Serial fresh fixtures; stop on first failed mode. HTTP creates customers/orders/files and uses shared PDA/pallet resources. SQL is read-only; no deploy or database mutation.', - reports:['manifest.json','summary.json',...F40_MODES.map(mode=>`${mode.toLowerCase()}.json`),'evidence/*.json'], + behavior:config.verifyDeliveryEventRecovery + ?'One explicit mode, first FBA receipt only: scoped ACK loss, original DEAD retry and verified Lastleg image/configuration restoration. HTTP creates fresh fixtures; SQL is read-only. Do not overlap deployments or other shared-environment writers.' + :'Serial fresh fixtures; stop on first failed mode. HTTP creates customers/orders/files and uses shared PDA/pallet resources. SQL is read-only; no deploy or database mutation.', + reports:['manifest.json','summary.json',...selectedModes(config).map(mode=>`${mode.toLowerCase()}.json`),'evidence/*.json'], retention:'Exclusive run directory and evidence files. Interrupted lock requires operator review; no automatic cleanup or resume.', - scopeLimit:'F40 three modes and receiving/completion evidence; not all F01-F40, UI acceptance or reverse event replay.'} + scopeLimit:config.verifyDeliveryEventRecovery?'Only the selected first delivery event and listed F40 mode; not all reverse events, fee-void recovery, or UI acceptance.' + :'Only the listed F40 modes and receiving/completion evidence; not all F01-F40, UI acceptance or reverse event replay.'} } export function validateCoreEnvironment(env,root=repositoryRoot) { if(basename(root)!=='baiyue-firstleg'||basename(dirname(root))!=='chongchong-global')throw Error('TEST workspace required') @@ -84,9 +95,11 @@ function createAcceptanceContext(mode,config,env,gateway) { markerNamespace:createHash('sha256').update(config.runId+':'+mode).digest('hex').slice(0,12)}) return new AcceptanceRun(options) } -export async function runFbaCore(config,{env=process.env,readers,createContext=createAcceptanceContext,outputRoot=repositoryRoot}={}) { +export async function runFbaCore(config,{env=process.env,readers,createContext=createAcceptanceContext,outputRoot=repositoryRoot,faultFactory}={}) { const scopeDescription=taxScope(config.taxMode) + const modes=selectedModes(config) const gateway=validateCoreEnvironment(env) + if(config.verifyDeliveryEventRecovery&&(env.CHONGCHONG_ENV!=='test'||!config.signatureMode))throw Error('Delivery fault requires explicit TEST and one signature mode') const output=resolve(outputRoot,config.outputDir),reports=resolve(outputRoot,'reports'),rel=relative(reports,output) if(!config.execute||config.environment!=='TEST'||!rel||rel.startsWith('..')||isAbsolute(rel))throw Error('Fresh report directory inside reports required') mkdirSync(reports,{recursive:true}) @@ -94,7 +107,7 @@ export async function runFbaCore(config,{env=process.env,readers,createContext=c const lockFd=openSync(lock,'wx') const secrets=[...REQUIRED_ENV_KEYS,'CHONGCHONG_MERCHANT_USERNAME','CHONGCHONG_MERCHANT_PASSWORD'] .filter(key=>/PASSWORD|TOKEN|USERNAME|JENKINS_USER/.test(key)).map(key=>env[key]) - const summary={runId:config.runId,environment:'TEST',taxMode:config.taxMode,startedAt:new Date().toISOString(),success:false,modes:[],scope:scopeDescription} + const summary={runId:config.runId,environment:'TEST',taxMode:config.taxMode,verifyDeliveryEventRecovery:config.verifyDeliveryEventRecovery===true,startedAt:new Date().toISOString(),success:false,modes:[],scope:scopeDescription} let initialized=false const merchantBefore={username:process.env.MERCHANT_TEST_USERNAME,password:process.env.MERCHANT_TEST_PASSWORD} try { @@ -105,7 +118,7 @@ export async function runFbaCore(config,{env=process.env,readers,createContext=c save('manifest.json',executionPlan(config)) readers ||= createTestEvidenceReaders(env) await readers.preflight() - for(const mode of F40_MODES) { + for(const mode of modes) { const context=createContext(mode,config,env,gateway) let sequence=0 const evidence=async(kind,scope,read)=>{ @@ -118,17 +131,42 @@ export async function runFbaCore(config,{env=process.env,readers,createContext=c context.context.f40TaxMode=config.taxMode context.context.f40ReadOnlyOutboxEvidence=scope=>evidence('completion',scope,readers.readCompletion) context.context.phase4ReceivingEvidenceReader=scope=>evidence('receiving',scope,readers.readReceiving) + let fault, faultPreparationAttempted=false + if(config.verifyDeliveryEventRecovery) { + if(typeof readers.readFleetInventory!=='function')throw Error('Complete raw inventory reader required for delivery recovery') + const factory=faultFactory||(await import('./test-event-fault-executor.mjs')).createTestDeliveryFaultExecutor + const executor=factory(env) + fault={...executor,prepare:async scope=>{ + save(`evidence/${mode.toLowerCase()}-delivery-scope.json`,{scope,stage:'PREPARE_INTENT'}) + faultPreparationAttempted=true;return executor.prepare(scope) + }} + context.context.f40DeliveryEventFaultExecutor=fault + context.context.f40ReadOnlyInventoryEvidence=scope=>evidence('inventory',scope,readers.readFleetInventory) + } let report try {report=await context.run()} catch (error) {report={success:false,fatal:'FBA_CORE_MODE_EXCEPTION',diagnostic:safeFailure(error),summary:{failed:1},createdResources:context.createdResources}} - finally {secrets.push(context.token,context.pdaToken,context.refreshToken,context.context?.merchantQualificationFixture?.password)} + finally { + secrets.push(context.token,context.pdaToken,context.refreshToken,context.context?.merchantQualificationFixture?.password) + if(fault) { + try { + const restoration=await fault.cleanup() + save(`evidence/${mode.toLowerCase()}-delivery-restoration.json`,{attempted:faultPreparationAttempted,restoration:restoration??null}) + if(faultPreparationAttempted&&(restoration?.restored!==true||restoration.healthy!==true||restoration.proxyStopped!==true))throw Error('Delivery restoration unproven') + } catch(error) {report={...report,success:false,deliveryRestorationFailed:true,restorationDiagnostic:safeFailure(error)}} + } + } + if(config.verifyDeliveryEventRecovery) { + const proof=report?.createdResources?.phase5RootFulfillment?.deliveryEventRecovery + if(proof?.status!=='VERIFIED')report={...report,success:false,deliveryRecoveryUnproven:true} + } report={...report,regressionScope:{taxMode:config.taxMode,signatureOrder:mode,description:scopeDescription}} const name=`${mode.toLowerCase()}.json`;save(name,report) const passed=modePassed(report) summary.modes.push({mode,taxMode:config.taxMode,report:name,passed,summary:report.summary,evidenceFiles:sequence}) if(!passed)break } - summary.success=summary.modes.length===F40_MODES.length&&summary.modes.every(mode=>mode.passed) + summary.success=summary.modes.length===modes.length&&summary.modes.every(mode=>mode.passed) } catch (error) {summary.failure='FBA_CORE_PREFLIGHT_OR_RUN_FAILED';summary.diagnostic=safeFailure(error)} finally { summary.finishedAt=new Date().toISOString() diff --git a/scripts/fba-core-regression-test-env.test.mjs b/scripts/fba-core-regression-test-env.test.mjs index e72680bc6..ace7bca1d 100644 --- a/scripts/fba-core-regression-test-env.test.mjs +++ b/scripts/fba-core-regression-test-env.test.mjs @@ -33,6 +33,16 @@ test('tax mode is explicit, defaults to BUY and rejects aliases, duplicate and u for(const args of [['--tax-mode','REFUND'],['--tax-mode','tax_refund'],['--tax-mode',''],['--taxMode','TAX_REFUND'], ['--tax-mode','TAX_REFUND','--tax-mode','BUY_DECLARATION']])assert.throws(()=>parseCoreArguments(args)) }) + +test('explicit single signature mode runs only that mode and reports the narrowed scope', async()=>{ + const cfg=parseCoreArguments(['--execute','--environment','TEST','--run-id','single-mode','--signature-mode','ROOT_FIRST']) + assert.deepEqual(executionPlan(cfg).modes,['ROOT_FIRST']) + assert.doesNotMatch(executionPlan(cfg).reports.join(','),/child_first/) + const seen=[] + const summary=await runFbaCore(cfg,{env:env(),readers:readers(),outputRoot:root(),createContext:mode=>({context:{},async run(){seen.push(mode);return pass()}})}) + assert.equal(summary.success,true);assert.deepEqual(seen,['ROOT_FIRST']);assert.equal(summary.modes.length,1) + for(const args of [['--signature-mode','ALL'],['--signature-mode','root_first'],['--signature-mode','ROOT_FIRST','--signature-mode','CHILD_FIRST']])assert.throws(()=>parseCoreArguments(args)) +}) test('both tax modes pass unchanged to all three existing contexts and persist on all report surfaces',async()=>{ for(const taxMode of ['BUY_DECLARATION','TAX_REFUND']){ const outputRoot=root(),cfg={...config(),taxMode},seen=[] @@ -119,3 +129,32 @@ test('reports redact dynamically created account secrets and environment/token v assert.equal(value.password,'[REDACTED]');assert.equal(value.pdaToken,'[REDACTED]') assert.equal(value.message,'failure [REDACTED]');assert.equal(value.nested.value,'safe') }) + +test('delivery fault requires one explicit mode and verified proof plus restoration, even after caught run failures', async()=>{ + assert.throws(()=>parseCoreArguments(['--plan','--verify-delivery-event-recovery'])) + const cfg=parseCoreArguments(['--execute','--environment','TEST','--run-id','fault-one','--signature-mode','ROOT_FIRST','--verify-delivery-event-recovery']) + assert.equal(executionPlan(cfg).verifyDeliveryEventRecovery,true) + for(const failure of ['none','context','proof','restore']) { + const outputRoot=root();let cleanups=0,prepares=0 + const r={...readers(),readFleetInventory:async scope=>({scope,balances:[],positions:[],ledger:[]})} + const summary=await runFbaCore(cfg,{env:{...env(),CHONGCHONG_ENV:'test'},readers:r,outputRoot, + faultFactory:()=>({prepare:async()=>{prepares++},cleanup:async()=>{ + cleanups++;if(failure==='restore')throw Error('private failure') + return {restored:true,healthy:true,proxyStopped:true} + }}),createContext:()=>({context:{},createdResources:{},async run(){ + await this.context.f40DeliveryEventFaultExecutor.prepare({marker:'F40',customerId:'90',deliveryId:'91'}) + await this.context.f40ReadOnlyInventoryEvidence({marker:'F40',customerId:'90',warehouseId:'92'}) + if(failure==='context')throw Error('private failure') + const report=pass() + if(failure!=='proof')report.createdResources.phase5RootFulfillment.deliveryEventRecovery={status:'VERIFIED'} + return report + }})}) + assert.equal(summary.success,failure==='none') + assert.equal(cleanups,1);assert.equal(prepares,1) + const evidencePath=join(outputRoot,cfg.outputDir,'evidence') + assert.ok(existsSync(join(evidencePath,'root_first-delivery-scope.json'))) + assert.ok(existsSync(join(evidencePath,'root_first-0001-inventory.json'))) + const report=JSON.parse(readFileSync(join(outputRoot,cfg.outputDir,'root_first.json'),'utf8')) + if(failure==='restore')assert.equal(report.deliveryRestorationFailed,true) + } +}) diff --git a/scripts/test-delivery-fault-proof.mjs b/scripts/test-delivery-fault-proof.mjs new file mode 100644 index 000000000..5ceb5a71f --- /dev/null +++ b/scripts/test-delivery-fault-proof.mjs @@ -0,0 +1,76 @@ +import { createHash } from 'node:crypto' +import { isDeepStrictEqual } from 'node:util' + +export const deliveryFaultScopeKeys = ['marker','customerId','deliveryId','sourceInboundId','sourceOrderId','rootOrderId'] +const canonical = value => Array.isArray(value) ? value.map(canonical) : value && typeof value === 'object' + ? Object.fromEntries(Object.keys(value).sort().map(key => [key, canonical(value[key])])) : value +export function assertTestDeliveryFaultProof(proof, scope) { + const need = (value, code) => { if (!value) throw Error('Invalid delivery fault proof: ' + code) } + need(proof?.readOnly === true && deliveryFaultScopeKeys.every(key => proof[key] === scope[key]), 'SCOPE') + const row=proof.producer, before=proof.beforeRetry, after=proof.afterRetry, env=row?.envelope, payload=env?.payload + need(row && /^[1-9][0-9]{0,19}$/.test(row.entryId) && /^(?:[0-9a-fA-F-]{36}|[0-9A-HJKMNP-TV-Z]{26})$/.test(row.eventId), 'IDENTITY') + need(row.eventType==='OverseasDeliveryCompleted' && row.customerId===scope.customerId && row.aggregateType==='INBOUND_ORDER' + && row.aggregateId===scope.sourceInboundId && payload?.deliveryOrderId===scope.deliveryId + && payload.sourceInboundId===scope.sourceInboundId && payload.sourceOrderId===scope.sourceOrderId + && payload.rootOrderId===scope.rootOrderId && payload.deliveryNo && payload.inboundCode && payload.sourceOrderNo + && payload.fbaWarehouseCode && payload.lastlegOrderId && payload.lastlegOrderNo && Number.isFinite(Date.parse(payload.completedAt)) + && row.eventKey===`OverseasDeliveryCompleted:${payload.deliveryNo}:${scope.sourceInboundId}`, 'PRODUCER') + need(env.eventId===row.eventId && env.eventType===row.eventType && env.tenantId===scope.customerId + && Number(env.eventVersion)===1 && env.aggregateType==='INBOUND_ORDER' && env.aggregateId===scope.sourceInboundId + && Number(env.aggregateVersion)>0, 'ENVELOPE') + const hash='sha256:'+createHash('sha256').update(JSON.stringify(canonical(payload))).digest('hex') + need(row.payloadHash===hash && env.payloadHash===hash, 'HASH') + need(before?.publishStatus==='DEAD' && before.failedAttempts===12 && row.publishStatus==='PUBLISHED' + && after?.publishStatus==='PUBLISHED' && [before,after].every(value => value.entryId===row.entryId + && value.eventId===row.eventId && value.payloadHash===hash && isDeepStrictEqual(value.envelope,env)), 'ORIGINAL_RETRY') + const inbox=proof.consumer + need(inbox?.count===1 && inbox.eventId===row.eventId && inbox.eventType===row.eventType && inbox.aggregateId===scope.sourceInboundId + && inbox.status==='PROCESSED' && inbox.payloadHash===hash && isDeepStrictEqual(inbox.payload,payload), 'INBOX') + const attempts=proof.attempts, matches=proof.proxyRecords?.filter(value=>value.matched===true) + need(Array.isArray(attempts) && attempts.length===13 && new Set(attempts.map(value=>value.attemptId)).size===13 + && attempts.every((value,index)=>value.entryId===row.entryId && value.eventId===row.eventId && value.serviceName==='lastleg' + && value.outcome===(index<11?'RETRY':index===11?'DEAD':'DUPLICATE')), 'ATTEMPTS') + need(matches?.length===13 && matches.every((value,index)=>value.eventId===row.eventId && value.payloadHash===hash + && value.attempt===index+1 && value.outcome===(index===0?'APPLIED':'DUPLICATE') + && (index<12?value.errorCode==='ACK_LOSS_INJECTED':value.errorCode===null)), 'REAL_ACK_LOSS') + const audit=proof.retryAudit + need(proof.retryAuditCount===1 && audit?.requestId===proof.retryRequestId && audit.entryId===row.entryId + && audit.eventId===row.eventId && audit.payloadHash===hash && audit.previousFailures===12, 'RETRY_AUDIT') + const facts=before.business + need(facts && isDeepStrictEqual(facts,after.business), 'BUSINESS_UNCHANGED') + for (const key of ['inbound','inboundHistory','inbox','order','orderHistory','completionEvent']) need(Array.isArray(facts[key]), 'BUSINESS_SHAPE') + need(facts.inbox.length===1 && facts.inbox[0].eventId===row.eventId && facts.inbox[0].status==='PROCESSED' + && facts.inbox[0].payloadHash===hash && facts.inbox[0].processedAt && isDeepStrictEqual(facts.inbox[0].payload,payload), 'PERSISTED_INBOX') + const inbounds=facts.inbound.filter(value=>value.id===scope.sourceInboundId) + need(inbounds.length===1 && inbounds[0].customerId===scope.customerId && inbounds[0].orderId===scope.sourceOrderId + && inbounds[0].orderNo===payload.sourceOrderNo && inbounds[0].inboundCode===payload.inboundCode + && inbounds[0].processStatus==='DELIVERED' && Number.isInteger(inbounds[0].version), 'INBOUND') + need(facts.inboundHistory.filter(value=>value.inboundId===scope.sourceInboundId && value.action==='COMPLETE_DELIVERY' + && value.fromStatus==='LOADED' && value.toStatus==='DELIVERED').length===1, 'INBOUND_HISTORY') + const instant=payload.completedAt.match(/^(\d{4}-\d{2}-\d{2}T\d{2}:\d{2}:\d{2})(?:\.(\d{1,9}))?Z$/) + const completedMillis=instant?Date.parse(instant[1]+'Z')+Math.floor((Number((instant[2]||'').padEnd(9,'0'))+500000)/1000000):NaN + const storedMillis=value=>Date.parse(String(value).replace(' ','T').replace(/Z$/,'')+'Z') + need(facts.inboundHistory.filter(value=>value.action==='COMPLETE_DELIVERY') + .every(value=>storedMillis(value.occurredAt)===completedMillis), 'COMPLETION_TIME') + need(facts.order.length>0 && new Set(facts.order.map(value=>value.id)).size===facts.order.length + && facts.order.every(value=>value.customerId===scope.customerId && value.rootOrderId===scope.rootOrderId && Number.isInteger(value.version)), 'ORDER_SCOPE') + const source=facts.order.find(value=>value.id===scope.sourceOrderId), root=facts.order.find(value=>value.id===scope.rootOrderId) + need(source && root && root.orderRole==='MASTER' && source.orderNo===payload.sourceOrderNo && source.inboundCode===payload.inboundCode, 'ORDER_IDENTITY') + const historyCount=(id,action)=>facts.orderHistory.filter(value=>value.orderId===id && value.action===action).length + need(historyCount(source.id,'OVERSEAS_DELIVERED')===1, 'SOURCE_SIGNED') + need(facts.orderHistory.filter(value=>value.orderId===source.id && value.action==='OVERSEAS_DELIVERED') + .every(value=>storedMillis(value.occurredAt)===completedMillis), 'SOURCE_SIGNED_TIME') + const rootCanComplete=historyCount(root.id,'OVERSEAS_DELIVERED')===1 + && facts.order.filter(value=>value.id!==root.id && value.status!=='CANCELLED').every(value=>value.status==='COMPLETED') + need(rootCanComplete ? root.status==='COMPLETED' : ['APPROVED','ACCEPTED','PROCESSING'].includes(root.status), 'ROOT_GATE') + need(source.id===root.id || source.status==='COMPLETED', 'SOURCE_COMPLETION') + need(source.status==='COMPLETED'?inbounds[0].orderStatusSnapshot==='COMPLETED':inbounds[0].orderStatusSnapshot!=='COMPLETED', 'INBOUND_ORDER_SNAPSHOT') + for (const order of facts.order) { + const events=facts.completionEvent.filter(value=>value.orderId===order.id) + need(events.length===(order.status==='COMPLETED'?1:0) && events.every(value=>value.status==='PUBLISHED' + && value.eventType==='FIRST_LEG_ORDER_FULFILLMENT_COMPLETED'), 'COMPLETION_EVENTS') + if(order.status==='COMPLETED') need(order.completedAt && historyCount(order.id,order.id===root.id?'FULFILLMENT_COMPLETED':'OVERSEAS_DELIVERED')===1, 'COMPLETION_HISTORY') + } + need(facts.completionEvent.every(value=>facts.order.some(order=>order.id===value.orderId)), 'FOREIGN_EVENT') + return proof +} diff --git a/scripts/test-event-fault-executor.mjs b/scripts/test-event-fault-executor.mjs index d808040e1..2f587be62 100644 --- a/scripts/test-event-fault-executor.mjs +++ b/scripts/test-event-fault-executor.mjs @@ -3,6 +3,8 @@ import { readFileSync } from 'node:fs' import { createHash } from 'node:crypto' import { isDeepStrictEqual } from 'node:util' +import { assertTestDeliveryFaultProof, deliveryFaultScopeKeys } from './test-delivery-fault-proof.mjs' +export { assertTestDeliveryFaultProof } from './test-delivery-fault-proof.mjs' import { validateEvidenceConfig } from './fba-core-regression-evidence.mjs' const runtime = readFileSync(new URL('./test-event-fault-runtime.py', import.meta.url), 'utf8') @@ -17,11 +19,11 @@ export function renderTestUnloadFaultAction(action, scope, agent) { return renderAction(action,scope,agent,'UNLOAD') } function renderAction(action,scope,agent,direction) { - const keys=direction==='UNLOAD'?['marker','customerId','containerBookingId']:['marker','customerId','loadingPlanId','containerBookingId'] + const keys=direction==='DELIVERY'?deliveryFaultScopeKeys:direction==='UNLOAD'?['marker','customerId','containerBookingId']:['marker','customerId','loadingPlanId','containerBookingId'] if (!['prepare', 'activate', 'proxy-ready', 'health', 'status', 'recover', 'proof', 'restore', 'finish'].includes(action) || !/^[A-Za-z0-9_.-]{1,128}$/.test(agent) || !/^[A-Za-z0-9_-]{1,96}$/.test(scope?.marker || '') || keys.filter(key=>key!=='marker').some(key => !/^[1-9][0-9]{0,19}$/.test(scope?.[key] || '')) - || (direction==='UNLOAD'&&Object.keys(scope).some(key=>!keys.includes(key)))) { + || (direction!=='LOADING'&&Object.keys(scope).some(key=>!keys.includes(key)))) { throw Error('Invalid TEST fault action or scope') } const config = encode(JSON.stringify({ action, scope, direction, ...(action === 'prepare' ? { proxySource } : {}) })) @@ -32,6 +34,9 @@ function renderAction(action,scope,agent,direction) { export function createTestFaultExecutor(env, fetcher = globalThis.fetch) { return createExecutor(env,fetcher,'LOADING') } +export function createTestDeliveryFaultExecutor(env, fetcher = globalThis.fetch) { + return createExecutor(env,fetcher,'DELIVERY') +} export function createTestUnloadFaultExecutor(env, fetcher = globalThis.fetch) { return createExecutor(env,fetcher,'UNLOAD') } @@ -117,23 +122,23 @@ function createExecutor(env,fetcher,direction) { await action('restore') await wait('health', row => row.healthy) restoration=await action('finish') - if(direction==='UNLOAD'&&!(/^sha256:[0-9a-f]{64}$/.test(preparedEvidence?.originalImageId||'') + if(direction!=='LOADING'&&!(/^sha256:[0-9a-f]{64}$/.test(preparedEvidence?.originalImageId||'') &&/^[0-9a-f]{64}$/.test(preparedEvidence?.originalFingerprint||'') &&restoration?.restored===true&&restoration.healthy===true&&restoration.proxyStopped===true &&restoration.producerName==='firstleg-test-lastleg'&&restoration.originalImageId===preparedEvidence?.originalImageId &&restoration.currentImageId===preparedEvidence?.originalImageId&&restoration.originalFingerprint===preparedEvidence?.originalFingerprint - &&restoration.currentFingerprint===preparedEvidence?.originalFingerprint))throw Error('Unload restoration evidence mismatch') + &&restoration.currentFingerprint===preparedEvidence?.originalFingerprint))throw Error((direction==='DELIVERY'?'Delivery':'Unload')+' restoration evidence mismatch') finished = true return restoration } return { async prepare(next) { if (scope) { - if(direction==='UNLOAD'&&!isDeepStrictEqual(scope,next))throw Error('TEST unload fault scope drift') + if(direction!=='LOADING'&&!isDeepStrictEqual(scope,next))throw Error('TEST unload fault scope drift') return // Pin only the first actual current-run plan; other events pass through. } - if(direction==='UNLOAD')renderAction('prepare',next,env.FBA_CORE_JENKINS_AGENT,direction) - scope = direction==='UNLOAD'?{...next}:{ marker: next.marker, customerId: next.customerId, + if(direction!=='LOADING')renderAction('prepare',next,env.FBA_CORE_JENKINS_AGENT,direction) + scope = direction!=='LOADING'?{...next}:{ marker: next.marker, customerId: next.customerId, loadingPlanId: next.loadingPlanId, containerBookingId: next.containerBookingId } preparedEvidence=await action('prepare') await wait('proxy-ready', row => row.ready, 10, true) @@ -141,16 +146,17 @@ function createExecutor(env,fetcher,direction) { await wait('health', row => row.healthy) }, async readProof(expected) { - if (!scope || (direction==='UNLOAD'?!isDeepStrictEqual(scope,expected):Object.keys(scope).some(key => expected?.[key] !== scope[key]))) throw Error('TEST fault scope drift') + if (!scope || (direction!=='LOADING'?!isDeepStrictEqual(scope,expected):Object.keys(scope).some(key => expected?.[key] !== scope[key]))) throw Error('TEST fault scope drift') try { await wait('status', row => row.producer?.publishStatus === 'DEAD' && row.producer.failedAttempts === 12, 180) await action('recover') await wait('status', row => row.producer?.publishStatus === 'PUBLISHED', 90) const proof = await action('proof') if(direction==='UNLOAD')assertTestUnloadFaultProof(proof,scope) + if(direction==='DELIVERY')assertTestDeliveryFaultProof(proof,scope) await cleanup() - return { ...proof, restored: true, ...(direction==='UNLOAD'?{restoration}:{}) } - } finally { if(direction==='UNLOAD')await cleanup() } + return { ...proof, restored: true, ...(direction!=='LOADING'?{restoration}:{}) } + } finally { if(direction!=='LOADING')await cleanup() } }, cleanup, } diff --git a/scripts/test-event-fault-executor.test.mjs b/scripts/test-event-fault-executor.test.mjs index 3da9b468c..5b0df0ed4 100644 --- a/scripts/test-event-fault-executor.test.mjs +++ b/scripts/test-event-fault-executor.test.mjs @@ -2,7 +2,8 @@ import assert from 'node:assert/strict' import test from 'node:test' import { createHash } from 'node:crypto' import { spawnSync } from 'node:child_process' -import { createTestFaultExecutor, createTestUnloadFaultExecutor, assertTestUnloadFaultProof, renderTestFaultAction } from './test-event-fault-executor.mjs' +import { createTestFaultExecutor, createTestUnloadFaultExecutor, createTestDeliveryFaultExecutor, assertTestDeliveryFaultProof, assertTestUnloadFaultProof, renderTestFaultAction } from './test-event-fault-executor.mjs' +import { withDeliveryEventRecovery } from './admin-test-env-modules/phase5-root-fulfillment.mjs' const scope = { marker: 'TEST-ROOT', customerId: '7', loadingPlanId: '99', containerBookingId: '66' } const environment = { CHONGCHONG_ENV: 'test', JENKINS_URL: 'https://jenkins.example.invalid/', @@ -219,3 +220,137 @@ test('failed activation retains task ownership for explicit restoration, never r assert.equal(calls.filter(name => name === 'activate').length, 1) assert.deepEqual(calls.slice(-3), ['restore', 'health', 'finish']) }) + +const deliveryScope={marker:'TEST-DELIVERY',customerId:'7',deliveryId:'90',sourceInboundId:'200',sourceOrderId:'100',rootOrderId:'100'} +function deliveryProof(mode='ROOT_PENDING') { + const scope={...deliveryScope,...(mode==='CHILD_FIRST'?{sourceOrderId:'101',sourceInboundId:'201'}:{})} + const eventId='00000000-0000-4000-8000-000000000001', entryId='123' + const payload={deliveryOrderId:'90',deliveryNo:'DL-OFFLINE',lastlegOrderId:'300',lastlegOrderNo:'LL-OFFLINE', + sourceInboundId:scope.sourceInboundId,inboundCode:'IN-OFFLINE',sourceOrderId:scope.sourceOrderId, + sourceOrderNo:'FL-OFFLINE',rootOrderId:'100',fbaWarehouseCode:'ABE4',completedAt:'2026-10-03T01:00:00Z'} + const payloadHash='sha256:'+createHash('sha256').update(JSON.stringify(payload,Object.keys(payload).sort())).digest('hex') + const envelope={eventId,eventType:'OverseasDeliveryCompleted',eventVersion:1,tenantId:'7',aggregateType:'INBOUND_ORDER',aggregateId:scope.sourceInboundId,aggregateVersion:5,payloadHash,payload} + const producer={entryId,eventId,eventType:envelope.eventType,customerId:'7',aggregateId:scope.sourceInboundId,aggregateType:'INBOUND_ORDER',eventKey:`OverseasDeliveryCompleted:DL-OFFLINE:${scope.sourceInboundId}`,envelope,payloadHash,publishStatus:'PUBLISHED'} + const ownRoot=mode!=='CHILD_FIRST', childComplete=mode!=='ROOT_PENDING', rootComplete=mode==='ALL_COMPLETE' + const orders=['100','101'].map(id=>({id,rootOrderId:'100',customerId:'7',orderRole:id==='100'?'MASTER':'CHILD', + orderNo:id===scope.sourceOrderId?'FL-OFFLINE':'OTHER',inboundCode:id===scope.sourceOrderId?'IN-OFFLINE':'OTHER', + status:(id==='100'?rootComplete:childComplete)?'COMPLETED':'APPROVED',version:3, + completedAt:(id==='100'?rootComplete:childComplete)?'2026-10-03 01:00:00':null})) + const consumer={eventId,eventType:envelope.eventType,aggregateId:scope.sourceInboundId,payloadHash,payload,status:'PROCESSED'} + const business={inbound:[{id:scope.sourceInboundId,customerId:'7',orderId:scope.sourceOrderId,orderNo:'FL-OFFLINE',inboundCode:'IN-OFFLINE',processStatus:'DELIVERED',orderStatusSnapshot:orders.find(o=>o.id===scope.sourceOrderId).status,version:2}], + inboundHistory:[{id:'41',inboundId:scope.sourceInboundId,action:'COMPLETE_DELIVERY',fromStatus:'LOADED',toStatus:'DELIVERED',occurredAt:'2026-10-03 01:00:00'}], + inbox:[{...consumer,processedAt:'2026-10-03 01:00:01'}],order:orders, + orderHistory:[...(ownRoot?[{id:'51',orderId:'100',action:'OVERSEAS_DELIVERED',occurredAt:'2026-10-03 01:00:00'}]:[]),...(childComplete?[{id:'52',orderId:'101',action:'OVERSEAS_DELIVERED',occurredAt:'2026-10-03 01:00:00'}]:[]),...(rootComplete?[{id:'53',orderId:'100',action:'FULFILLMENT_COMPLETED'}]:[])], + completionEvent:orders.filter(o=>o.status==='COMPLETED').map((o,i)=>({id:String(60+i),orderId:o.id,eventType:'FIRST_LEG_ORDER_FULFILLMENT_COMPLETED',status:'PUBLISHED',payload:{orderId:o.id}}))} + const old=unloadProof() + return JSON.parse(JSON.stringify({...scope,readOnly:true,producer,consumer:{...consumer,count:1}, + beforeRetry:{...producer,publishStatus:'DEAD',failedAttempts:12,business},afterRetry:{...producer,business}, + attempts:old.attempts,proxyRecords:old.proxyRecords.map(p=>({...p,payloadHash})),retryAudit:{...old.retryAudit,payloadHash},retryAuditCount:1,retryRequestId:old.retryRequestId})) +} +function deliveryTransport(calls, proof=deliveryProof(), finish={}) { + let recovered=false + return async(_url,options)=>{ + const {action,direction,scope}=requestedAction(options) + assert.equal(direction,'DELIVERY');assert.deepEqual(scope,deliveryScope);calls.push(action) + if(action==='recover')recovered=true + const data=action==='prepare'?{originalImageId,originalFingerprint}:action==='proxy-ready'?{ready:true}:action==='health'?{healthy:true} + :action==='status'?{producer:{publishStatus:recovered?'PUBLISHED':'DEAD',failedAttempts:12}}:action==='proof'?proof + :action==='finish'?{restored:true,healthy:true,proxyStopped:true,producerName:'firstleg-test-lastleg',originalImageId,currentImageId:originalImageId,originalFingerprint,currentFingerprint:originalFingerprint,...finish}:{} + return {ok:true,text:async()=> 'TEST_EVENT_JSON='+JSON.stringify({ok:true,data})} + } +} +test('DELIVERY has a separate strict scope and restores before accepting original event proof',async()=>{ + const calls=[],executor=createTestDeliveryFaultExecutor(environment,deliveryTransport(calls)) + await assert.rejects(executor.prepare({...deliveryScope,eventType:'OverseasContainerUnloaded'}),/Invalid/) + await executor.prepare(deliveryScope) + await assert.rejects(executor.prepare({...deliveryScope,deliveryId:'91'}),/scope drift/) + const proof=await executor.readProof(deliveryScope) + assert.equal(proof.restored,true);assert.equal(proof.restoration.currentImageId,originalImageId) + assert.deepEqual(calls,['prepare','proxy-ready','activate','health','status','recover','status','proof','restore','health','finish']) +}) +test('DELIVERY proof distinguishes root pending, child first and all completed with immutable original evidence',()=>{ + for(const mode of ['ROOT_PENDING','CHILD_FIRST','ALL_COMPLETE']) { + const p=deliveryProof(mode),scope={...deliveryScope,sourceOrderId:p.sourceOrderId,sourceInboundId:p.sourceInboundId} + assertTestDeliveryFaultProof(p,scope) + } + for(const mutate of [p=>p.producer.envelope.payload.deliveryId='90',p=>delete p.producer.envelope.payload.deliveryOrderId, + p=>p.producer.customerId='0',p=>p.producer.eventKey='wrong',p=>p.beforeRetry.entryId='124', + p=>p.consumer.count=2,p=>p.attempts.pop(),p=>p.proxyRecords[0].outcome='DUPLICATE', + p=>p.retryAuditCount=2,p=>p.afterRetry.business.inbound[0].version++,p=>p.afterRetry.business.order[0].version++, + p=>p.beforeRetry.business.inbox[0].status='RECEIVED',p=>p.beforeRetry.business.inboundHistory.push({...p.beforeRetry.business.inboundHistory[0]}), + p=>{p.beforeRetry.business.order[0].status=p.afterRetry.business.order[0].status='COMPLETED'}, + p=>p.beforeRetry.business.completionEvent.push({orderId:'100',status:'PUBLISHED'})]) { + const proof=deliveryProof();mutate(proof);assert.throws(()=>assertTestDeliveryFaultProof(proof,deliveryScope),/Invalid delivery/) + } +}) +test('DELIVERY evidence failure still restores and restoration drift is a blocking failure',async()=>{ + const p=deliveryProof();p.consumer.count=0 + const calls=[],executor=createTestDeliveryFaultExecutor(environment,deliveryTransport(calls,p)) + await executor.prepare(deliveryScope);await assert.rejects(executor.readProof(deliveryScope),/INBOX/) + assert.deepEqual(calls.slice(-3),['restore','health','finish']) + const broken=createTestDeliveryFaultExecutor(environment,deliveryTransport([],deliveryProof(),{proxyStopped:false})) + await broken.prepare(deliveryScope);await assert.rejects(broken.readProof(deliveryScope),/Delivery restoration/) +}) +test('F40 hook faults only first signature, captures whole inventory and always cleans failed signing',async()=>{ + const calls=[],restoration={restored:true,healthy:true,proxyStopped:true} + const inventory={scope:{...deliveryScope,warehouseId:'9',coverage:'CUSTOMER_WAREHOUSE_COMPLETE'},balances:[],positions:[],ledger:[]} + const context={context:{f40DeliveryEventFaultExecutor:{prepare:async s=>{assert.deepEqual(s,deliveryScope);calls.push('prepare')}, + readProof:async()=>{calls.push('proof');return {...deliveryProof(),restored:true,restoration}},cleanup:async()=>{calls.push('cleanup');return restoration}}, + f40ReadOnlyInventoryEvidence:async s=>{assert.equal(s.warehouseId,'9');calls.push('inventory');return structuredClone(inventory)}},createdResources:{phase5RootFulfillment:{}}} + assert.equal(await withDeliveryEventRecovery(context,deliveryScope,'9',0,async()=>{calls.push('sign');return 'signed'}),'signed') + assert.deepEqual(calls,['inventory','prepare','sign','proof','inventory','cleanup']) + assert.equal(context.createdResources.phase5RootFulfillment.deliveryEventRecovery.status,'VERIFIED') + await withDeliveryEventRecovery(context,deliveryScope,'9',1,async()=>calls.push('second')) + assert.equal(calls.at(-1),'second') + context.createdResources.phase5RootFulfillment={} + await assert.rejects(withDeliveryEventRecovery(context,deliveryScope,'9',0,async()=>{throw Error('sign failed')}),/sign failed/) + assert.equal(calls.at(-1),'cleanup') + assert.equal(context.createdResources.phase5RootFulfillment.deliveryEventRecovery.restored,true) +}) +test('Python DELIVERY SQL uses actual payload, original Lastleg retry, and Order database without inventory queries',()=>{ + const result=spawnSync('python',['-c',String.raw` +import importlib.util,sys,ast,pathlib,copy +from unittest.mock import patch,MagicMock,mock_open +sys.dont_write_bytecode=True +p=pathlib.Path('scripts/test-event-fault-runtime.py');ast.parse(p.read_text(encoding='utf-8')) +spec=importlib.util.spec_from_file_location('runtime',p);m=importlib.util.module_from_spec(spec);spec.loader.exec_module(m) +s={'marker':'TEST-DELIVERY','customerId':'7','deliveryId':'90','sourceInboundId':'200','sourceOrderId':'100','rootOrderId':'100'} +assert m.scope_directory(s,'DELIVERY')!=m.scope_directory({'marker':s['marker'],'customerId':'7','containerBookingId':'200'},'UNLOAD') +assert m.producer_name({'direction':'DELIVERY'})=='firstleg-test-lastleg' +assert m.target_setting({'direction':'DELIVERY'})==('FIRSTLEG_SERVICE_URL','http://firstleg:9202') +queries=[] +with patch.object(m,'inspect',side_effect=lambda name:{'Name':'/'+name}),patch.object(m,'require_test'),patch.object(m,'read_sql',side_effect=lambda container,sql:queries.append((container['Name'],sql)) or []): + assert m.producer(s,'DELIVERY') is None + m.delivery_facts(s,'00000000-0000-4000-8000-000000000001') +assert [q[0] for q in queries]==['/firstleg-test-lastleg','/firstleg-test-business','/firstleg-test-order'] +assert "$.payload.deliveryOrderId" in queries[0][1] and "customer_id='7'" in queries[0][1] +assert 'fl_inbound_status_history' in queries[1][1] and 'fl_container_event_inbox' in queries[1][1] +assert 'ord_order_status_history' in queries[2][1] and 'FIRST_LEG_ORDER_FULFILLMENT_COMPLETED' in queries[2][1] +assert all('inventory' not in q[1] for q in queries) +original={'Name':'/firstleg-test-lastleg','Id':'original-container','Image':'sha256:'+'a'*64, + 'Config':{'Image':'lastleg:test','Env':['FIRSTLEG_SERVICE_URL=http://firstleg:9202'],'Labels':{'com.docker.compose.project':'firstleg-test'}}, + 'HostConfig':{},'Mounts':[],'NetworkSettings':{'Networks':{'firstleg-test-network':{'IPAddress':'172.20.0.3','Aliases':['lastleg-service']}}}, + 'State':{'Running':True,'Health':{'Status':'healthy'}}} +consumer=copy.deepcopy(original);consumer['Name']='/firstleg-test-business' +with patch.object(m,'producer',return_value=None),patch.object(m,'inspect',side_effect=lambda name:original if name=='firstleg-test-lastleg' else consumer),patch.object(m.Path,'mkdir'),patch.object(m.Path,'write_bytes'),patch('builtins.open',mock_open()),patch.object(m.socket,'socket') as sock,patch.object(m.subprocess,'Popen') as process,patch.object(m,'private_write'): + sock.return_value.getsockname.return_value=('127.0.0.1',18001) + m.prepare(s,pathlib.Path('/offline'),{'direction':'DELIVERY','proxySource':''}) + args=process.call_args.args[0] + assert args[args.index('--delivery-id')+1]=='90' and args[args.index('--event-type')+1]=='OverseasDeliveryCompleted' + assert '--container-booking-id' not in args and args[args.index('--upstream')+1].endswith(':9202/internal/firstleg/v1/container-events') +state={'direction':'DELIVERY','restored':False,'prepared':True,'image':'lastleg:test','originalId':'original-container', + 'originalImageId':original['Image'],'originalFingerprint':m.fingerprint(original),'originalBody':m.recreation(original,m.env_of(original),full=True),'proxyPort':18001} +restored=copy.deepcopy(original);restored['Id']='restored-new-id' +with patch.object(m,'control',side_effect=OSError('proxy already stopped')),patch.object(m,'inspect',return_value=restored),patch.object(m,'private_write'),patch.object(m,'replace') as replace,patch.object(m,'docker') as docker: + assert m.restore(pathlib.Path('/offline'),state)=={'restoreStarted':True} + assert state['prepared'] is False + replace.assert_not_called();docker.assert_not_called() +state['prepared']=True +restored['Config']['Env']=['FIRSTLEG_SERVICE_URL=http://firstleg:9202','UNEXPECTED=1'] +with patch.object(m,'control'),patch.object(m,'inspect',return_value=restored),patch.object(m,'replace') as replace: + try: m.restore(pathlib.Path('/offline'),state);assert False + except RuntimeError as error: assert str(error)=='CONCURRENT_PRODUCER_CHANGE' + replace.assert_not_called() +`],{encoding:'utf8',cwd:new URL('..',import.meta.url),timeout:10000}) + assert.equal(result.status,0,result.stderr||result.error?.message) +}) diff --git a/scripts/test-event-fault-runtime.py b/scripts/test-event-fault-runtime.py index 6bde84707..7a4662f5d 100644 --- a/scripts/test-event-fault-runtime.py +++ b/scripts/test-event-fault-runtime.py @@ -103,18 +103,18 @@ def private_write(path, value): def scope_directory(scope, direction='LOADING'): - if direction not in ('LOADING', 'UNLOAD'): + if direction not in ('LOADING', 'UNLOAD', 'DELIVERY'): raise RuntimeError('INVALID_DIRECTION') marker = scope.get('marker', '') if not re.fullmatch(r'[A-Za-z0-9_-]{1,96}', marker): raise RuntimeError('INVALID_SCOPE') - keys = ('customerId', 'containerBookingId') if direction == 'UNLOAD' else ('customerId', 'loadingPlanId', 'containerBookingId') - if direction == 'UNLOAD' and set(scope) != {'marker', *keys}: + keys = ('customerId', 'deliveryId', 'sourceInboundId', 'sourceOrderId', 'rootOrderId') if direction == 'DELIVERY' else ('customerId', 'containerBookingId') if direction == 'UNLOAD' else ('customerId', 'loadingPlanId', 'containerBookingId') + if direction != 'LOADING' and set(scope) != {'marker', *keys}: raise RuntimeError('INVALID_SCOPE') for key in keys: if not re.fullmatch(r'[1-9][0-9]{0,19}', scope.get(key, '')): raise RuntimeError('INVALID_SCOPE') - identity = 'UNLOAD:' + marker if direction == 'UNLOAD' else marker + identity = direction + ':' + marker if direction != 'LOADING' else marker return Path(tempfile.gettempdir()) / ('chongchong-event-' + hashlib.sha256(identity.encode()).hexdigest()[:24]) @@ -122,12 +122,20 @@ def unload(state): return state.get('direction', 'LOADING') == 'UNLOAD' +def delivery(state): + return state.get('direction') == 'DELIVERY' + + +def reverse_delivery(state): + return state.get('direction') in ('UNLOAD', 'DELIVERY') + + def producer_name(state): - return 'firstleg-test-lastleg' if unload(state) else 'firstleg-test-business' + return 'firstleg-test-lastleg' if reverse_delivery(state) else 'firstleg-test-business' def target_setting(state): - return ('FIRSTLEG_SERVICE_URL', 'http://firstleg:9202') if unload(state) else ('LASTLEG_SERVICE_URL', 'http://lastleg-service:9303') + return ('FIRSTLEG_SERVICE_URL', 'http://firstleg:9202') if reverse_delivery(state) else ('LASTLEG_SERVICE_URL', 'http://lastleg-service:9303') def fingerprint(container): @@ -159,7 +167,7 @@ def read_sql(container, sql): if re.search(r'\b(INSERT|UPDATE|DELETE|ALTER|DROP|TRUNCATE|CALL|INTO|OUTFILE)\b', sql, re.I): raise RuntimeError('NON_READ_ONLY_SQL') e = env_of(container) - prefix = 'FIRSTLEG' if container['Name'] == '/firstleg-test-business' else 'LASTLEG' + prefix = {'/firstleg-test-business': 'FIRSTLEG', '/firstleg-test-lastleg': 'LASTLEG', '/firstleg-test-order': 'ORDER'}[container['Name']] url = urllib.parse.urlsplit(e[prefix + '_DB_URL'].removeprefix('jdbc:')) host = '127.0.0.1' if url.hostname == 'host.docker.internal' else url.hostname transaction = 'SET SESSION TRANSACTION ISOLATION LEVEL REPEATABLE READ; START TRANSACTION WITH CONSISTENT SNAPSHOT, READ ONLY;\n' + sql + '\nCOMMIT;' @@ -174,7 +182,10 @@ def read_sql(container, sql): def producer(scope, direction='LOADING'): - if direction == 'UNLOAD': + if direction == 'DELIVERY': + rows = read_sql(inspect('firstleg-test-lastleg'), + "SELECT JSON_OBJECT('entryId',CAST(id AS CHAR),'eventId',JSON_UNQUOTE(JSON_EXTRACT(payload_json,'$.eventId')),'payloadHash',JSON_UNQUOTE(JSON_EXTRACT(payload_json,'$.payloadHash')),'eventType',event_type,'customerId',customer_id,'aggregateId',aggregate_id,'aggregateType',aggregate_type,'eventKey',event_key,'envelope',JSON_EXTRACT(payload_json,'$'),'publishStatus',status,'failedAttempts',retry_count) FROM ll_event_outbox WHERE customer_id='" + scope['customerId'] + "' AND event_type='OverseasDeliveryCompleted' AND JSON_UNQUOTE(JSON_EXTRACT(payload_json,'$.payload.deliveryOrderId'))='" + scope['deliveryId'] + "';") + elif direction == 'UNLOAD': rows = read_sql(inspect('firstleg-test-lastleg'), "SELECT JSON_OBJECT('entryId',CAST(id AS CHAR),'eventId',JSON_UNQUOTE(JSON_EXTRACT(payload_json,'$.eventId')),'payloadHash',JSON_UNQUOTE(JSON_EXTRACT(payload_json,'$.payloadHash')),'eventType',event_type,'customerId',customer_id,'aggregateId',aggregate_id,'aggregateType',aggregate_type,'eventKey',event_key,'envelope',JSON_EXTRACT(payload_json,'$'),'publishStatus',status,'failedAttempts',retry_count) FROM ll_event_outbox WHERE customer_id='0' AND aggregate_id='" + scope['containerBookingId'] + "' AND event_type='OverseasContainerUnloaded' AND event_key='OverseasContainerUnloaded:" + scope['containerBookingId'] + "';") @@ -201,18 +212,50 @@ def business_ids(scope): return [row['id'] for row in rows] +def business_snapshot(scope, state, row): + if delivery(state): + return {'business': delivery_facts(scope, row['eventId'])} + return {'container': container_facts(scope)} if unload(state) else {'businessOrderIds': business_ids(scope)} + + +def delivery_facts(scope, event_id): + if not re.fullmatch(r'[0-9a-fA-F-]{36}|[0-9A-HJKMNP-TV-Z]{26}', event_id): + raise RuntimeError('INVALID_EVENT_IDENTITY') + customer, source, root = scope['customerId'], scope['sourceOrderId'], scope['rootOrderId'] + firstleg = inspect('firstleg-test-business') + order = inspect('firstleg-test-order') + require_test(firstleg, 'firstleg-test-business') + require_test(order, 'firstleg-test-order') + # Each service is read in its own consistent read-only transaction; no cross-schema join. + inbound_sql = "tenant_id=" + customer + " AND deleted=0 AND order_id IN (" + source + "," + root + ")" + first_rows = read_sql(firstleg, + "SELECT JSON_OBJECT('kind','inbound','id',CAST(id AS CHAR),'customerId',CAST(tenant_id AS CHAR),'orderId',CAST(order_id AS CHAR),'orderNo',order_no,'inboundCode',inbound_code,'processStatus',process_status,'orderStatusSnapshot',order_status_snapshot,'version',version,'updatedAt',CAST(updated_at AS CHAR)) FROM fl_inbound_order WHERE " + inbound_sql + " ORDER BY id;\n" + + "SELECT JSON_OBJECT('kind','inboundHistory','id',CAST(id AS CHAR),'inboundId',CAST(inbound_id AS CHAR),'action',action,'fromStatus',from_status,'toStatus',to_status,'occurredAt',CAST(occurred_at AS CHAR)) FROM fl_inbound_status_history WHERE tenant_id=" + customer + " AND inbound_id=" + scope['sourceInboundId'] + " ORDER BY id;\n" + + "SELECT JSON_OBJECT('kind','inbox','eventId',event_id,'eventType',event_type,'aggregateId',aggregate_id,'aggregateVersion',aggregate_version,'payloadHash',payload_hash,'payload',JSON_EXTRACT(payload_json,'$'),'status',status,'receivedAt',CAST(received_at AS CHAR),'processedAt',CAST(processed_at AS CHAR)) FROM fl_container_event_inbox WHERE event_id='" + event_id + "';") + tree = "tenant_id=" + customer + " AND deleted=0 AND (id=" + root + " OR root_order_id=" + root + ")" + ids = "SELECT id FROM ord_order WHERE " + tree + order_rows = read_sql(order, + "SELECT JSON_OBJECT('kind','order','id',CAST(id AS CHAR),'customerId',CAST(tenant_id AS CHAR),'rootOrderId',CAST(root_order_id AS CHAR),'orderRole',order_role,'orderNo',order_no,'inboundCode',inbound_code,'status',status,'version',version,'completedAt',CAST(completed_at AS CHAR),'updatedAt',CAST(updated_at AS CHAR)) FROM ord_order WHERE " + tree + " ORDER BY id;\n" + + "SELECT JSON_OBJECT('kind','orderHistory','id',CAST(id AS CHAR),'orderId',CAST(order_id AS CHAR),'action',action,'fromStatus',from_status,'toStatus',to_status,'occurredAt',CAST(occurred_at AS CHAR)) FROM ord_order_status_history WHERE tenant_id=" + customer + " AND order_id IN (" + ids + ") ORDER BY id;\n" + + "SELECT JSON_OBJECT('kind','completionEvent','id',CAST(id AS CHAR),'orderId',CAST(aggregate_id AS CHAR),'eventType',event_type,'status',publish_status,'payload',JSON_EXTRACT(payload_json,'$')) FROM ord_outbox_event WHERE tenant_id=" + customer + " AND aggregate_id IN (" + ids + ") AND event_type='FIRST_LEG_ORDER_FULFILLMENT_COMPLETED' ORDER BY id;") + return {kind: [row for row in first_rows + order_rows if row['kind'] == kind] for kind in + ('inbound', 'inboundHistory', 'inbox', 'order', 'orderHistory', 'completionEvent')} + + def prepare(scope, directory, config): + if delivery(config) and producer(scope, 'DELIVERY') is not None: + raise RuntimeError('DELIVERY_EVENT_ALREADY_EXISTS') directory.mkdir(mode=0o700) original = inspect(producer_name(config)) require_test(original, producer_name(config)) - consumer_name = 'firstleg-test-business' if unload(config) else 'firstleg-test-lastleg' + consumer_name = 'firstleg-test-business' if reverse_delivery(config) else 'firstleg-test-lastleg' consumer = inspect(consumer_name) require_test(consumer, consumer_name) old_env = env_of(original) setting, direct = target_setting(config) if old_env.get(setting) != direct: raise RuntimeError('PRODUCER_NOT_DIRECT') - original_body = recreation(original, old_env, full=unload(config)) + original_body = recreation(original, old_env, full=reverse_delivery(config)) token = secrets.token_urlsafe(32) # Reserve task-specific ports and let the existing proxy select exactly this plan. sockets = [socket.socket() for _ in range(2)] @@ -225,8 +268,10 @@ def prepare(scope, directory, config): proxy.write_bytes(base64.b64decode(config['proxySource'])) ip = consumer['NetworkSettings']['Networks']['firstleg-test-network']['IPAddress'] log = open(directory / 'proxy-output.log', 'wb') - upstream = ':9202/internal/firstleg/v1/container-events' if unload(config) else ':9303/internal/lastleg/events' - selector = ['--event-type', 'OverseasContainerUnloaded', '--container-booking-id', scope['containerBookingId']] if unload(config) else ['--event-type', 'FirstLegLoadingPlanCompleted', '--loading-plan-id', scope['loadingPlanId']] + upstream = ':9202/internal/firstleg/v1/container-events' if reverse_delivery(config) else ':9303/internal/lastleg/events' + selector = [] if delivery(config) else ['--event-type', 'OverseasContainerUnloaded', '--container-booking-id', scope['containerBookingId']] if unload(config) else ['--event-type', 'FirstLegLoadingPlanCompleted', '--loading-plan-id', scope['loadingPlanId']] + if delivery(config): + selector = ['--event-type', 'OverseasDeliveryCompleted', '--delivery-id', scope['deliveryId']] process = subprocess.Popen(['python3', str(proxy), '--upstream', 'http://' + ip + upstream, *selector, '--bind', '0.0.0.0', '--port', str(ports[0]), '--control-port', str(ports[1]), '--log', str(directory / 'proxy-evidence.jsonl')], @@ -248,7 +293,7 @@ def activate(scope, directory, state): setting, direct = target_setting(state) if original['Id'] != state['originalId'] or original['Config']['Image'] != state['image'] or env_of(original).get(setting) != direct: raise RuntimeError('CONCURRENT_PRODUCER_CHANGE') - if unload(state) and (original['Image'] != state['originalImageId'] or fingerprint(original) != state['originalFingerprint']): + if reverse_delivery(state) and (original['Image'] != state['originalImageId'] or fingerprint(original) != state['originalFingerprint']): raise RuntimeError('CONCURRENT_PRODUCER_CHANGE') body = copy.deepcopy(state['originalBody']) e = env_of(original) @@ -270,7 +315,7 @@ def recover(scope, directory, state): pinned = control(state) if pinned['eventId'] != row['eventId'] or pinned['payloadHash'] != row['payloadHash']: raise RuntimeError('PINNED_EVENT_MISMATCH') - state['beforeRetry'] = {**row, **({'container': container_facts(scope)} if unload(state) else {'businessOrderIds': business_ids(scope)})} + state['beforeRetry'] = {**row, **business_snapshot(scope, state, row)} private_write(directory / 'state.json', state) control(state, 'PASS') container = inspect(producer_name(state)) @@ -278,7 +323,7 @@ def recover(scope, directory, state): ip = container['NetworkSettings']['Networks']['firstleg-test-network']['IPAddress'] body = {'requestId': state['operationsRequestId'], 'operator': 'admin-full-test', 'reason': scope['marker'] + ': scoped ACK-loss recovery', 'expectedPayloadHash': row['payloadHash']} - endpoint = ':9303/internal/lastleg/event-operations/' if unload(state) else ':9202/internal/firstleg/event-operations/' + endpoint = ':9303/internal/lastleg/event-operations/' if reverse_delivery(state) else ':9202/internal/firstleg/event-operations/' url = 'http://' + ip + endpoint + row['entryId'] + '/retry' request = urllib.request.Request(url, json.dumps(body).encode(), {'X-Event-Operations-Token': e['EVENT_OPERATIONS_TOKEN'], 'Content-Type': 'application/json'}) @@ -295,7 +340,7 @@ def restore(directory, state): try: control(state, 'PASS') except Exception: - if not unload(state): + if not reverse_delivery(state): raise # A dead proxy must not prevent restoring the producer's direct target. if state['prepared']: @@ -309,8 +354,8 @@ def restore(directory, state): state['prepared'] = False private_write(directory / 'state.json', state) return {'restoreStarted': True} - if current['Id'] == state['originalId'] or (unload(state) and current['Image'] == state['originalImageId'] and fingerprint(current) == state['originalFingerprint']): - if unload(state) and not current['State'].get('Running'): + if current['Id'] == state['originalId'] or (reverse_delivery(state) and current['Image'] == state['originalImageId'] and fingerprint(current) == state['originalFingerprint']): + if reverse_delivery(state) and not current['State'].get('Running'): docker('POST', '/containers/' + current['Id'] + '/start') state['prepared'] = False private_write(directory / 'state.json', state) @@ -330,15 +375,15 @@ def proof(scope, state): event_id, entry_id = row['eventId'], row['entryId'] if not re.fullmatch(r'[0-9a-fA-F-]{36}|[0-9A-HJKMNP-TV-Z]{26}', event_id) or not re.fullmatch(r'[1-9][0-9]{0,19}', entry_id): raise RuntimeError('INVALID_EVENT_IDENTITY') - event_type = 'OverseasContainerUnloaded' if unload(state) else 'FirstLegLoadingPlanCompleted' + event_type = 'OverseasDeliveryCompleted' if delivery(state) else 'OverseasContainerUnloaded' if unload(state) else 'FirstLegLoadingPlanCompleted' attempts = read_sql(inspect(producer_name(state)), "SELECT JSON_OBJECT('attemptId',attempt_id,'serviceName',service_name,'entryId',entry_id,'eventId',event_id,'targetName',target_name,'outcome',outcome) FROM event_delivery_attempt WHERE entry_id='" + entry_id + "' AND event_type='" + event_type + "' ORDER BY started_at,attempt_id;") audit = read_sql(inspect(producer_name(state)), "SELECT JSON_OBJECT('requestId',request_id,'entryId',entry_id,'eventId',event_id,'payloadHash',payload_hash,'previousFailures',previous_failures) FROM event_delivery_retry_audit WHERE request_id='" - + state['operationsRequestId'] + "';" if not unload(state) else - "SELECT JSON_OBJECT('requestId',request_id,'entryId',entry_id,'eventId',event_id,'payloadHash',payload_hash,'previousFailures',previous_failures) FROM event_delivery_retry_audit WHERE entry_id='" + entry_id + "' AND event_type='OverseasContainerUnloaded';") - if unload(state): + + state['operationsRequestId'] + "';" if not reverse_delivery(state) else + "SELECT JSON_OBJECT('requestId',request_id,'entryId',entry_id,'eventId',event_id,'payloadHash',payload_hash,'previousFailures',previous_failures) FROM event_delivery_retry_audit WHERE entry_id='" + entry_id + "' AND event_type='" + event_type + "';") + if reverse_delivery(state): consumers = read_sql(inspect('firstleg-test-business'), "SELECT JSON_OBJECT('eventId',event_id,'payloadHash',payload_hash,'status',status,'eventType',event_type,'aggregateId',aggregate_id,'payload',JSON_EXTRACT(payload_json,'$')) FROM fl_container_event_inbox WHERE event_id='" + event_id + "';") else: @@ -348,7 +393,7 @@ def proof(scope, state): consumer = {**(consumers[0] if len(consumers) == 1 else {}), 'count': len(consumers)} result = {**scope, 'readOnly': True, 'producer': row, 'consumer': consumer, 'attempts': attempts, 'proxyRecords': control(state)['records'], 'beforeRetry': state['beforeRetry'], - 'afterRetry': {**row, **({'container': container_facts(scope)} if unload(state) else {'businessOrderIds': business_ids(scope)})}, + 'afterRetry': {**row, **business_snapshot(scope, state, row)}, 'retryAudit': audit[0] if len(audit) == 1 else None, 'retryAuditCount': len(audit), 'retryRequestId': state['operationsRequestId'], 'restored': False} private_write(scope_directory(scope, state.get('direction', 'LOADING')) / 'proof.json', result) return result @@ -363,14 +408,14 @@ def finish(directory, state): raise RuntimeError('RESTORED_SERVICE_NOT_HEALTHY') if sorted(current['Config']['Env']) != sorted(state['originalBody']['Env']): raise RuntimeError('RESTORED_CONFIGURATION_MISMATCH') - if unload(state) and (current['Image'] != state['originalImageId'] or fingerprint(current) != state['originalFingerprint']): + if reverse_delivery(state) and (current['Image'] != state['originalImageId'] or fingerprint(current) != state['originalFingerprint']): raise RuntimeError('RESTORED_CONFIGURATION_MISMATCH') cmdline = Path('/proc/' + str(state['proxyPid']) + '/cmdline') if cmdline.exists(): if str(directory / 'proxy.py').encode() not in cmdline.read_bytes().split(b'\x00'): raise RuntimeError('PROXY_PROCESS_OWNERSHIP_MISMATCH') os.kill(state['proxyPid'], signal.SIGTERM) - if unload(state): + if reverse_delivery(state): for _ in range(50): if not cmdline.exists() or not cmdline.read_bytes(): break @@ -378,7 +423,7 @@ def finish(directory, state): else: raise RuntimeError('PROXY_NOT_STOPPED') result = {'restored': True} - if unload(state): + if reverse_delivery(state): result.update(healthy=True, proxyStopped=True, producerName=producer_name(state), originalImageId=state['originalImageId'], currentImageId=current['Image'], originalFingerprint=state['originalFingerprint'], currentFingerprint=fingerprint(current))