From 4b531c7e226b38aaf4df930411595a934c8ccd5b Mon Sep 17 00:00:00 2001 From: ouyangxq <775153912@qq.com> Date: Fri, 2 Oct 2026 20:14:17 +0800 Subject: [PATCH] test: integrate reliable event recovery and receiving evidence --- scripts/README.md | 8 +- scripts/admin-test-env-acceptance.mjs | 82 +++- scripts/admin-test-env-modules/README.md | 2 + ...ct-delivery-replay-proof-contract.test.mjs | 36 ++ .../direct-delivery-replay-proof.mjs | 44 ++ .../phase3-loading-finalize.mjs | 4 + .../phase4-cross-region-events.mjs | 15 +- .../phase4-overseas-pda-contract.test.mjs | 99 +++- .../phase4-overseas-pda.mjs | 97 +++- scripts/event-delivery-fault-proxy.py | 18 +- scripts/event-delivery-fault-proxy.test.mjs | 45 +- scripts/fba-core-regression-evidence.mjs | 185 ++++++++ scripts/fba-core-regression-evidence.test.mjs | 60 +++ ...verseas-acceptance-entry-contract.test.mjs | 61 ++- scripts/test-event-fault-executor.mjs | 157 +++++++ scripts/test-event-fault-executor.test.mjs | 221 +++++++++ scripts/test-event-fault-runtime.py | 430 ++++++++++++++++++ 17 files changed, 1527 insertions(+), 37 deletions(-) create mode 100644 scripts/admin-test-env-modules/direct-delivery-replay-proof-contract.test.mjs create mode 100644 scripts/admin-test-env-modules/direct-delivery-replay-proof.mjs create mode 100644 scripts/fba-core-regression-evidence.mjs create mode 100644 scripts/fba-core-regression-evidence.test.mjs create mode 100644 scripts/test-event-fault-executor.mjs create mode 100644 scripts/test-event-fault-executor.test.mjs create mode 100644 scripts/test-event-fault-runtime.py diff --git a/scripts/README.md b/scripts/README.md index 8b892e876..afe32d1a5 100644 --- a/scripts/README.md +++ b/scripts/README.md @@ -1,6 +1,6 @@ # 后端脚本管理与全量 HTTP 验收索引 -`event-delivery-fault-proxy.py` 是 TEST 环境定向事件 ACK 丢失工具，不创建或重放事件。运行者显式传入 `--upstream` 完整消费者地址、`--event-type`，并只选 `--event-id` 或 `--loading-plan-id`；`EVENT_FAULT_CONTROL_TOKEN` 从外部环境注入，保护固定本机监听的控制端口。默认 PASS；经鉴权切为 FAULT 后，仅在选定事件已真实转发且收到匹配的 APPLIED/DUPLICATE ACK 时向发送方返回 503，其他事件直通。日志只保留事件标识、摘要和结果，不保留正文或凭据。它会影响实际投递，仅限明确授权的测试环境；生产者配置、重试、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`；柜次选择精确匹配 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；环境执行仍需本轮写入授权。 @@ -351,5 +351,11 @@ Dry-run 只校验两个统一 SQL 的入口标记，打印当前 SHA-256、行 部署离线合同：`overseas-deployment-contract.test.mjs` 现验证统一TEST拓扑，保留历史文件名便于调用；`deploy-built-service-rollback.test.mjs`以mock Docker验证失败与同契约回滚，无真实环境操作。唯一执行清单见 `../deploy/TEST单机部署执行清单.md`。 # 事件人工重试运维入口 +`fba-core-regression-evidence.mjs` 是 Admin/F40 共用的内部只读取证器，不是独立 CLI。只在明确的 TEST Jenkins 节点读取 `firstleg-test-order`、`firstleg-test-lastleg` 对应数据库的一致性只读快照，原始库存、位置、流水及事件身份严格按本轮客户/仓库/柜次过滤。连接凭据只在执行节点内存中处理，不执行 SQL 写入；调用者保存独立报告，失败不能用空结果代替。直接合同为 `fba-core-regression-evidence.test.mjs`，覆盖只读 SQL 边界、完整作用域、长 ID、零事件保留与敏感错误脱敏；离线合同不代替真实数据库取证。 + +海外卸柜反向故障验收从 Admin 总入口显式传 `--modules phase4-overseas-pda --verify-unload-event-recovery true`，按正式依赖准备新客户、新柜及实收，首个本轮柜完成前设置独立故障 hook；普通收货默认不启用。目标事件固定为 `OverseasContainerUnloaded`，生产者为 Lastleg，消费者为 Firstleg，事件平台作用域固定 0、库存仍按本轮实际客户隔离。按原事件验证 ACK 丢失、DEAD、人工 retry 与重复确认；在恢复服务配置后，用相同只读库存取证器采集 `AFTER_EVENT_RECOVERY`，核对原始库存、位置、托盘组成、收货完整键及流水零漂移。任一方向恢复失败保留任务恢复记录并判失败；不得与独立部署或共享环境造数并行，不修改业务 SQL。报告沿用显式新 `--output`，旧失败报告和业务样本保留，不自动清理。此项不覆盖签收完成回写、设备和车辆新版链路。 + +Admin 全量中的 `phase4-cross-region-events` 使用内部 `test-event-fault-executor.mjs` 和 `test-event-fault-runtime.py`，不是额外 CLI 入口。它们仅在显式 `CHONGCHONG_ENV=test` 和已授权环境中运行：在首个本轮计划完成前启动已有 ACK 丢失代理，短暂重建同镜像 Firstleg 容器，保留数据卷、端口和网络身份，只变更事件目标与本次故障退避参数。仅选定的装柜事件被抑制 ACK；其他事件直通。核对 12 次失败、DEAD、受保护原事件 retry、DUPLICATE 及业务订单零新增，结束或失败都恢复原镜像配置并检查健康，随后停止本任务代理。不修改业务 SQL、载荷、事件 ID 或 hash，不重放已发布事件。恢复失败必须保留阻断和任务元数据；独立后端发布或其他环境造数不得与此窗口并行。凭据仅在执行节点内存与权限600的临时恢复文件中使用，恢复完成后移除临时文件中的凭据。离线合同为 `test-event-fault-executor.test.mjs`。 + `event-outbox-operations.mjs`：单测试环境的内部运维脚本。只连接显式指定的生产服务地址，不通过公网 Gateway，不自动重试或跟随重定向。凭据只从进程环境 `EVENT_OPERATIONS_TOKEN` 读取，另须 `CHONGCHONG_ENV=test`；不要把真实凭据放入参数或仓库。`--service firstleg|lastleg|finance --base <服务源地址> --entry ` 只读查询状态和摘要，不返回业务负载。追加 `--retry --request-id <稳定UUID> --operator <操作人> --reason <原因> --hash ` 会写入目标测试服务的 Outbox 与审计表，执行前须已有该环境操作授权。仅 DEAD 原事件可重试，事件身份、负载及摘要保持不变；重复提交使用同一个请求编号，参数变化会被拒绝。三个服务须先部署 `event_delivery_retry_audit` 表，未设置至少32字节的运维凭据时入口关闭。脚本将脱敏结果输出到标准输出，不持久化凭据；离线验证为 `node --test scripts/event-outbox-operations.test.mjs`。 diff --git a/scripts/admin-test-env-acceptance.mjs b/scripts/admin-test-env-acceptance.mjs index dcf0b8e87..d1232dc94 100644 --- a/scripts/admin-test-env-acceptance.mjs +++ b/scripts/admin-test-env-acceptance.mjs @@ -383,6 +383,7 @@ export function parseArguments(argv) { routeMode: values.get('route-mode') || 'fixed', quiet: values.get('quiet') === 'true', returnMainChainOnly: values.get('return-main-chain-only') === 'true', + verifyUnloadEventRecovery: values.get('verify-unload-event-recovery') === 'true', manualSplitFixtureReport: values.get('manual-split-fixture-report') || '', manualSplitPlanId: values.get('manual-split-plan-id') || '', manualPreloadSourceReport: values.get('manual-preload-source-report') || '', @@ -394,6 +395,9 @@ export function parseArguments(argv) { } export function validateExecutionOptions(options) { + if (options.verifyUnloadEventRecovery && !options.modules.includes('phase4-overseas-pda')) { + throw new Error('卸柜事件恢复验收必须包含正式 phase4-overseas-pda 收货链') + } const unavailable = options.modules.filter((module) => !AVAILABLE_MODULES.has(module)) if (unavailable.length > 0) throw new Error(`未注册的 HTTP 验收模块: ${unavailable.join(',')}`) for (const [name, value] of [['gateway', options.gateway], ['username', options.username], ['password', options.password]]) { @@ -1424,7 +1428,7 @@ export class AcceptanceRun { const attempt = (this.phase4ProducerAttempts.get(label) || 0) + 1 this.phase4ProducerAttempts.set(label, attempt) const variantMarker = `${matrixMarker}-P4-${label}${attempt === 1 ? '' : `-R${attempt}`}` - const contextKeys = ['orderDestinationType', 'orderFulfillmentType', 'phase4ProducerMatrixMarker'] + const contextKeys = ['orderDestinationType', 'orderFulfillmentType', 'phase4ProducerMatrixMarker', 'waitSplitRoute'] const createdKeys = [ 'phase3Loading', 'phase3LoadingCompletedProducer', 'phase3ContainerDocuments', 'orchestrationFixtures', 'oceanBooking', 'phase3CustomsSupplier', 'phase3Customs', 'serviceCenterFlow', 'portSchedule', @@ -1557,6 +1561,11 @@ export class AcceptanceRun { if (!hasOperationsFbaProducer()) { await this.runPhase4ProducerVariant('FBA-OPERATIONS', 'FBA', 'STANDARD', true) } + const hasRouteFbaProducer = () => (this.createdResources.phase4InboundProducerFixtures?.containers || []) + .some((item) => /-P4-FBA-ROUTE(?:-R\d+)?:/.test(String(item?.resourceKey || ''))) + if (!hasRouteFbaProducer()) { + await this.runPhase4ProducerVariant('FBA-ROUTE', 'FBA', 'STANDARD', true) + } if (Number(readiness().overseasContainerCount || 0) < 1) { await this.runPhase4ProducerVariant('OVERSEAS-PRIMARY', 'OVERSEAS', 'ONE_CLICK_DROPSHIP', true) } @@ -1576,6 +1585,8 @@ export class AcceptanceRun { /-P4-FBA-SECOND(?:-R\d+)?:/.test(String(item?.resourceKey || ''))) const operationsFbaReady = containers.find((item) => /-P4-FBA-OPERATIONS(?:-R\d+)?:/.test(String(item?.resourceKey || ''))) + const routeFbaReady = containers.find((item) => + /-P4-FBA-ROUTE(?:-R\d+)?:/.test(String(item?.resourceKey || ''))) const awaitingVariants = containers.filter((item) => ['SECONDARY'] .some((label) => new RegExp(`-P4-${label}(?:-R\\d+)?:`).test(String(item?.resourceKey || '')))) const shippingMatrixReady = dropshipReady?.shippingReady === true @@ -1583,6 +1594,7 @@ export class AcceptanceRun { && standardFbaReady?.shippingReady === true && Boolean(standardFbaReady?.carrierScheduleQueryId) && secondStandardFbaReady?.shippingReady === true && Boolean(secondStandardFbaReady?.carrierScheduleQueryId) && operationsFbaReady?.shippingReady === true && Boolean(operationsFbaReady?.carrierScheduleQueryId) + && routeFbaReady?.shippingReady === true && Boolean(routeFbaReady?.carrierScheduleQueryId) && overseasReady?.shippingReady === true && Boolean(overseasReady?.carrierScheduleQueryId) && awaitingVariants.length >= 1 && awaitingVariants.every((item) => item.shippingReady === false @@ -1830,6 +1842,11 @@ export class AcceptanceRun { } } } + if (this.options.verifyUnloadEventRecovery) { + this.addStep('phase4-overseas-pda', '显式卸柜故障场景必须实际完成且只验证本轮一个柜', null, + unloadRecoveryWasVerified(this), '不以未触发故障或普通收货成功代替事件恢复证据', + { scenario: 'reverse-unload-event-recovery', deploymentBlocker: true }) + } const reportSurface = JSON.stringify({ steps: this.steps, modules: this.modules, scenarios: this.scenarios, deploymentBlockers: this.deploymentBlockers, coverageGaps: this.coverageGaps, @@ -1870,11 +1887,72 @@ export class AcceptanceRun { } } +export function unloadRecoveryWasVerified(run) { + const records = Object.entries(run.createdResources?.phase4OverseasPda?.receivingCompletionEvidence || {}) + .filter(([, row]) => row.eventRecovery) + if (records.length !== 1) return false + const [bookingId, row] = records[0] + return row.status === 'VERIFIED' && row.eventRecoveryInventoryUnchanged === true + && row.eventRecovery.restored === true && row.eventRecovery.marker === run.marker + && row.eventRecovery.containerBookingId === bookingId +} + +export async function configureReadOnlyEvidence(run, environment, createReaders, faultFactories) { + if (!run.options.modules.some(name => ['phase4-overseas-pda', 'phase5-root-fulfillment', 'manual-state-matrix'].includes(name))) return + const factory = createReaders || (await import('./fba-core-regression-evidence.mjs')).createTestEvidenceReaders + const readers = factory(environment) + await readers.preflight() + run.context.phase4ReceivingEvidenceReader = readers.readReceiving + run.context.f40ReadOnlyOutboxEvidence = readers.readCompletion + const faults = [] + if (run.options.modules.includes('phase4-cross-region-events')) { + const { createTestFaultExecutor } = faultFactories || await import('./test-event-fault-executor.mjs') + const fault = createTestFaultExecutor(environment) + run.context.beforeLoadingComplete = fault.prepare + run.context.directDeliveryReplayEvidenceReader = fault.readProof + faults.push(fault) + } + if (run.options.verifyUnloadEventRecovery) { + const { createTestUnloadFaultExecutor } = faultFactories || await import('./test-event-fault-executor.mjs') + const fault = createTestUnloadFaultExecutor(environment) + faults.push(fault) + let selected = null + run.context.beforeReceivingComplete = async scope => { + if (selected) return + selected = { marker: run.marker, customerId: scope.customerId, containerBookingId: scope.bookingId } + await fault.prepare(selected) + } + run.context.afterReceivingComplete = async scope => { + if (!selected || scope.bookingId !== selected.containerBookingId || scope.customerId !== selected.customerId) return null + return fault.readProof(selected) + } + } + if (faults.length) return { + cleanup: async () => { + const errors = [] + for (const fault of [...faults].reverse()) { + try { await fault.cleanup() } catch { errors.push('TEST事件故障验收恢复失败，保留原任务恢复记录') } + } + if (errors.length) throw new Error(errors.join('; ')) + }, + } +} + async function main() { const options = validateExecutionOptions(parseArguments(process.argv.slice(2))) const timestamp = new Date().toISOString().replace(/\D/g, '').slice(0, 14) const outputPath = resolve(options.output || `reports/admin-test-env-acceptance-${timestamp}.json`) - const report = await new AcceptanceRun(options).run() + const run = new AcceptanceRun(options) + const fault = await configureReadOnlyEvidence(run, process.env) + let report + try { report = await run.run() } + finally { + try { await fault?.cleanup() } + catch (error) { run.addStep('phase4-cross-region-events', '故障验收后必须恢复原服务配置与健康状态', null, + false, String(error.message), { scenario: 'direct-delivery-replay', deploymentBlocker: true }) } + } + report.summary = buildAcceptanceSummary(run.steps, run.modules, run.coverageGaps) + report.success = report.fatal === null && report.summary.failed === 0 await mkdir(dirname(outputPath), { recursive: true }) await writeFile(outputPath, `${JSON.stringify(report, null, 2)}\n`, 'utf8') console.log(`REPORT=${outputPath}`) diff --git a/scripts/admin-test-env-modules/README.md b/scripts/admin-test-env-modules/README.md index ee4ee0940..4f915ea17 100644 --- a/scripts/admin-test-env-modules/README.md +++ b/scripts/admin-test-env-modules/README.md @@ -356,3 +356,5 @@ node --test scripts/admin-test-env-modules/merchant-qualification-fixture-contra 国内拖车自动完成与应付回归由 `phase3-outbound-dispatch.mjs` 在真实离仓后只读核验：订舱全部柜数、逐柜离仓、唯一 `AUTO_COMPLETE`、按拖车编号追溯柜次应付、金额守恒及 Finance 来源入池（有界等待 Outbox）。该模块的原有确认/撤销边界操作仍会写目标环境，执行须获得对应环境授权。`manual-state-ocean-trucking.mjs` 只构造可人工操作的状态，不伪造已完成拖车；自动终态由上述离仓链路覆盖。 `event-bridge-pending-recovery.mjs` 为纯离线证据校验器，文件名保留为索引入口；无事件桥调用或历史事件常量。合同覆盖 Compose DNS、三服务内部凭据版本一致、有限重试/DEAD、原事件摘要、ACK 丢失后 DUPLICATE 与单次业务效果。合成合同通过不代表环境故障演练通过；`phase4-cross-region-events` 保留真实重放证据阻断项。环境验收需本轮 marker 事件逐次投递记录、Inbox 与业务计数；仅使用受保护 `event-outbox-operations.mjs` 对 DEAD 原事件执行另行授权重试，不重放 SENT，不改表或伪造新事件。 + +`direct-delivery-replay-proof.mjs` 是事件模块的内部证据校验器，不是独立执行入口。要求本轮原装柜事件的 ID/hash、12 次失败至 DEAD、人工恢复审计、持久 Inbox 及 DUPLICATE、恢复前后业务订单身份集合零新增和环境恢复同时成立；单纯重复 GET 或历史故障报告不能满足条件。离线合同为 `direct-delivery-replay-proof-contract.test.mjs`。 diff --git a/scripts/admin-test-env-modules/direct-delivery-replay-proof-contract.test.mjs b/scripts/admin-test-env-modules/direct-delivery-replay-proof-contract.test.mjs new file mode 100644 index 000000000..9e57753cb --- /dev/null +++ b/scripts/admin-test-env-modules/direct-delivery-replay-proof-contract.test.mjs @@ -0,0 +1,36 @@ +import assert from 'node:assert/strict' +import test from 'node:test' +import { validDirectDeliveryReplayProof } from './direct-delivery-replay-proof.mjs' + +const eventId = 'd54a246f-c71c-462f-ab5b-a44f0e23d29a' +const payloadHash = `sha256:${'a'.repeat(64)}` +const scope = { marker: 'current-run', customerId: '7', loadingPlanId: '99', containerBookingId: '66' } +function proof() { + return { ...scope, readOnly: true, restored: true, + producer: { eventId, payloadHash, entryId: '88', publishStatus: 'PUBLISHED', eventType: 'FirstLegLoadingPlanCompleted' }, + consumer: { eventId, payloadHash, status: 'PROCESSED', count: 1 }, + attempts: Array.from({ length: 13 }, (_, i) => ({ attemptId: `attempt-${i}`, eventId, entryId: '88', + targetName: 'lastleg-service/internal/lastleg/events', outcome: i === 12 ? 'DUPLICATE' : i === 11 ? 'DEAD' : 'RETRY' })), + proxyRecords: Array.from({ length: 13 }, (_, i) => ({ eventId, payloadHash, matched: true, + outcome: i === 0 ? 'APPLIED' : 'DUPLICATE', errorCode: i === 12 ? null : 'ACK_LOSS_INJECTED' })), + retryAudit: { eventId, payloadHash, entryId: '88', previousFailures: 12, requestId: 'retry-1' }, + beforeRetry: { eventId, payloadHash, publishStatus: 'DEAD', failedAttempts: 12, businessOrderIds: ['11', '12'] }, + afterRetry: { eventId, payloadHash, publishStatus: 'PUBLISHED', businessOrderIds: ['12', '11'] } } +} +test('only current original event finite failure, authenticated recovery and persistent duplicate consumption together pass', () => { + assert.equal(validDirectDeliveryReplayProof(proof(), scope), true) +}) +test('old evidence, GET stability, incomplete attempts, hash drift and business duplication fail closed', () => { + const changes = [ + p => { p.marker = 'old' }, p => { p.attempts = [] }, p => { p.proxyRecords = [] }, + p => { p.consumer.count = 2 }, p => { p.attempts[12].outcome = 'APPLIED' }, + p => { p.retryAudit.payloadHash = `sha256:${'b'.repeat(64)}` }, + p => { p.afterRetry.businessOrderIds.push('13') }, p => { p.restored = false }, + p => { p.attempts[11].outcome = 'RETRY' }, p => { p.beforeRetry.failedAttempts = 11 }, + p => { p.attempts[12].targetName = 'lastleg-service/internal/finance/events' }, + ] + for (const change of changes) { + const p = proof(); change(p) + assert.equal(validDirectDeliveryReplayProof(p, scope), false) + } +}) diff --git a/scripts/admin-test-env-modules/direct-delivery-replay-proof.mjs b/scripts/admin-test-env-modules/direct-delivery-replay-proof.mjs new file mode 100644 index 000000000..630f94098 --- /dev/null +++ b/scripts/admin-test-env-modules/direct-delivery-replay-proof.mjs @@ -0,0 +1,44 @@ +// This checks live, scoped evidence supplied by the TEST fault executor. +// It never accepts historical reports or initiates event replay itself. +const identifier = value => typeof value === 'string' && /^[1-9][0-9]{0,19}$/.test(value) +const eventIdentifier = value => typeof value === 'string' + && /^(?:[0-9a-f]{8}(?:-[0-9a-f]{4}){3}-[0-9a-f]{12}|[0-9A-HJKMNP-TV-Z]{26})$/i.test(value) +const hash = value => typeof value === 'string' && /^sha256:[0-9a-f]{64}$/i.test(value) +const sortedIds = rows => [...rows].sort() + +export function validDirectDeliveryReplayProof(proof, scope) { + if (!proof || proof.readOnly !== true || proof.marker !== scope.marker + || !identifier(scope.loadingPlanId) || !identifier(scope.customerId) + || proof.loadingPlanId !== scope.loadingPlanId || proof.customerId !== scope.customerId + || proof.containerBookingId !== scope.containerBookingId) return false + const { producer, consumer, attempts, proxyRecords, beforeRetry, afterRetry, retryAudit } = proof + if (!eventIdentifier(producer?.eventId) || !hash(producer?.payloadHash) + || !identifier(producer?.entryId) || producer.publishStatus !== 'PUBLISHED' + || producer.eventType !== 'FirstLegLoadingPlanCompleted' + || consumer?.eventId !== producer.eventId || consumer?.payloadHash !== producer.payloadHash + || consumer?.status !== 'PROCESSED' || consumer?.count !== 1) return false + if (!Array.isArray(attempts) || attempts.length !== 13 + || new Set(attempts.map(row => row.attemptId)).size !== attempts.length + || attempts.some(row => !row.attemptId || row.eventId !== producer.eventId + || row.entryId !== producer.entryId || row.targetName !== 'lastleg-service/internal/lastleg/events') + || attempts.at(-1)?.outcome !== 'DUPLICATE' + || attempts[11]?.outcome !== 'DEAD' + || attempts.slice(0, 11).some(row => row.outcome !== 'RETRY')) return false + if (!Array.isArray(proxyRecords) || proxyRecords.length !== 13 + || proxyRecords.some(row => row.matched !== true || row.eventId !== producer.eventId + || row.payloadHash !== producer.payloadHash) + || !proxyRecords.some(row => row.outcome === 'APPLIED' && row.errorCode === 'ACK_LOSS_INJECTED') + || !proxyRecords.some(row => row.outcome === 'DUPLICATE' && row.errorCode === 'ACK_LOSS_INJECTED')) return false + if (retryAudit?.eventId !== producer.eventId || retryAudit?.payloadHash !== producer.payloadHash + || retryAudit?.entryId !== producer.entryId || !retryAudit.requestId + || retryAudit.previousFailures !== 12 || beforeRetry?.publishStatus !== 'DEAD' + || beforeRetry?.failedAttempts !== 12 || afterRetry?.publishStatus !== 'PUBLISHED' + || beforeRetry.eventId !== producer.eventId || afterRetry.eventId !== producer.eventId + || beforeRetry.payloadHash !== producer.payloadHash || afterRetry.payloadHash !== producer.payloadHash) return false + const before = beforeRetry.businessOrderIds, after = afterRetry.businessOrderIds + return Array.isArray(before) && Array.isArray(after) && before.length > 0 + && before.every(identifier) && after.every(identifier) + && new Set(before).size === before.length && new Set(after).size === after.length + && JSON.stringify(sortedIds(before)) === JSON.stringify(sortedIds(after)) + && proof.restored === true +} diff --git a/scripts/admin-test-env-modules/phase3-loading-finalize.mjs b/scripts/admin-test-env-modules/phase3-loading-finalize.mjs index c16d32705..5e46e93a4 100644 --- a/scripts/admin-test-env-modules/phase3-loading-finalize.mjs +++ b/scripts/admin-test-env-modules/phase3-loading-finalize.mjs @@ -229,6 +229,10 @@ export async function run(context) { packageCount: Number(item?.packageCount), picked: Number(item?.pickedPackageCount) })), }) const operationsBeforeSignature = JSON.stringify(operationsBefore || []) + if (typeof context.context?.beforeLoadingComplete === 'function') { + await context.context.beforeLoadingComplete({ marker: context.marker, customerId: tenantId, + loadingPlanId, containerBookingId }) + } const completeAttempts = [1, 2].map((index) => ({ body: { expectedTaskVersion: Number(taskBefore.version), diff --git a/scripts/admin-test-env-modules/phase4-cross-region-events.mjs b/scripts/admin-test-env-modules/phase4-cross-region-events.mjs index c02993209..932f63f0b 100644 --- a/scripts/admin-test-env-modules/phase4-cross-region-events.mjs +++ b/scripts/admin-test-env-modules/phase4-cross-region-events.mjs @@ -1,3 +1,5 @@ +import { validDirectDeliveryReplayProof } from './direct-delivery-replay-proof.mjs' + const MODULE = 'phase4-cross-region-events' const OPTIONS = { scenario: 'phase4-cross-region-events', deploymentBlocker: true } @@ -139,7 +141,18 @@ export async function run(context) { success(writtenBackAgain) && frozen.every((field) => (replaySnapshot[field] ?? null) === (snapshot[field] ?? null)) && traced(writtenBackAgain), evidence(writtenBackAgain, `containerId=${id(domesticMatches[0].id)}, sourceVersion=${snapshot.emptyReportSourceVersion ?? 'missing'}`), OPTIONS) - blockingGap(context, '真实投递重试与同事件幂等证据', + const reader = context.context?.directDeliveryReplayEvidenceReader + if (typeof reader !== 'function') blockingGap(context, '真实投递重试与同事件幂等证据', '两次GET仅证明读取稳定；需按event-outbox-operations.mjs的受保护内部运维流程，在本轮事件上验证故障后有限重试/DEAD及同eventId+payloadHash重复消费，保存逐次尝试与业务计数证据。不得重放SENT、改表或伪造新事件。', 'missing-direct-delivery-replay-evidence') + const scope = { marker: context.marker, loadingPlanId: id(phase3.loadingPlanId), + customerId: id(phase3.tenantId), containerBookingId: id(phase3.containerBookingId) } + const proof = await reader(scope) + context.createdResources.directDeliveryReplayProof = proof + const valid = validDirectDeliveryReplayProof(proof, scope) + context.assert(MODULE, '本轮原事件12次失败DEAD人工恢复DUPLICATE且业务零新增', null, valid, + `loadingPlanId=${scope.loadingPlanId}, attempts=${proof?.attempts?.length || 0}, restored=${proof?.restored === true}`, OPTIONS) + if (!valid) blockingGap(context, '真实投递重试证据不完整或身份漂移', + '本轮事件身份/摘要、12次有限重试、DEAD、人工恢复审计、重复消费及业务计数和恢复直连必须同时核平', + 'invalid-direct-delivery-replay-evidence') } diff --git a/scripts/admin-test-env-modules/phase4-overseas-pda-contract.test.mjs b/scripts/admin-test-env-modules/phase4-overseas-pda-contract.test.mjs index a70e5bbdf..d90c64c69 100644 --- a/scripts/admin-test-env-modules/phase4-overseas-pda-contract.test.mjs +++ b/scripts/admin-test-env-modules/phase4-overseas-pda-contract.test.mjs @@ -6,7 +6,54 @@ import { fileURLToPath } from 'node:url' import { expandOverseasCartonCandidates, resolveCartonFixtures, run, uniqueContainerProducer, crossContainerCartonProbe, normalizeReceivingEvidence, receivingEvidenceValid, - receivingCompletionProof, captureReceivingEvidence, probeCartonResolveNegatives } from './phase4-overseas-pda.mjs' + receivingCompletionProof, captureReceivingEvidence, receivingEventRecoveryInventoryUnchanged, + probeCartonResolveNegatives, awaitReceivingCandidate } from './phase4-overseas-pda.mjs' + +function candidateContext(responses) { + const calls=[],steps=[] + return {calls,steps,createdResources:{},addStep:(_module,name,_response,passed)=>steps.push({name,passed}), + request:async(method,path,_body,token)=>{calls.push({method,path,token});return responses[Math.min(calls.length-1,responses.length-1)]}} +} +const candidateResponse = data => ({http:200,body:{code:'200',data},requestId:'offline-request',traceId:'offline-trace'}) +const exactCandidate = {containerNo:'TSTU6403402',sourceContainerBookingId:'2105838069516079106',operable:true,totalPackages:1} +const exactProducer = {containerNo:exactCandidate.containerNo,containerBookingId:exactCandidate.sourceContainerBookingId} + +test('original r2 async shipping gate race waits same booking and container until visible without writing',async()=>{ + const context=candidateContext([candidateResponse([]),candidateResponse([]),candidateResponse([exactCandidate])]),delays=[] + const result=await awaitReceivingCandidate(context,'offline-picker',exactProducer,{attempts:3,delay:async ms=>delays.push(ms)}) + assert.deepEqual(result.candidate,exactCandidate) + assert.deepEqual(delays,[1000,1000]) + assert.equal(context.calls.length,3) + assert.ok(context.calls.every(call=>call.method==='GET'&&call.token==='offline-picker'&&call.path.endsWith('/candidates'))) + const proof=context.createdResources.phase4OverseasPda.candidateWaitEvidence[0] + assert.equal(proof.outcome,'READY');assert.deepEqual(proof.attempts.map(v=>v.exactMatches),[0,0,1]) + assert.ok(proof.attempts.every(v=>v.requestId&&v.traceId)) +}) +test('same container from another booking, other container and nonoperable original cannot satisfy wait',async()=>{ + for(const candidate of [{...exactCandidate,sourceContainerBookingId:'999'}, + {...exactCandidate,containerNo:'ANOTHER'},{...exactCandidate,operable:false}]) { + const context=candidateContext([candidateResponse([candidate])]);let waits=0 + const result=await awaitReceivingCandidate(context,'offline',exactProducer,{attempts:2,delay:async()=>waits++}) + assert.equal(result.candidate,null);assert.equal(result.reason,'TIMED_OUT') + assert.equal(context.calls.length,2);assert.equal(waits,1) + } +}) +test('ambiguous exact candidates and authentication or malformed replies stop without masking server errors',async()=>{ + for(const response of [candidateResponse([exactCandidate,exactCandidate]),{http:401,body:{code:'401'}},candidateResponse({records:[exactCandidate]})]) { + const context=candidateContext([response]);let waits=0 + const result=await awaitReceivingCandidate(context,'offline',exactProducer,{attempts:3,delay:async()=>waits++}) + assert.equal(result.candidate,null);assert.equal(context.calls.length,1);assert.equal(waits,0) + assert.ok(['AMBIGUOUS','QUERY_FAILED'].includes(result.reason)) + } +}) +test('candidate waits are per invocation with no cross-mode cache and enforce exact scope/bounds',async()=>{ + const context=candidateContext([candidateResponse([exactCandidate]),candidateResponse([])]) + assert.ok((await awaitReceivingCandidate(context,'first',exactProducer,{attempts:1})).candidate) + assert.equal((await awaitReceivingCandidate(context,'second',exactProducer,{attempts:1})).candidate,null) + for(const producer of [{containerNo:'TSTU6403402'},{}])await assert.rejects(awaitReceivingCandidate(context,'offline',producer)) + await assert.rejects(awaitReceivingCandidate(context,'offline',exactProducer,{attempts:31})) + assert.equal(context.calls.length,2) +}) test('single-container continuation retains malformed/unknown probes without requiring unshipped cancelled CHILD; default chain still blocks missing real other container', async () => { const requests = [], gaps = [], checks = [] @@ -73,6 +120,28 @@ test('receiving completion freezes all raw balances when group MIN represents tw assert.equal(stages.AFTER_COMPLETE.positions.length, 2) }) +test('unrelated canonical UNBOUND stock remains in all snapshots but new receiving still requires a real pallet', () => { + const { scope, stages } = receivingProofFixture() + const unboundPosition = { ...stages.AFTER_COMPLETE.positions[0], id: '301', balanceId: '30', + palletId: '0', palletNo: 'UNBOUND', availablePackages: 1 } + for (const snapshot of Object.values(stages)) { + const balance = { ...snapshot.balances[0], id: '30', inboundOrderId: '30', + inboundCode: 'UNBOUND-SOURCE', skuCode: 'UNRELATED', availablePackages: 1, inTransitPackages: 0 } + snapshot.balances.push(balance) + snapshot.httpGroups.push({ ...balance }) + snapshot.positions.push({ ...unboundPosition }) + snapshot.palletComposition.push({ ...unboundPosition }) + assert.equal(receivingEvidenceValid(snapshot, scope), true) + } + assert.equal(receivingCompletionProof(scope, stages).passed, true) + stages.AFTER_COMPLETE.stockReceipts[0].warehousePalletId = '0' + assert.equal(receivingEvidenceValid(stages.AFTER_COMPLETE, scope), false) + const malformed = structuredClone(stages.BEFORE_COMPLETE) + malformed.positions[0].palletNo = 'UNKNOWN' + malformed.palletComposition[0].palletNo = 'UNKNOWN' + assert.equal(receivingEvidenceValid(malformed, scope), false) +}) + test('receiving proof rejects missing raw source, wrong customer/warehouse, prefix source, missing bucket, ledger key, duplicate receipt and operation mismatch', () => { const mutations = [ row => row.balances.pop(), row => row.positions[0].customerId = 'OTHER', @@ -111,6 +180,21 @@ test('normalization is deterministic, preserves zero buckets, and excludes token assert.equal(normalizeReceivingEvidence({ ...after, stockReceipts: undefined }), null) }) +test('reverse event recovery requires complete owned inventory evidence and every immutable fact unchanged', () => { + const { scope, stages } = receivingProofFixture() + const before = stages.AFTER_FRESH_KEY_REJECTION + assert.equal(receivingEventRecoveryInventoryUnchanged(scope, before, structuredClone(before)), true) + assert.equal(receivingEventRecoveryInventoryUnchanged(scope, before, null), false) + assert.equal(receivingEventRecoveryInventoryUnchanged({ ...scope, customerId: 'OTHER' }, before, before), false) + for (const mutate of [row => row.balances[0].availablePackages++, row => row.stockReceipts.pop(), + row => row.positions[0].warehouseLocationCode = 'ANOTHER', row => row.stockInLedger[0].sourceEventId = 'OTHER', + row => row.receivingOperations.push({ ...row.receivingOperations[0], id: '999' })]) { + const changed = structuredClone(before) + mutate(changed) + assert.equal(receivingEventRecoveryInventoryUnchanged(scope, before, changed), false) + } +}) + test('missing read-only receipt/key reader is a blocker and retains an unproven boundary without inventing HTTP', async () => { const checks = [], target = { stages: {} } const context = { context: {}, addCoverageGap() {}, assert(m, n, r, passed) { checks.push(passed) }, @@ -166,7 +250,7 @@ test('physical OVERSEAS and dropship probes still require another real loaded pr }) test('receiving selects one current booking and supplies mandatory booking identity', () => { - assert.match(source, /id\(item.sourceContainerBookingId\) === preferredBookingId/) + assert.match(source, /item.containerNo === containerNo && id\(item.sourceContainerBookingId\) === bookingId/) assert.match(source, /matchingCandidates.length === 1/) assert.match(source, /if \(!longId\(sourceContainerBookingId\)\)/) assert.match(source, /sourceContainerBookingId=\$\{encodeURIComponent\(sourceContainerBookingId\)\}/) @@ -314,7 +398,8 @@ test('PDA module covers the container-number receiving contract and excludes out assert.match(source, /standardFbaProducers = \(Array\.isArray\(matrix\?\.containers\) \? matrix\.containers : \[\]\)/) assert.match(source, /item\?\.fulfillmentType !== 'ONE_CLICK_DROPSHIP'/) assert.match(source, /FBA-OPERATIONS/) - assert.match(source, /\.slice\(0, 3\)/) + assert.match(source, /FBA-ROUTE/) + assert.match(source, /\.slice\(0, 4\)/) assert.match(source, /const producers = \[dropship, \.\.\.standardFbaProducers,/) assert.doesNotMatch(source, /创建当前柜实收托盘|创建当前柜实收库位|async function createPallet/) assert.match(source, /retained as acceptance evidence/) @@ -417,7 +502,7 @@ test('real-device deployed FBA resolve A0404 is retained until server-side inbou test('PDA receiving lifecycle is ordered from candidate claim through photos, mode-specific receipts, pallet and complete', () => { const orderedTokens = [ - "'查询当前客户待收柜号'", "'选择柜号并建立WAIT_SIGN会话'", + 'const candidates = await awaitReceivingCandidate', "'选择柜号并建立WAIT_SIGN会话'", 'uploadPhotos(context, token)', "'绑定车牌封签开柜三角色照片并签收'", 'resolveGeneratedPallet(context, inbound)', "`解析${inbound.destinationType}实物托盘`", 'receiveFba(context, token', 'receiveOverseas(context, token', "'回查柜内逐入仓码实收进度'", @@ -464,7 +549,7 @@ test('missing OVERSEAS_WAREHOUSE_PDA_OPERATOR overseas authority is a deployment assert.ok(steps.some((step) => step?.passed === false && step?.options?.deploymentBlocker)) }) -test('empty candidate response cannot become a zero-business-request pass', async () => { +test('missing original producer identity blocks before candidate lookup and still cleans temporary identities', async () => { const calls = []; const gaps = []; const steps = [] const response = (data, requestPath, method = 'GET') => ({ method, path: requestPath, http: 200, body: { code: '200', data }, requestId: 'req-contract', traceId: 'trace-contract' }) @@ -509,9 +594,9 @@ test('empty candidate response cannot become a zero-business-request pass', asyn addCoverageGap(...args) { gaps.push(args) }, } await run(context) - assert.ok(calls.some((call) => call.requestPath.endsWith('/candidates'))) + assert.equal(calls.some((call) => call.requestPath.endsWith('/candidates')),false) assert.ok(steps.some((step) => step.name.includes('正常空态') && step.passed)) - assert.ok(steps.some((step) => step.name.includes('PDA柜号候选覆盖前置') && !step.passed && step.options.deploymentBlocker)) + assert.ok(steps.some((step) => step.name.includes('本轮原柜候选身份覆盖前置') && !step.passed && step.options.deploymentBlocker)) assert.equal(gaps.length, 1) assert.equal(context.createdResources.phase4OverseasPda.finalStatus, null) assert.equal(context.createdResources.phase4OverseasPda.pickerMembershipIds.length, 2) diff --git a/scripts/admin-test-env-modules/phase4-overseas-pda.mjs b/scripts/admin-test-env-modules/phase4-overseas-pda.mjs index 3d05cd717..7dce2f867 100644 --- a/scripts/admin-test-env-modules/phase4-overseas-pda.mjs +++ b/scripts/admin-test-env-modules/phase4-overseas-pda.mjs @@ -99,7 +99,8 @@ export function receivingEvidenceValid(snapshot, scope) { || !snapshot.httpGroups.every(row => owned(row) && bucketsValid(row, RECEIVING_BALANCE_BUCKETS)) || !snapshot.httpLedger.every(owned) || !snapshot.positions.every(row => owned(row) && bucketsValid(row, RECEIVING_POSITION_BUCKETS) - && snapshot.balances.some(balance => balance.id === row.balanceId) && longId(row.palletId) && nonBlank(row.palletNo) + && snapshot.balances.some(balance => balance.id === row.balanceId) + && ((longId(row.palletId) && nonBlank(row.palletNo)) || (row.palletId === '0' && row.palletNo === 'UNBOUND')) && longId(row.warehouseLocationId) && nonBlank(row.warehouseLocationCode)) || !sameEvidence(snapshot.positions, snapshot.palletComposition)) return false if (!snapshot.balances.every(balance => RECEIVING_POSITION_BUCKETS.every(field => @@ -116,6 +117,7 @@ export function receivingEvidenceValid(snapshot, scope) { || !snapshot.httpGroups.every(row => groups.has(balanceGroupKey(row)) && RECEIVING_BALANCE_BUCKETS.every(field => row[field] === groups.get(balanceGroupKey(row))[field]))) return false if (!snapshot.stockReceipts.every(row => row.customerId === scope.customerId && scope.inboundCodes.includes(row.inboundCode) && longId(row.planId) && longId(row.taskId) && longId(row.allocationId) && longId(row.lineId) && longId(row.orderId) + && longId(row.warehousePalletId) && nonBlank(row.palletNo) && nonBlank(row.skuCode) && Number.isSafeInteger(row.receivedPackages) && row.receivedPackages > 0)) return false if (snapshot.inventoryOperations.length !== snapshot.stockReceipts.length || snapshot.stockInLedger.length !== snapshot.stockReceipts.length) return false for (const receipt of snapshot.stockReceipts) { @@ -220,6 +222,11 @@ export async function captureReceivingEvidence(context, scope, stage, target) { } } +export function receivingEventRecoveryInventoryUnchanged(scope, before, after) { + return receivingEvidenceValid(before, scope) && receivingEvidenceValid(after, scope) + && sameEvidence(before, after) +} + export function expandOverseasCartonCandidates(inbounds, producerCartonMarks = []) { const marks = (Array.isArray(producerCartonMarks) ? producerCartonMarks : [producerCartonMarks]) .map(id).filter(nonBlank) @@ -339,7 +346,17 @@ export async function createPicker(context, index, overseasWarehouseId, access) } context.addStep(MODULE, `海外PDA操作员${index}安全投影已收敛`, projectionResponse, projectionReady, `attempts=${projectionAttempt + 1}; ${trace(projectionResponse)}`, OPTIONS) - return { membershipId, username, token: projectionReady ? token : null } + return { membershipId, username, token: projectionReady ? token : null, + refreshToken: login.body?.data?.refreshToken, password, + userUpdate: { displayName: payload.displayName, phone, email: payload.email, tenantId: '0', + roleIds: payload.roleIds, overseasWarehouseId, domesticWarehouseId: null } } +} + +export function pickerSessionHandoff(picker) { + if (!picker) return null + const { membershipId, username, token, refreshToken, password, userUpdate } = picker + return { membershipId, username, token, refreshToken, password, + userUpdate: userUpdate ? { ...userUpdate, roleIds: [...userUpdate.roleIds] } : undefined } } export async function disablePicker(context, picker, index) { @@ -827,7 +844,37 @@ async function receiveDropship(context, token, producer, inbounds) { return accepted === expectedPending } -export async function runReceiving(context, token, secondaryToken, merchantToken, preferredProducer = null, runAuthorizationProbes = true) { +export async function awaitReceivingCandidate(context, token, producer, { attempts = 30, intervalMs = 1000, + delay = ms => new Promise(resolve => setTimeout(resolve, ms)) } = {}) { + const containerNo = id(producer?.containerNo), bookingId = id(producer?.containerBookingId) + if (!nonBlank(containerNo) || !longId(bookingId) || !Number.isInteger(attempts) || attempts < 1 || attempts > 30 + || !Number.isInteger(intervalMs) || intervalMs < 0 || intervalMs > 1000) throw Error('Invalid exact receiving candidate scope or wait bound') + const proof = { containerNo, sourceContainerBookingId:bookingId, attempts:[], outcome:'PENDING' } + const resource = context.createdResources.phase4OverseasPda ||= {} + ;(resource.candidateWaitEvidence ||= []).push(proof) + let result + for (let attempt = 1; attempt <= attempts; attempt++) { + result = await pda(context, token, 'GET', `查询本轮原柜候选收敛第${attempt}次`, '/api/pda/v1/last-leg/receiving-containers/candidates') + const rows = Array.isArray(result.data) ? result.data : [] + const matchingCandidates = rows.filter(item => item.containerNo === containerNo && id(item.sourceContainerBookingId) === bookingId) + proof.attempts.push({ attempt, http:result.response?.http, code:code(result.response), requestId:result.response?.requestId, + traceId:result.response?.traceId, candidateCount:rows.length, exactMatches:matchingCandidates.length, + sameNumberCandidates:rows.filter(item => item.containerNo === containerNo).map(item => ({ + containerNo:item.containerNo, sourceContainerBookingId:id(item.sourceContainerBookingId), + operable:item.operable, blockReason:item.blockReason })) }) + if (!success(result.response) || !Array.isArray(result.data)) { proof.outcome='QUERY_FAILED'; break } + if (matchingCandidates.length > 1) { proof.outcome='AMBIGUOUS'; break } + if (matchingCandidates.length === 1 && matchingCandidates[0].operable === true) { + proof.outcome='READY' + return {...result,candidate:matchingCandidates[0]} + } + if (attempt < attempts) await delay(intervalMs) + } + if (proof.outcome === 'PENDING') proof.outcome='TIMED_OUT' + return {...result,candidate:null,reason:proof.outcome} +} + +export async function runReceiving(context, token, secondaryToken, merchantToken, preferredProducer = null, runAuthorizationProbes = true, candidateWait = {}) { if (runAuthorizationProbes && context.token) { const denied = await context.request('GET', '/api/pda/v1/last-leg/receiving-containers/candidates', undefined, context.token, {}) context.addStep(MODULE, '平台管理员访问令牌不能冒充仓库PDA操作员', denied, forbidden(denied) && traced(denied), trace(denied), OPTIONS) @@ -844,20 +891,19 @@ export async function runReceiving(context, token, secondaryToken, merchantToken blockerGap(context, '可重复的完整柜号验收', `账号已有未完成柜 ${before.body?.data?.containerNo || 'unknown'}，本轮不覆盖或篡改遗留现场会话`, 'dirty-fixture', before) return } - const candidates = await pda(context, token, 'GET', '查询当前客户待收柜号', '/api/pda/v1/last-leg/receiving-containers/candidates') - if (!Array.isArray(candidates.data) || candidates.data.length === 0) { - blockerGap(context, 'PDA柜号候选', `当前客户没有由三期装柜完成事件生成的待收入仓计划；${trace(candidates.response)}`, 'producer-chain', candidates.response) - return - } const preferredContainer = id(preferredProducer?.containerNo) || context.createdResources?.phase3Loading?.containerNo - const preferredBookingId = id(preferredProducer?.containerBookingId) - const matchingCandidates = candidates.data.filter(item => item.containerNo === preferredContainer - && (!preferredBookingId || id(item.sourceContainerBookingId) === preferredBookingId)) - const candidate = matchingCandidates.length === 1 ? matchingCandidates[0] : null + const preferredBookingId = id(preferredProducer?.containerBookingId || context.createdResources?.phase3Loading?.containerBookingId) + if (!nonBlank(preferredContainer) || !longId(preferredBookingId)) { + blockerGap(context, '本轮原柜候选身份', '缺少本轮containerNo或containerBookingId，禁止仅按柜号匹配历史柜', 'producer-chain') + return + } + const candidates = await awaitReceivingCandidate(context,token, + {containerNo:preferredContainer,containerBookingId:preferredBookingId},candidateWait) + const candidate = candidates.candidate if (!candidate) { blockerGap(context, `本轮柜号${preferredContainer || 'missing'}候选`, - '三期装柜完成生产者已落地，但海外候选尚未出现该柜；禁止回退处理历史或其他柜', 'producer-chain', candidates.response) + `本轮原柜候选未在有界等待内唯一可操作：${candidates.reason}；禁止回退处理历史或其他柜`, 'producer-chain', candidates.response) return } const containerNo = id(candidate?.containerNo) @@ -1015,6 +1061,9 @@ export async function runReceiving(context, token, secondaryToken, merchantToken const receivingProof = { scope, stages: {}, proof: null, status: 'UNPROVEN' } context.createdResources.phase4OverseasPda.receivingCompletionEvidence ||= {} context.createdResources.phase4OverseasPda.receivingCompletionEvidence[sourceContainerBookingId] = receivingProof + if (typeof context.context.beforeReceivingComplete === 'function') { + await context.context.beforeReceivingComplete(scope) + } await captureReceivingEvidence(context, scope, 'BEFORE_COMPLETE', receivingProof) const completed = await pda(context, token, 'POST', '完成当前柜并写入库存与卸柜事件', '/api/pda/v1/last-leg/receiving-containers/current/complete', completeBody) await captureReceivingEvidence(context, scope, 'AFTER_COMPLETE', receivingProof) @@ -1033,6 +1082,20 @@ export async function runReceiving(context, token, secondaryToken, merchantToken receivingProof.status = receivingProof.proof.passed && Object.values(receivingProof.commandResults).every(Boolean) ? 'VERIFIED' : 'UNPROVEN' for (const [check, passed] of Object.entries(receivingProof.proof.checks)) context.assert(MODULE, `收货完成库存守恒 ${check}`, null, passed, `bookingId=${sourceContainerBookingId}, proofStatus=${receivingProof.status}`, OPTIONS) + if (typeof context.context.afterReceivingComplete === 'function') { + if (receivingProof.status !== 'VERIFIED') throw new Error('收货完成库存证据未闭合，禁止继续故障恢复验收') + const eventRecovery = await context.context.afterReceivingComplete(scope) + if (eventRecovery) { + receivingProof.eventRecovery = eventRecovery + const afterRecovery = await captureReceivingEvidence(context, scope, 'AFTER_EVENT_RECOVERY', receivingProof) + const unchanged = eventRecovery.restored === true && receivingEventRecoveryInventoryUnchanged(scope, + receivingProof.stages.AFTER_FRESH_KEY_REJECTION.evidence, afterRecovery) + receivingProof.eventRecoveryInventoryUnchanged = unchanged + context.assert(MODULE, '原卸柜事件DEAD恢复后库存位置托盘组成及不可变流水零漂移', null, unchanged, + `bookingId=${sourceContainerBookingId}`, OPTIONS) + if (!unchanged) throw new Error('卸柜事件恢复后库存证据变化或服务恢复未验证') + } + } const after = await context.request('GET', '/api/pda/v1/last-leg/receiving-containers/current', undefined, token, {}) context.addStep(MODULE, '完成后current返回404正常空态', after, notFound(after), trace(after), OPTIONS) context.createdResources.phase4OverseasPda.finalStatus = completed.data?.status || null @@ -1110,8 +1173,8 @@ export async function run(context) { } context.context.phase4OverseasPdaOperations = { marker: context.marker, - primary: primary ? { membershipId: primary.membershipId, username: primary.username, token: primary.token } : null, - secondary: secondary ? { membershipId: secondary.membershipId, username: secondary.username, token: secondary.token } : null, + primary: pickerSessionHandoff(primary), + secondary: pickerSessionHandoff(secondary), overseasWarehouseId: sharedWarehouse.id, merchantToken, completedFixtures: [], @@ -1127,9 +1190,9 @@ export async function run(context) { const standardFbaProducers = (Array.isArray(matrix?.containers) ? matrix.containers : []) .filter((item) => item?.destinationType === 'FBA') .filter((item) => item?.fulfillmentType !== 'ONE_CLICK_DROPSHIP') - .filter((item) => /-P4-(?:FBA-STANDARD|FBA-SECOND|FBA-OPERATIONS)(?:-R\d+)?:/.test(String(item?.resourceKey || ''))) + .filter((item) => /-P4-(?:FBA-STANDARD|FBA-SECOND|FBA-OPERATIONS|FBA-ROUTE)(?:-R\d+)?:/.test(String(item?.resourceKey || ''))) .sort((left, right) => String(left?.resourceKey).localeCompare(String(right?.resourceKey))) - .slice(0, 3) + .slice(0, 4) // The sealed dropship pallet proves the indivisible handover. A separate standard FBA // fixture supplies multiple cartons for reversible operations. The second standard FBA // fixture remains available for the multi-destination dispatch and delivery split proof. diff --git a/scripts/event-delivery-fault-proxy.py b/scripts/event-delivery-fault-proxy.py index 0dfbd0a6d..7bd8fa203 100644 --- a/scripts/event-delivery-fault-proxy.py +++ b/scripts/event-delivery-fault-proxy.py @@ -32,10 +32,11 @@ def encoded(value): class State: - def __init__(self, event_type, event_id, loading_plan_id, log_path=None): + def __init__(self, event_type, event_id, loading_plan_id, log_path=None, container_booking_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.log_path = log_path self.lock = threading.Lock() self.mode = "PASS" @@ -65,9 +66,13 @@ class State: return None, False, None, 0, self.mode identity = {"eventId": event_id, "payloadHash": payload_hash.lower()} payload = event.get("payload") - qualifies = event.get("eventType") == self.event_type and ( - event_id == self.selector_id if self.selector_id else - isinstance(payload, dict) and str(payload.get("loadingPlanId")) == self.plan_id) + if self.selector_id: + selected = event_id == self.selector_id + elif self.booking_id: + selected = isinstance(payload, dict) and str(payload.get("sourceContainerBookingId")) == self.booking_id + else: + selected = isinstance(payload, dict) and str(payload.get("loadingPlanId")) == self.plan_id + qualifies = event.get("eventType") == self.event_type and selected envelope_hash = hashlib.sha256(json.dumps(event, sort_keys=True, ensure_ascii=False, separators=(",", ":")).encode("utf-8")).hexdigest() with self.lock: @@ -254,6 +259,7 @@ def main(): selector = parser.add_mutually_exclusive_group(required=True) selector.add_argument("--event-id") selector.add_argument("--loading-plan-id") + selector.add_argument("--container-booking-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) @@ -270,9 +276,11 @@ def main(): parser.error("event-id must be a UUID or ULID") if args.loading_plan_id and not re.fullmatch(r"[1-9][0-9]{0,19}", args.loading_plan_id): 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 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) + state = State(args.event_type, args.event_id, args.loading_plan_id, args.log, args.container_booking_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 ad2828bb3..5d1dde42b 100644 --- a/scripts/event-delivery-fault-proxy.test.mjs +++ b/scripts/event-delivery-fault-proxy.test.mjs @@ -1,5 +1,5 @@ import assert from 'node:assert/strict' -import { spawn } from 'node:child_process' +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' @@ -19,7 +19,7 @@ const event = (id = targetId, planId = '2100000000000000001') => ({ eventId: id, payload: { loadingPlanId: planId, syntheticPrivateField: secret } }) const close = server => new Promise(resolve => server.close(resolve)) -async function fixture(selector = ['--loading-plan-id', '2100000000000000001']) { +async function fixture(selector = ['--loading-plan-id', '2100000000000000001'], eventType = 'LoadingCompleted') { const calls = [] const applied = new Set() let invalidAck = false @@ -40,7 +40,7 @@ async function fixture(selector = ['--loading-plan-id', '2100000000000000001']) const directory = await mkdtemp(join(tmpdir(), 'event-fault-offline-')) const log = join(directory, 'evidence.jsonl') const child = spawn(process.env.PYTHON || 'python', [script, '--upstream', - `http://127.0.0.1:${consumer.address().port}/internal/events`, '--event-type', 'LoadingCompleted', + `http://127.0.0.1:${consumer.address().port}/internal/events`, '--event-type', eventType, ...selector, '--log', log], { env: { ...process.env, EVENT_FAULT_CONTROL_TOKEN: controlToken }, stdio: ['ignore', 'pipe', 'pipe'] }) let output = ''; let errors = '' @@ -179,3 +179,42 @@ test('explicit event-ID selector requires exact event type and forwards unrelate assert.equal(status.attempt, 1) assert.equal(status.eventId, targetId) }) + +test('unloading selector pins only the exact source booking and preserves reverse-event ACK recovery', async t => { + const bookingId = '2100000000000000066' + const proxy = await fixture(['--container-booking-id', bookingId], 'OverseasContainerUnloaded') + t.after(() => proxy.stop()) + await proxy.control('/mode', { mode: 'FAULT' }) + const unloaded = { ...event(), eventType: 'OverseasContainerUnloaded', + payload: { sourceContainerBookingId: bookingId, containerNo: 'TEST-CONTAINER' } } + for (const unrelated of [ + { ...unloaded, eventId: otherId, payload: { sourceContainerBookingId: '2100000000000000067' } }, + { ...unloaded, eventId: otherId, payload: { loadingPlanId: bookingId } }, + { ...unloaded, eventId: otherId, eventType: 'OverseasDeliveryCompleted' }, + ]) assert.equal((await proxy.send(unrelated)).status, 200) + assert.equal((await proxy.send(unloaded)).status, 503) + assert.equal((await proxy.send(unloaded)).status, 503) + assert.equal((await proxy.send({ ...unloaded, eventId: samePlanOtherId })).status, 200) + const before = proxy.calls.length + assert.equal((await proxy.send({ ...unloaded, payload: { sourceContainerBookingId: '2100000000000000067' } })).status, 409) + assert.equal(proxy.calls.length, before) + await proxy.control('/mode', { mode: 'PASS' }) + const recovered = await proxy.send(unloaded) + 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']) +}) + +test('booking selector rejects malformed identities and mixed selection before starting listeners', () => { + for (const args of [['--container-booking-id', '0'], ['--container-booking-id', '../66'], + ['--container-booking-id', '66', '--loading-plan-id', '77'], ['--container-booking-id', '66', '--event-id', targetId]]) { + const result = spawnSync(process.env.PYTHON || 'python', [script, '--upstream', 'http://127.0.0.1:1/internal/events', + '--event-type', 'OverseasContainerUnloaded', ...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-evidence.mjs b/scripts/fba-core-regression-evidence.mjs new file mode 100644 index 000000000..d0457d08c --- /dev/null +++ b/scripts/fba-core-regression-evidence.mjs @@ -0,0 +1,185 @@ +// Internal explicit TEST evidence queries; no network or environment reads on import. +const stages = new Set(['BEFORE_COMPLETE', 'AFTER_COMPLETE', 'AFTER_SAME_KEY_REPLAY', 'AFTER_FRESH_KEY_REJECTION', 'AFTER_EVENT_RECOVERY']) +const numeric = new Set(['availablePackages', 'holdPackages', 'allocatedPackages', 'pickedPackages', 'inTransitPackages', 'receivingPackages', 'procurementPackages', 'receivedPackages', 'beforePackages', 'deltaPackages', 'afterPackages', 'packageCount']) +const fields = { + sources: 'id customerId overseasWarehouseId sourceContainerBookingId containerNo sourceInboundId inboundCode sourceRegion packageCount', + sessions: 'id sourceContainerBookingId containerNo overseasWarehouseId', + balances: 'id customerId overseasWarehouseId destinationWarehouseId destinationWarehouseCode inboundOrderId inboundOrderNo inboundCode skuCode availablePackages holdPackages allocatedPackages pickedPackages inTransitPackages receivingPackages procurementPackages', + positions: 'id balanceId customerId overseasWarehouseId destinationWarehouseCode palletId palletNo batchNo warehouseLocationId warehouseLocationCode status availablePackages holdPackages allocatedPackages pickedPackages receivingPackages', + stockReceipts: 'id customerId planId taskId allocationId lineId orderId inboundCode skuCode warehousePalletId palletNo warehouseLocationId warehouseLocationCode batchNo receivedPackages inventoryBalanceId inventoryPositionId inventoryOperationKey', + inventoryOperations: 'id customerId idempotencyKey operationType requestHash status', + stockInLedger: 'id customerId balanceId positionId overseasWarehouseId inboundOrderId inboundCode palletId palletNo batchNo skuCode businessType businessOrderNo sourceEventId idempotencyKey statusFrom statusTo beforePackages deltaPackages afterPackages', + receivingOperations: 'id sessionId action idempotencyKey requestHash', +} +const snake = value => value.replace(/[A-Z]/g, letter => '_' + letter.toLowerCase()) +function identifier(value) { + if (typeof value !== 'string' || !/^[1-9][0-9]{0,19}$/.test(value)) throw Error('Invalid receiving evidence identifier') + return value +} +// UTF-8 hex literals avoid quote, backslash and SQL-mode ambiguity. +const literal = value => `(CONVERT(0x${Buffer.from(value, 'utf8').toString('hex')} USING utf8mb4) COLLATE utf8mb4_bin)` +function validateScope(scope) { + for (const name of ['customerId', 'warehouseId', 'bookingId']) identifier(scope?.[name]) + if (!stages.has(scope.stage) || typeof scope.containerNo !== 'string' || !scope.containerNo.trim() + || !Array.isArray(scope.inboundCodes) || !scope.inboundCodes.length + || scope.inboundCodes.some(code => typeof code !== 'string' || !code.trim() || code.length > 128) + || new Set(scope.inboundCodes).size !== scope.inboundCodes.length + || !Array.isArray(scope.sourceInboundIds) || scope.sourceInboundIds.length !== scope.inboundCodes.length + || new Set(scope.sourceInboundIds).size !== scope.sourceInboundIds.length) throw Error('Invalid receiving evidence scope') + scope.sourceInboundIds.forEach(identifier) + for (const code of scope.inboundCodes) if (!Number.isSafeInteger(scope.expectedPackagesByInbound?.[code]) + || scope.expectedPackagesByInbound[code] <= 0) throw Error('Invalid receiving evidence quantity') +} +export function buildReceivingEvidenceSql(scope) { + validateScope(scope) + const owned = alias => `${alias}.customer_id=${literal(scope.customerId)} AND ${alias}.overseas_warehouse_id=${literal(scope.warehouseId)}` + const source = `o.customer_id=${literal(scope.customerId)} AND o.source_container_booking_id=${literal(scope.bookingId)}` + const orderIds = `SELECT o.id FROM ll_order o WHERE ${source}` + const session = `s.source_container_booking_id=${literal(scope.bookingId)} AND s.container_no=${literal(scope.containerNo)} AND s.overseas_warehouse_id=${scope.warehouseId}` + const taskIds = `SELECT t.id FROM ll_inbound_pda_task t JOIN ll_inbound_plan p ON p.id=t.plan_id JOIN ll_order o ON o.id=p.source_order_id WHERE ${source} AND BINARY t.customer_id=BINARY o.customer_id AND BINARY p.customer_id=BINARY o.customer_id` + const query = (name, alias, from, where) => { + const pairs = fields[name].split(' ').map(field => { + const column = `${alias}.${snake(field)}` + return `'${field}',${numeric.has(field) ? column : `CAST(${column} AS CHAR)`}` + }).join(',') + return `SELECT JSON_OBJECT('collection','${name}','row',JSON_OBJECT(${pairs})) FROM ${from} WHERE ${where} ORDER BY ${alias}.id;` + } + return [ + 'SET NAMES utf8mb4 COLLATE utf8mb4_bin;', + 'SET SESSION TRANSACTION ISOLATION LEVEL REPEATABLE READ;', + 'START TRANSACTION WITH CONSISTENT SNAPSHOT, READ ONLY;', + query('sources', 'o', 'll_order o', source), + query('sessions', 's', 'll_receiving_container_session s', session), + query('balances', 'b', 'll_inventory_balance b', owned('b')), + query('positions', 'p', 'll_inventory_position p', owned('p')), + query('stockReceipts', 'r', 'll_inbound_stock_receipt r', `r.customer_id=${literal(scope.customerId)} AND r.order_id IN (${orderIds})`), + query('inventoryOperations', 'i', 'll_inventory_operation i', `i.customer_id=${literal(scope.customerId)} AND i.operation_type=${literal('STOCK_IN')} AND EXISTS (SELECT 1 FROM ll_inbound_pda_task t WHERE t.id IN (${taskIds}) AND i.idempotency_key LIKE (CONCAT('PDA_STOCK_IN:',t.id,':%') COLLATE utf8mb4_bin))`), + query('stockInLedger', 'l', 'll_inventory_ledger l', `${owned('l')} AND l.business_type='STOCK_IN' AND l.inbound_order_id IN (${orderIds})`), + query('receivingOperations', 'r', 'll_receiving_container_operation r', `r.action='COMPLETE' AND r.session_id IN (SELECT s.id FROM ll_receiving_container_session s WHERE ${session})`), + 'COMMIT;', + ].join('\n') +} +export function receivingEvidenceFromRows(scope, rows) { + validateScope(scope) + const result = Object.fromEntries(Object.keys(fields).map(name => [name, []])) + if (!Array.isArray(rows)) throw Error('Receiving evidence rows missing') + for (const item of rows) { + if (!Object.hasOwn(result, item?.collection) || !item.row) throw Error('Invalid receiving evidence collection') + result[item.collection].push(item.row) + } + const sources = result.sources + if (sources.length !== scope.inboundCodes.length + || new Set(sources.map(row => row.inboundCode)).size !== sources.length + || new Set(sources.map(row => row.sourceInboundId)).size !== sources.length + || sources.some(row => row.customerId !== scope.customerId || row.overseasWarehouseId !== scope.warehouseId + || row.sourceContainerBookingId !== scope.bookingId || row.containerNo !== scope.containerNo + || row.sourceRegion !== 'FIRSTLEG' || !scope.sourceInboundIds.includes(row.sourceInboundId) + || !scope.inboundCodes.includes(row.inboundCode) || row.packageCount !== scope.expectedPackagesByInbound[row.inboundCode]) + || result.sessions.length !== 1 || result.sessions.some(row => row.sourceContainerBookingId !== scope.bookingId + || row.containerNo !== scope.containerNo || row.overseasWarehouseId !== scope.warehouseId)) throw Error('Receiving evidence source/session scope mismatch') + delete result.sources + delete result.sessions + for (const rows of Object.values(result)) rows.sort((a, b) => String(a.id).localeCompare(String(b.id))) + result.palletComposition = result.positions.map(row => ({ ...row })) + return result +} +export function createReceivingEvidenceReader(executeReadOnly) { + if (typeof executeReadOnly !== 'function') throw Error('Receiving evidence transport missing') + return async scope => receivingEvidenceFromRows(scope, await executeReadOnly(buildReceivingEvidenceSql(scope))) +} + +const transactionStart = ['SET NAMES utf8mb4 COLLATE utf8mb4_bin;', + 'SET SESSION TRANSACTION ISOLATION LEVEL REPEATABLE READ;', + 'START TRANSACTION WITH CONSISTENT SNAPSHOT, READ ONLY;'] +export function assertReadOnlySql(sql) { + if (typeof sql !== 'string' || /--|\/\*|\b(?:INSERT|UPDATE|DELETE|REPLACE|DROP|ALTER|TRUNCATE|CALL|EXECUTE|PREPARE|INTO|OUTFILE|DUMPFILE|LOAD_FILE|SLEEP|BENCHMARK|GET_LOCK)\b/i.test(sql)) + throw Error('Read-only SQL rejected') + const lines = sql.split('\n').filter(Boolean) + if (!transactionStart.every((line,i) => lines[i] === line) || lines.at(-1) !== 'COMMIT;' + || lines.length < 5 || lines.slice(3,-1).some(line => !/^SELECT [^;]+;$/.test(line))) throw Error('Read-only transaction required') + return sql +} +export function buildCompletionEvidenceSql(scope) { + identifier(scope?.customerId); identifier(scope?.rootOrderId) + if (scope.eventType !== 'FIRST_LEG_ORDER_FULFILLMENT_COMPLETED' || !Array.isArray(scope.sourceOrderIds) + || scope.sourceOrderIds.length !== 2 || new Set(scope.sourceOrderIds).size !== 2 + || !scope.sourceOrderIds.includes(scope.rootOrderId)) throw Error('Invalid completion evidence scope') + scope.sourceOrderIds.forEach(identifier) + return [...transactionStart, + `SELECT JSON_OBJECT('orderId',CAST(o.id AS CHAR),'customerId',CAST(o.tenant_id AS CHAR),'rootOrderId',CAST(o.root_order_id AS CHAR),'eventType','FIRST_LEG_ORDER_FULFILLMENT_COMPLETED','count',COUNT(x.id),'publishedCount',COALESCE(SUM(x.publish_status='PUBLISHED'),0),'pendingCount',COALESCE(SUM(x.publish_status='PENDING'),0),'failedCount',COALESCE(SUM(x.publish_status NOT IN ('PENDING','PUBLISHED')),0)) FROM ord_order o LEFT JOIN ord_outbox_event x ON x.aggregate_id=o.id AND x.tenant_id=o.tenant_id AND x.event_type='FIRST_LEG_ORDER_FULFILLMENT_COMPLETED' WHERE o.tenant_id=${scope.customerId} AND o.root_order_id=${scope.rootOrderId} AND o.id IN (${scope.sourceOrderIds.join(',')}) GROUP BY o.id,o.tenant_id,o.root_order_id ORDER BY o.id;`, + 'COMMIT;'].join('\n') +} +export function completionEvidenceFromRows(scope, rows) { + buildCompletionEvidenceSql(scope) + if (!Array.isArray(rows) || rows.length !== 2 || new Set(rows.map(row => row.orderId)).size !== 2 + || rows.some(row => !scope.sourceOrderIds.includes(row.orderId) || row.customerId !== scope.customerId + || row.rootOrderId !== scope.rootOrderId || row.eventType !== scope.eventType + || ['count','publishedCount','pendingCount','failedCount'].some(key => !Number.isSafeInteger(row[key]) || row[key] < 0))) + throw Error('Completion evidence scope or counts mismatch') + return {readOnly:true,customerId:scope.customerId,rootOrderId:scope.rootOrderId,orders:rows} +} +export const EVIDENCE_ENV_KEYS = ['JENKINS_URL','JENKINS_USER','JENKINS_API_TOKEN','FBA_CORE_JENKINS_AGENT'] +export function validateEvidenceConfig(env) { + if (EVIDENCE_ENV_KEYS.some(key => typeof env[key] !== 'string' || !env[key].trim())) throw Error('Missing TEST evidence configuration') + const url = new URL(env.JENKINS_URL) + if (!['http:','https:'].includes(url.protocol) || url.username || url.password || url.search || url.hash + || !/^[A-Za-z0-9_.-]{1,128}$/.test(env.FBA_CORE_JENKINS_AGENT)) throw Error('Invalid TEST evidence configuration') +} +const base64Expression = value => '['+Buffer.from(value).toString('base64').match(/.{1,4000}/g).map(part=>`'${part}'`).join(',')+"].join('')" +// Credentials stay in the executor's memory. No deploy, SQL mutation, or arbitrary SQL entry point. +export function buildEvidenceRemoteScript(service, sql, agent) { + assertReadOnlySql(sql) + if (!['order','lastleg'].includes(service) || !/^[A-Za-z0-9_.-]{1,128}$/.test(agent)) throw Error('Invalid TEST evidence target') + const prefix=service.toUpperCase(),container=`firstleg-test-${service}` + const python = `import json,os,subprocess,urllib.parse,base64,re +try: + c=json.loads(subprocess.check_output(['docker','inspect','${container}'],text=True,stderr=subprocess.DEVNULL))[0] + if c.get('Name')!='/${container}' or c.get('Config',{}).get('Labels',{}).get('com.docker.compose.project')!='firstleg-test': raise RuntimeError('target') + e=dict(v.split('=',1) for v in c['Config']['Env']) + u=urllib.parse.urlsplit(e['${prefix}_DB_URL'].removeprefix('jdbc:')) + if not u.hostname or not u.path.lstrip('/'): raise RuntimeError('database') + childEnv=dict(os.environ,MYSQL_PWD=e['${prefix}_DB_PASSWORD']) + host='127.0.0.1' if u.hostname=='host.docker.internal' else u.hostname + sql=base64.b64decode('${Buffer.from(sql).toString('base64')}').decode('utf-8') + p=subprocess.run(['mysql','--host='+host,'--port='+str(u.port or 3306),'--user='+e['${prefix}_DB_USERNAME'],'--batch','--raw','--skip-column-names',u.path.lstrip('/')],input=sql,capture_output=True,text=True,env=childEnv,timeout=40) + if p.returncode: + code=re.search(r'ERROR ([0-9]{3,5})',p.stderr) + print('FBA_CORE_JSON='+json.dumps({'ok':False,'error':'MYSQL_'+(code.group(1) if code else 'UNKNOWN')})) + else: print('FBA_CORE_JSON='+json.dumps({'ok':True,'rows':[json.loads(line) for line in p.stdout.splitlines() if line.strip()]})) +except Exception: + print('FBA_CORE_JSON='+json.dumps({'ok':False,'error':'REMOTE_QUERY_FAILED'})) +` + const groovy=`def p=new ProcessBuilder(['python3','-c',new String(java.util.Base64.decoder.decode(${base64Expression(python)}),'UTF-8')]).redirectErrorStream(true).start(); def output=p.inputStream.getText('UTF-8'); p.waitFor(); print(output)` + return `def c=jenkins.model.Jenkins.instance.getComputer('${agent}'); if(c==null || c.channel==null) throw new RuntimeException('TEST evidence agent unavailable'); print(hudson.util.RemotingDiagnostics.executeGroovy(new String(java.util.Base64.decoder.decode(${base64Expression(groovy)}),'UTF-8'),c.channel));` +} +export function createTestEvidenceReaders(env, fetcher = globalThis.fetch) { + validateEvidenceConfig(env) + async function execute(service, sql) { + const script=buildEvidenceRemoteScript(service,sql,env.FBA_CORE_JENKINS_AGENT) + let response + try { + response=await fetcher(new URL('scriptText',env.JENKINS_URL.replace(/\/?$/,'/')),{ + method:'POST',headers:{Authorization:'Basic '+Buffer.from(env.JENKINS_USER+':'+env.JENKINS_API_TOKEN).toString('base64'), + 'Content-Type':'application/x-www-form-urlencoded'},body:new URLSearchParams({script}),signal:AbortSignal.timeout(55000)}) + } catch { throw Error('TEST evidence transport failed') } + if (!response.ok) throw Error('TEST evidence HTTP '+response.status) + let raw + try { raw=await response.text() } catch { throw Error('TEST evidence response failed') } + const lines=raw.split(/\r?\n/).filter(line=>line.startsWith('FBA_CORE_JSON=')) + if(lines.length!==1) throw Error('TEST evidence JSON missing or ambiguous') + let result + try { result=JSON.parse(lines[0].slice('FBA_CORE_JSON='.length)) } catch { throw Error('TEST evidence JSON invalid') } + if(result?.ok!==true || !Array.isArray(result.rows)) throw Error(/^MYSQL_[0-9]{3,5}$/.test(result?.error)?result.error:'TEST evidence query failed') + return result.rows + } + return { + readReceiving: createReceivingEvidenceReader(sql => execute('lastleg',sql)), + readCompletion: async scope => completionEvidenceFromRows(scope,await execute('order',buildCompletionEvidenceSql(scope))), + preflight: async () => { + for(const service of ['order','lastleg']) { + const rows=await execute(service,[...transactionStart,"SELECT JSON_OBJECT('testProbe',1);",'COMMIT;'].join('\n')) + if(rows.length!==1 || rows[0].testProbe!==1)throw Error('TEST evidence preflight failed') + } + }, + } +} diff --git a/scripts/fba-core-regression-evidence.test.mjs b/scripts/fba-core-regression-evidence.test.mjs new file mode 100644 index 000000000..60fb68ec2 --- /dev/null +++ b/scripts/fba-core-regression-evidence.test.mjs @@ -0,0 +1,60 @@ +import test from 'node:test' +import assert from 'node:assert/strict' +import { assertReadOnlySql, buildReceivingEvidenceSql, receivingEvidenceFromRows, buildCompletionEvidenceSql, + completionEvidenceFromRows, buildEvidenceRemoteScript, createTestEvidenceReaders } from './fba-core-regression-evidence.mjs' + +const env=()=>({JENKINS_URL:'https://ci.example.invalid/',JENKINS_USER:'offline-ci',JENKINS_API_TOKEN:'offline-api-secret',FBA_CORE_JENKINS_AGENT:'java-test-agent'}) +const scope=()=>({customerId:'2105209721220861954',warehouseId:'193000000000200003',bookingId:'2105220190013755394', + stage:'BEFORE_COMPLETE',containerNo:'TEST-ONE',inboundCodes:['IN-1'],sourceInboundIds:['2105220155339444226'], + expectedPackagesByInbound:{'IN-1':1}}) +const completion=()=>({customerId:'90',rootOrderId:'100',sourceOrderIds:['100','101'],eventType:'FIRST_LEG_ORDER_FULFILLMENT_COMPLETED'}) +const completionRows=()=>['100','101'].map(orderId=>({customerId:'90',rootOrderId:'100',orderId, + eventType:'FIRST_LEG_ORDER_FULFILLMENT_COMPLETED',count:orderId==='100'?1:0,publishedCount:orderId==='100'?1:0,pendingCount:0,failedCount:0})) + +test('both queries use consistent read-only transactions and exact scope without losing long identifiers',()=>{ + const sql=buildReceivingEvidenceSql(scope()) + assert.equal(buildReceivingEvidenceSql({...scope(),stage:'AFTER_EVENT_RECOVERY'}),sql) + assert.throws(()=>buildReceivingEvidenceSql({...scope(),stage:'UNKNOWN_RECOVERY'})) + assert.equal(assertReadOnlySql(sql),sql) + assert.match(sql,/START TRANSACTION WITH CONSISTENT SNAPSHOT, READ ONLY/) + assert.ok(sql.includes(Buffer.from(scope().customerId).toString('hex'))) + assert.doesNotMatch(sql,/customer_id=2105209721220861954/) + assert.match(sql,/CAST\(b.id AS CHAR\)/) + assert.match(assertReadOnlySql(buildCompletionEvidenceSql(completion())),/LEFT JOIN ord_outbox_event/) +}) +test('read-only guard rejects mutation, multiple statements, lock or unsafe select effects',()=>{ + const sql=buildCompletionEvidenceSql(completion()) + for(const statement of ['DELETE FROM ord_order;','SELECT 1 INTO OUTFILE \'x\';','SELECT SLEEP(1);', + 'SELECT 1; DROP TABLE x;','SET SESSION autocommit=1;'])assert.throws(()=>assertReadOnlySql(sql.replace(/SELECT .+;/,statement))) + assert.throws(()=>assertReadOnlySql('SELECT 1;')) + assert.throws(()=>buildEvidenceRemoteScript('prod',sql,'test')) +}) +test('invalid and injected IDs fail before transport; cancelled child zero evidence cannot disappear',()=>{ + assert.throws(()=>buildReceivingEvidenceSql({...scope(),bookingId:'1 OR 1=1'})) + assert.throws(()=>buildCompletionEvidenceSql({...completion(),sourceOrderIds:['100','100']})) + assert.equal(completionEvidenceFromRows(completion(),completionRows()).orders[1].count,0) + for(const mutate of [r=>r.pop(),r=>r[1].customerId='91',r=>r[1].count=null,r=>r[1].orderId='100']) { + const rows=completionRows();mutate(rows);assert.throws(()=>completionEvidenceFromRows(completion(),rows)) + } +}) +test('receiving rows require exact source and session; empty or foreign scope fails',()=>{ + assert.throws(()=>receivingEvidenceFromRows(scope(),[])) + const s=scope(),rows=[{collection:'sources',row:{id:'1',customerId:s.customerId,overseasWarehouseId:s.warehouseId, + sourceContainerBookingId:s.bookingId,containerNo:s.containerNo,sourceInboundId:s.sourceInboundIds[0],inboundCode:'IN-1',sourceRegion:'FIRSTLEG',packageCount:1}}, + {collection:'sessions',row:{id:'2',sourceContainerBookingId:s.bookingId,containerNo:s.containerNo,overseasWarehouseId:s.warehouseId}}] + assert.deepEqual(receivingEvidenceFromRows(s,rows).balances,[]) + rows[0].row.customerId='99';assert.throws(()=>receivingEvidenceFromRows(s,rows)) +}) +test('transport frames valid empty rows and suppresses raw remote and connection secrets',async()=>{ + let calls=0 + const transport=createTestEvidenceReaders(env(),async()=>{calls++;return {ok:true,text:async()=> 'FBA_CORE_JSON='+JSON.stringify({ok:true,rows:[{testProbe:1}]})}}) + await transport.preflight();assert.equal(calls,2) + for(const output of ['secret-leak','FBA_CORE_JSON={bad','FBA_CORE_JSON={"ok":false,"error":"password=secret-leak"}']) { + const reader=createTestEvidenceReaders(env(),async()=>({ok:true,text:async()=>output})) + await assert.rejects(reader.preflight(),error=>!error.message.includes('secret-leak')) + } + const reader=createTestEvidenceReaders(env(),async()=>{throw Error('offline-api-secret')}) + await assert.rejects(reader.preflight(),/TEST evidence transport failed/) + const script=buildEvidenceRemoteScript('order',buildCompletionEvidenceSql(completion()),'java-test-agent') + assert.equal(script.includes(env().JENKINS_API_TOKEN),false) +}) diff --git a/scripts/phase4-overseas-acceptance-entry-contract.test.mjs b/scripts/phase4-overseas-acceptance-entry-contract.test.mjs index 5a1e18355..9d8498e37 100644 --- a/scripts/phase4-overseas-acceptance-entry-contract.test.mjs +++ b/scripts/phase4-overseas-acceptance-entry-contract.test.mjs @@ -2,10 +2,69 @@ import test from 'node:test' import assert from 'node:assert/strict' import { readFileSync } from 'node:fs' import { resolve } from 'node:path' -import { AcceptanceRun, parseArguments, validateExecutionOptions } from './admin-test-env-acceptance.mjs' +import { AcceptanceRun, configureReadOnlyEvidence, unloadRecoveryWasVerified, parseArguments, validateExecutionOptions } from './admin-test-env-acceptance.mjs' const source = readFileSync(resolve('scripts/admin-test-env-acceptance.mjs'), 'utf8') +test('full CLI evidence preflight attaches readers and fails before business writes when unavailable', async () => { + const run = { options: { modules: ['manual-state-matrix'] }, context: {} } + const readReceiving = () => {}, readCompletion = () => {} + let checked = false + await configureReadOnlyEvidence(run, {}, () => ({ readReceiving, readCompletion, + preflight: async () => { checked = true } })) + assert.equal(checked, true) + assert.equal(run.context.phase4ReceivingEvidenceReader, readReceiving) + assert.equal(run.context.f40ReadOnlyOutboxEvidence, readCompletion) + const unavailable = { options: run.options, context: {} } + await assert.rejects(configureReadOnlyEvidence(unavailable, {}, () => ({ + preflight: async () => { throw Error('evidence unavailable') } })), /unavailable/) + assert.deepEqual(unavailable.context, {}) +}) + +test('explicit unloading fault uses a separate first-container hook and restores both directions on failure', async () => { + const calls = [] + const run = { marker: 'TEST-UNLOAD', options: { modules: ['phase4-overseas-pda', 'phase4-cross-region-events'], + verifyUnloadEventRecovery: true }, context: {} } + const fault = name => ({ prepare: async scope => calls.push([name, 'prepare', scope]), + readProof: async scope => ({ name, scope, restored: true }), + cleanup: async () => { calls.push([name, 'cleanup']); if (name === 'reverse') throw Error('recovery failed') } }) + const cleanup = await configureReadOnlyEvidence(run, {}, () => ({ readReceiving() {}, readCompletion() {}, preflight: async () => {} }), + { createTestFaultExecutor: () => fault('forward'), createTestUnloadFaultExecutor: () => fault('reverse') }) + assert.equal(calls.length, 0) + await run.context.beforeReceivingComplete({ customerId: '7', bookingId: '66' }) + await run.context.beforeReceivingComplete({ customerId: '8', bookingId: '77' }) + assert.deepEqual(calls, [['reverse', 'prepare', { marker: 'TEST-UNLOAD', customerId: '7', containerBookingId: '66' }]]) + assert.equal(await run.context.afterReceivingComplete({ customerId: '8', bookingId: '77' }), null) + const proof = await run.context.afterReceivingComplete({ customerId: '7', bookingId: '66' }) + assert.equal(proof.name, 'reverse') + await assert.rejects(cleanup.cleanup(), /恢复失败/) + assert.deepEqual(calls.slice(-2), [['reverse', 'cleanup'], ['forward', 'cleanup']]) +}) + +test('unloading recovery is opt-in and cannot run against unrelated modules', () => { + const args = ['--modules', 'phase4-overseas-return-followups', '--gateway', 'http://localhost:9000', + '--username', 'contract', '--password', 'contract'] + assert.equal(parseArguments(args).verifyUnloadEventRecovery, false) + assert.throws(() => validateExecutionOptions(parseArguments([...args, '--verify-unload-event-recovery', 'true'])), /收货链/) +}) + +test('requested unloading fault cannot pass without exact current-run recovery and inventory proof', () => { + const run = { marker: 'CURRENT', createdResources: {} } + assert.equal(unloadRecoveryWasVerified(run), false) + const row = { status: 'VERIFIED', eventRecoveryInventoryUnchanged: true, + eventRecovery: { restored: true, marker: 'CURRENT', containerBookingId: '66' } } + run.createdResources.phase4OverseasPda = { receivingCompletionEvidence: { '66': row } } + assert.equal(unloadRecoveryWasVerified(run), true) + for (const change of [r => r.eventRecoveryInventoryUnchanged = false, r => r.status = 'UNPROVEN', + r => r.eventRecovery.restored = false, r => r.eventRecovery.marker = 'OLD', r => r.eventRecovery.containerBookingId = '77']) { + const changed = structuredClone(run) + change(changed.createdResources.phase4OverseasPda.receivingCompletionEvidence['66']) + assert.equal(unloadRecoveryWasVerified(changed), false) + } + run.createdResources.phase4OverseasPda.receivingCompletionEvidence['77'] = structuredClone(row) + assert.equal(unloadRecoveryWasVerified(run), false) +}) + test('F40 optional chain always creates its customer and rejects absent domestic PDA credentials', () => { const options = parseArguments(['--modules', 'phase5-root-fulfillment', '--gateway', 'http://localhost:9000', '--username', 'contract-user', '--password', 'contract-password', diff --git a/scripts/test-event-fault-executor.mjs b/scripts/test-event-fault-executor.mjs new file mode 100644 index 000000000..d808040e1 --- /dev/null +++ b/scripts/test-event-fault-executor.mjs @@ -0,0 +1,157 @@ +// Internal single TEST fault lifecycle. Uses the existing ACK-loss proxy and +// authenticated original-DEAD retry endpoint, never direct business SQL writes. +import { readFileSync } from 'node:fs' +import { createHash } from 'node:crypto' +import { isDeepStrictEqual } from 'node:util' +import { validateEvidenceConfig } from './fba-core-regression-evidence.mjs' + +const runtime = readFileSync(new URL('./test-event-fault-runtime.py', import.meta.url), 'utf8') +const proxySource = readFileSync(new URL('./event-delivery-fault-proxy.py', import.meta.url)).toString('base64') +const encode = value => Buffer.from(value).toString('base64') +const expression = value => '[' + encode(value).match(/.{1,4000}/g).map(part => `'${part}'`).join(',') + "].join('')" + +export function renderTestFaultAction(action, scope, agent) { + return renderAction(action,scope,agent,'LOADING') +} +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'] + 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)))) { + throw Error('Invalid TEST fault action or scope') + } + const config = encode(JSON.stringify({ action, scope, direction, ...(action === 'prepare' ? { proxySource } : {}) })) + const groovy = `def p=new ProcessBuilder(['python3','-c',new String(java.util.Base64.decoder.decode(${expression(runtime)}),'UTF-8'),'${config}']).redirectErrorStream(true).start();def output=p.inputStream.getText('UTF-8');p.waitFor();print(output)` + return `def c=jenkins.model.Jenkins.instance.getComputer('${agent}');if(c==null || c.channel==null)throw new RuntimeException('TEST fault agent unavailable');print(hudson.util.RemotingDiagnostics.executeGroovy(new String(java.util.Base64.decoder.decode(${expression(groovy)}),'UTF-8'),c.channel))` +} + +export function createTestFaultExecutor(env, fetcher = globalThis.fetch) { + return createExecutor(env,fetcher,'LOADING') +} +export function createTestUnloadFaultExecutor(env, fetcher = globalThis.fetch) { + return createExecutor(env,fetcher,'UNLOAD') +} +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 mysqlDatetimeMillisFromInstant(value) { + const match=typeof value==='string'&&value.match(/^(\d{4}-\d{2}-\d{2}T\d{2}:\d{2}:\d{2})(?:\.(\d{1,9}))?Z$/) + if(!match)return NaN + const seconds=Date.parse(match[1]+'Z') + const nanos=Number((match[2]||'').padEnd(9,'0')) + // DATETIME(3) rounds fractional seconds; Date.parse alone truncates them. + return seconds+Math.floor((nanos+500000)/1000000) +} +export function assertTestUnloadFaultProof(proof,scope) { + const require=(passed,code)=>{if(!passed)throw Error('Invalid unload fault proof: '+code)} + require(proof?.readOnly===true&&Object.keys(scope).every(key=>proof[key]===scope[key]),'SCOPE') + const before=proof.beforeRetry,after=proof.afterRetry,row=proof.producer,env=row?.envelope + require(row&&env&&/^[1-9][0-9]{0,19}$/.test(row.entryId)&&/^(?:[0-9a-fA-F-]{36}|[0-9A-HJKMNP-TV-Z]{26})$/.test(row.eventId),'IDENTITY') + require(row.eventType==='OverseasContainerUnloaded'&&row.customerId==='0'&&row.aggregateId===scope.containerBookingId + &&row.eventKey===`OverseasContainerUnloaded:${scope.containerBookingId}`&&row.aggregateType==='CONTAINER_BOOKING','PRODUCER') + require(env.eventId===row.eventId&&env.eventType===row.eventType&&Number(env.eventVersion)===1&&env.tenantId==='0' + &&env.aggregateId===scope.containerBookingId&&env.aggregateType==='CONTAINER_BOOKING' + &&env.payload?.sourceContainerBookingId===scope.containerBookingId&&Boolean(env.payload.containerNo),'ENVELOPE') + const hash='sha256:'+createHash('sha256').update(JSON.stringify(canonical(env.payload))).digest('hex') + require(row.payloadHash===hash&&env.payloadHash===hash,'HASH') + require(before?.publishStatus==='DEAD'&&before.failedAttempts===12&&row.publishStatus==='PUBLISHED' + &&before.entryId===row.entryId&&before.eventId===row.eventId&&before.payloadHash===hash + &&isDeepStrictEqual(before.envelope,env)&&after?.publishStatus==='PUBLISHED'&&after.entryId===row.entryId + &&after.eventId===row.eventId&&after.payloadHash===hash&&isDeepStrictEqual(after.envelope,env),'ORIGINAL_RETRY') + const inbox=proof.consumer + require(inbox?.count===1&&inbox.status==='PROCESSED'&&inbox.eventId===row.eventId&&inbox.payloadHash===hash + &&inbox.eventType===row.eventType&&inbox.aggregateId===scope.containerBookingId&&isDeepStrictEqual(inbox.payload,env.payload),'INBOX') + const attempts=proof.attempts||[],proxy=proof.proxyRecords||[] + require(attempts.length===13&&new Set(attempts.map(a=>a.attemptId)).size===13&&attempts.every(a=>a.eventId===row.eventId&&a.entryId===row.entryId&&a.serviceName==='lastleg') + &&attempts.slice(0,11).every(a=>a.outcome==='RETRY')&&attempts[11].outcome==='DEAD'&&attempts[12].outcome==='DUPLICATE','ATTEMPTS') + const matches=proxy.filter(p=>p.matched===true) + require(matches.length===13&&matches.every((p,i)=>p.eventId===row.eventId&&p.payloadHash===hash&&p.attempt===i+1) + &&matches.slice(0,12).every((p,i)=>p.errorCode==='ACK_LOSS_INJECTED'&&p.outcome===(i===0?'APPLIED':'DUPLICATE')) + &&matches[12].outcome==='DUPLICATE'&&matches[12].errorCode==null,'REAL_ACK_LOSS') + const audit=proof.retryAudit + require(audit&&proof.retryAuditCount===1&&audit.requestId===proof.retryRequestId&&audit.entryId===row.entryId + &&audit.eventId===row.eventId&&audit.payloadHash===hash&&audit.previousFailures===12,'RETRY_AUDIT') + const container=before.container + require(container&&container.id===scope.containerBookingId&&container.containerNo===env.payload.containerNo + &&isDeepStrictEqual(container,after.container)&&container.version!=null&&container.emptyReportSourceVersion!=null,'CONTAINER_UNCHANGED') + const time=value=>Date.parse(String(value).replace(' ','T').replace(/Z$/,'')+'Z') + require(time(container.unloadingCompletedAt)===mysqlDatetimeMillisFromInstant(env.payload.unloadingCompletedAt) + &&time(container.unloadingPhotoUploadedAt)===mysqlDatetimeMillisFromInstant(env.payload.unloadingPhotoUploadedAt) + &&time(container.estimatedEmptyReportedAt)-time(container.unloadingCompletedAt)===72*3600000 + &&(env.payload.actualEmptyReportedAt==null?container.actualEmptyReportedAt===null:time(container.actualEmptyReportedAt)===Date.parse(env.payload.actualEmptyReportedAt)) + &&String(container.emptyReportSourceVersion)===String(env.aggregateVersion),'CONTAINER_FACTS') + return proof +} +function createExecutor(env,fetcher,direction) { + validateEvidenceConfig(env) + if (env.CHONGCHONG_ENV !== 'test') throw Error('TEST fault requires CHONGCHONG_ENV=test') + let scope = null, finished = false, preparedEvidence=null, restoration=null + async function action(name) { + const script = renderAction(name, scope, env.FBA_CORE_JENKINS_AGENT,direction) + const response = await fetcher(new URL('scriptText', env.JENKINS_URL.replace(/\/?$/, '/')), { + method: 'POST', redirect: 'error', headers: { Authorization: 'Basic ' + encode(env.JENKINS_USER + ':' + env.JENKINS_API_TOKEN), + 'Content-Type': 'application/x-www-form-urlencoded' }, body: new URLSearchParams({ script }), signal: AbortSignal.timeout(55000) }) + if (!response.ok) throw Error('TEST fault HTTP ' + response.status) + const rows = (await response.text()).split(/\r?\n/).filter(row => row.startsWith('TEST_EVENT_JSON=')) + if (rows.length !== 1) throw Error('TEST fault evidence missing or ambiguous') + const result = JSON.parse(rows[0].slice('TEST_EVENT_JSON='.length)) + if (!result?.ok) throw Error('TEST fault action failed: ' + (/^[A-Z_0-9]{1,80}$/.test(result?.error) ? result.error : 'UNKNOWN')) + return result.data + } + async function wait(name, check, attempts = 120, tolerateStartup = false) { + for (let index = 0; index < attempts; index++) { + let value + try { value = await action(name) } catch (error) { + if (!tolerateStartup || !String(error.message).endsWith('TEST_EVENT_ACTION_FAILED')) throw error + } + if (value && check(value)) return value + await new Promise(resolve => setTimeout(resolve, 1000)) + } + throw Error('TEST fault observation timed out: ' + name) + } + async function cleanup() { + if (!scope || finished) return restoration + await action('restore') + await wait('health', row => row.healthy) + restoration=await action('finish') + if(direction==='UNLOAD'&&!(/^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') + finished = true + return restoration + } + return { + async prepare(next) { + if (scope) { + if(direction==='UNLOAD'&&!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, + loadingPlanId: next.loadingPlanId, containerBookingId: next.containerBookingId } + preparedEvidence=await action('prepare') + await wait('proxy-ready', row => row.ready, 10, true) + await action('activate') + 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') + 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) + await cleanup() + return { ...proof, restored: true, ...(direction==='UNLOAD'?{restoration}:{}) } + } finally { if(direction==='UNLOAD')await cleanup() } + }, + cleanup, + } +} diff --git a/scripts/test-event-fault-executor.test.mjs b/scripts/test-event-fault-executor.test.mjs new file mode 100644 index 000000000..3da9b468c --- /dev/null +++ b/scripts/test-event-fault-executor.test.mjs @@ -0,0 +1,221 @@ +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' + +const scope = { marker: 'TEST-ROOT', customerId: '7', loadingPlanId: '99', containerBookingId: '66' } +const environment = { CHONGCHONG_ENV: 'test', JENKINS_URL: 'https://jenkins.example.invalid/', + JENKINS_USER: 'offline', JENKINS_API_TOKEN: 'offline-secret', FBA_CORE_JENKINS_AGENT: 'java-test-agent' } +function requestedAction(options) { + const script = options.body.get('script') + const chunks = [...script.matchAll(/'([A-Za-z0-9+/=]+)'/g)].map(match => match[1]).join('') + const groovy = Buffer.from(chunks, 'base64').toString('utf8') + const argument = groovy.match(/'UTF-8'\),'([A-Za-z0-9+/=]+)'\]\)/)?.[1] + return JSON.parse(Buffer.from(argument, 'base64').toString('utf8')) +} +function transport(calls, failActivate = false) { + let recovered = false + return async (_url, options) => { + assert.equal(options.redirect, 'error') + const { action } = requestedAction(options) + calls.push(action) + if (action === 'recover') recovered = true + const data = action === 'proxy-ready' ? { ready: true } : action === 'health' ? { healthy: true } + : action === 'status' ? { producer: { publishStatus: recovered ? 'PUBLISHED' : 'DEAD', failedAttempts: 12 } } + : action === 'proof' ? { ...scope, restored: false } : {} + const result = action === 'activate' && failActivate ? { ok: false, error: 'DOCKER_409' } : { ok: true, data } + return { ok: true, text: async () => 'TEST_EVENT_JSON=' + JSON.stringify(result) } + } +} +test('scope and actions are bounded and renderer never embeds caller credentials', () => { + const rendered = renderTestFaultAction('status', scope, environment.FBA_CORE_JENKINS_AGENT) + assert.ok(!rendered.includes(environment.JENKINS_API_TOKEN)) + for (const change of [{ ...scope, customerId: '7;DROP' }, { ...scope, marker: '../other' }]) { + assert.throws(() => renderTestFaultAction('status', change, environment.FBA_CORE_JENKINS_AGENT)) + } + assert.throws(() => renderTestFaultAction('arbitrary', scope, 'java-test-agent')) + assert.throws(() => createTestFaultExecutor({ ...environment, CHONGCHONG_ENV: 'prod' })) +}) + +const unloadScope = { marker: 'TEST-ROOT', customerId: '7', containerBookingId: '66' } +const originalImageId = 'sha256:' + 'a'.repeat(64), originalFingerprint = 'b'.repeat(64) +function unloadProof(nanoseconds = false) { + const payload = { actualEmptyReportedAt: null, containerNo: 'OFFLINE66', sourceContainerBookingId: '66', + unloadingCompletedAt: nanoseconds ? '2026-10-02T01:00:00.136827515Z' : '2026-10-02T01:00:00Z', unloadingPhotoUploadedAt: '2026-10-02T00:59:00Z' } + const payloadHash = 'sha256:' + createHash('sha256').update(JSON.stringify(payload)).digest('hex') + const eventId = '00000000-0000-4000-8000-000000000001', entryId = '123' + const envelope = { eventId, eventType: 'OverseasContainerUnloaded', eventVersion: 1, tenantId: '0', + aggregateId: '66', aggregateType: 'CONTAINER_BOOKING', aggregateVersion: 3, payloadHash, payload } + const row = { entryId, eventId, payloadHash, envelope, eventType: envelope.eventType, customerId: '0', + aggregateId: '66', aggregateType: 'CONTAINER_BOOKING', eventKey: 'OverseasContainerUnloaded:66', publishStatus: 'PUBLISHED' } + const container = { id: '66', containerNo: 'OFFLINE66', version: '9', emptyReportSourceVersion: '3', + unloadingCompletedAt: nanoseconds ? '2026-10-02 01:00:00.137' : '2026-10-02 01:00:00', unloadingPhotoUploadedAt: '2026-10-02 00:59:00', + estimatedEmptyReportedAt: nanoseconds ? '2026-10-05 01:00:00.137' : '2026-10-05 01:00:00', actualEmptyReportedAt: null } + return JSON.parse(JSON.stringify({ ...unloadScope, readOnly: true, producer: row, + beforeRetry: { ...row, publishStatus: 'DEAD', failedAttempts: 12, container }, afterRetry: { ...row, container }, + consumer: { count: 1, status: 'PROCESSED', eventId, payloadHash, eventType: envelope.eventType, aggregateId: '66', payload }, + attempts: Array.from({ length: 13 }, (_, i) => ({ attemptId: `attempt-${i}`, entryId, eventId, serviceName: 'lastleg', outcome: i < 11 ? 'RETRY' : i === 11 ? 'DEAD' : 'DUPLICATE' })), + proxyRecords: Array.from({ length: 13 }, (_, i) => ({ eventId, payloadHash, attempt: i + 1, matched: true, + outcome: i ? 'DUPLICATE' : 'APPLIED', errorCode: i < 12 ? 'ACK_LOSS_INJECTED' : null })), + retryRequestId: '00000000-0000-4000-8000-000000000002', retryAuditCount: 1, + retryAudit: { requestId: '00000000-0000-4000-8000-000000000002', entryId, eventId, payloadHash, previousFailures: 12 } })) +} +function unloadTransport(calls, proof = unloadProof(), finish = {}) { + let recovered = false + return async (_url, options) => { + const config = requestedAction(options), { action } = config + assert.equal(config.direction, 'UNLOAD') + assert.deepEqual(config.scope, unloadScope) + 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('unload factory pins exactly three scope fields and verifies restoration before success', async () => { + const calls = [], executor = createTestUnloadFaultExecutor(environment, unloadTransport(calls)) + await assert.rejects(executor.prepare({ ...unloadScope, eventType: 'Other' }), /Invalid/) + await executor.prepare(unloadScope) + await assert.rejects(executor.prepare({ ...unloadScope, containerBookingId: '67' }), /scope drift/) + await assert.rejects(executor.readProof({ ...unloadScope, eventType: 'Other' }), /scope drift/) + const proof = await executor.readProof(unloadScope) + assert.equal(proof.restored, true) + assert.equal(proof.restoration.currentImageId, originalImageId) + await executor.cleanup() + assert.deepEqual(calls, ['prepare', 'proxy-ready', 'activate', 'health', 'status', 'recover', 'status', 'proof', 'restore', 'health', 'finish']) +}) +test('unload proof rejects changed original event, fabricated ACK, nonunique audit and changed business facts', () => { + assertTestUnloadFaultProof(unloadProof(), unloadScope) + for (const mutate of [ + p => { p.producer.customerId = '7' }, p => { p.producer.eventKey = 'other' }, + p => { p.producer.envelope.tenantId = '7' }, p => { p.producer.envelope.payload.containerNo = 'CHANGED' }, + p => { p.beforeRetry.entryId = '124' }, p => { p.beforeRetry.envelope.aggregateVersion = 4 }, + p => { p.beforeRetry.failedAttempts = 11 }, p => { p.afterRetry.eventId = 'OTHER' }, + p => { p.consumer.count = 2 }, p => { p.consumer.status = 'RECEIVED' }, p => { p.consumer.payload = {} }, + p => { p.attempts.pop() }, p => { p.attempts[0].serviceName = 'firstleg' }, + p => { p.proxyRecords[0].outcome = 'DUPLICATE' }, p => { p.proxyRecords[0].errorCode = null }, + p => { p.proxyRecords[12].attempt = 1 }, p => { p.retryAuditCount = 2 }, + p => { p.retryAudit.requestId = 'OTHER' }, p => { p.afterRetry.container.version = '10' }, + p => { p.beforeRetry.container.estimatedEmptyReportedAt = p.afterRetry.container.estimatedEmptyReportedAt = '2026-10-04 01:00:00' }, + ]) { + const proof = unloadProof(); mutate(proof) + assert.throws(() => assertTestUnloadFaultProof(proof, unloadScope), /Invalid unload fault proof/) + } +}) + +test('unload timestamps follow DATETIME(3) rounding rather than allowing time drift', () => { + const proof=unloadProof(true) + assertTestUnloadFaultProof(proof,unloadScope) + for(const row of [proof.beforeRetry,proof.afterRetry]) { + row.container.unloadingCompletedAt='2026-10-02 01:00:00.138' + row.container.estimatedEmptyReportedAt='2026-10-05 01:00:00.138' + } + assert.throws(()=>assertTestUnloadFaultProof(proof,unloadScope),/CONTAINER_FACTS/) +}) +test('invalid unload proof still restores and incomplete restoration never becomes success', async () => { + const invalid = unloadProof(); invalid.consumer.count = 0 + const calls = [], executor = createTestUnloadFaultExecutor(environment, unloadTransport(calls, invalid)) + await executor.prepare(unloadScope) + await assert.rejects(executor.readProof(unloadScope), /INBOX/) + assert.deepEqual(calls.slice(-3), ['restore', 'health', 'finish']) + for (const change of [{ restored: false }, { healthy: false }, { proxyStopped: false }, + { currentImageId: 'sha256:' + 'c'.repeat(64) }, { currentFingerprint: 'c'.repeat(64) }]) { + const failed = createTestUnloadFaultExecutor(environment, unloadTransport([], unloadProof(), change)) + await failed.prepare(unloadScope) + await assert.rejects(failed.readProof(unloadScope), /restoration evidence mismatch/) + } +}) +test('Python reverse SQL routing, scope isolation and full configuration recovery fail closed offline', () => { + const result = spawnSync('python', ['-c', String.raw` +import ast, copy, importlib.util, io, pathlib, sys +from unittest.mock import patch +sys.dont_write_bytecode = True +path = pathlib.Path('scripts/test-event-fault-runtime.py') +ast.parse(path.read_text(encoding='utf-8')) +spec = importlib.util.spec_from_file_location('fault_runtime', path) +m = importlib.util.module_from_spec(spec); spec.loader.exec_module(m) +scope = {'marker':'TEST-ROOT','customerId':'7','containerBookingId':'66'} +assert m.target_setting({'direction':'UNLOAD'}) == ('FIRSTLEG_SERVICE_URL','http://firstleg:9202') +assert m.scope_directory(scope, 'UNLOAD') != m.scope_directory(dict(scope, loadingPlanId='99')) +try: m.scope_directory(dict(scope, eventType='Other'), 'UNLOAD'); assert False +except RuntimeError as error: assert str(error) == 'INVALID_SCOPE' +queries=[] +with patch.object(m, 'inspect', side_effect=lambda name: name), patch.object(m, 'read_sql', side_effect=lambda name, sql: queries.append((name, sql)) or []): + assert m.producer(scope, 'UNLOAD') is None +assert queries[0][0] == 'firstleg-test-lastleg' +assert "customer_id='0'" in queries[0][1] and "event_key='OverseasContainerUnloaded:66'" in queries[0][1] +assert "JSON_EXTRACT(payload_json,'$')" in queries[0][1] and "publishStatus',status" in queries[0][1] +row={'entryId':'123','eventId':'00000000-0000-4000-8000-000000000001','payloadHash':'sha256:'+'a'*64,'publishStatus':'DEAD','failedAttempts':12,'envelope':{'eventType':'OverseasContainerUnloaded'}} +retry_state={'direction':'UNLOAD','operationsRequestId':'00000000-0000-4000-8000-000000000002'} +retry_container={'Config':{'Env':['EVENT_OPERATIONS_TOKEN=offline']},'NetworkSettings':{'Networks':{'firstleg-test-network':{'IPAddress':'172.20.0.3'}}}} +with patch.object(m,'producer',return_value=row),patch.object(m,'control',return_value=row),patch.object(m,'container_facts',return_value={'version':'9'}),patch.object(m,'business_ids',side_effect=AssertionError('reverse must not query inventory or orders')),patch.object(m,'private_write'),patch.object(m,'inspect',return_value=retry_container),patch.object(m.urllib.request,'urlopen',return_value=io.BytesIO(b'{"code":"200"}')) as http: + assert m.recover(scope,pathlib.Path('/offline'),retry_state)['retrySubmitted'] + assert http.call_args.args[0].full_url == 'http://172.20.0.3:9303/internal/lastleg/event-operations/123/retry' + assert retry_state['beforeRetry']['container']=={'version':'9'} + assert retry_state['beforeRetry']['envelope']==row['envelope'] +row['publishStatus']='PUBLISHED' +queries=[] +with patch.object(m,'producer',return_value=row),patch.object(m,'control',return_value={'records':[]}),patch.object(m,'container_facts',return_value={'version':'9'}),patch.object(m,'business_ids',side_effect=AssertionError('reverse must not query inventory or orders')),patch.object(m,'inspect',side_effect=lambda name:name),patch.object(m,'read_sql',side_effect=lambda name,sql: queries.append((name,sql)) or []),patch.object(m,'private_write') as save: + proof=m.proof(scope,retry_state) + save.assert_called_once_with(m.scope_directory(scope,'UNLOAD')/'proof.json',proof) +assert proof['retryAuditCount']==0 and proof['retryRequestId']==retry_state['operationsRequestId'] +assert [item[0] for item in queries]==['firstleg-test-lastleg','firstleg-test-lastleg','firstleg-test-business'] +assert "WHERE entry_id='123' AND event_type='OverseasContainerUnloaded'" in queries[1][1] +assert 'fl_container_event_inbox' in queries[2][1] +container={'Name':'/firstleg-test-lastleg','Id':'container123456789','Image':'sha256:'+'a'*64, + 'Config':{'Image':'lastleg:test','Env':['FIRSTLEG_SERVICE_URL=http://firstleg:9202','A=1'],'Labels':{'com.docker.compose.project':'firstleg-test'}}, + 'HostConfig':{'Binds':['source:/data:rw'],'PortBindings':{'9303/tcp':[{'HostPort':'9303'}]}}, + 'Mounts':[{'Type':'bind','Source':'/source','Destination':'/data','RW':True}], + 'NetworkSettings':{'Networks':{'firstleg-test-network':{'IPAddress':'172.20.0.3','Aliases':['lastleg-service','container123']}}}, + 'State':{'Health':{'Status':'healthy'}}} +fingerprint=m.fingerprint(container) +multi=copy.deepcopy(container) +multi['NetworkSettings']['Networks']['firstleg-test-network']['MacAddress']='02:42:ac:15:00:04' +multi['NetworkSettings']['Networks']['second-test-network']={'IPAddress':'172.22.0.7','MacAddress':'02:42:ac:16:00:07','Aliases':['lastleg']} +legacy=copy.deepcopy(multi); legacy['Config']['MacAddress']='02:42:ac:16:00:07' +assert m.fingerprint(legacy)==m.fingerprint(multi) +changed=copy.deepcopy(legacy); changed['Config']['MacAddress']='02:42:ac:99:00:01' +assert m.fingerprint(changed)!=m.fingerprint(multi) +changed=copy.deepcopy(legacy); changed['NetworkSettings']['Networks']['second-test-network']['MacAddress']='02:42:ac:16:00:08' +assert m.fingerprint(changed)!=m.fingerprint(multi) +state={'direction':'UNLOAD','restored':False,'image':container['Config']['Image'],'originalImageId':container['Image'], + 'originalFingerprint':fingerprint,'originalBody':m.recreation(container,m.env_of(container)),'proxyPid':123,'controlToken':'offline'} +for mutate in [lambda c:c['HostConfig'].update(Binds=['other:/data:rw']),lambda c:c['NetworkSettings']['Networks']['firstleg-test-network'].update(IPAddress='172.20.0.4'),lambda c:c.update(Image='sha256:'+'b'*64),lambda c:c['State']['Health'].update(Status='unhealthy')]: + changed=copy.deepcopy(container); mutate(changed) + with patch.object(m,'inspect',return_value=changed), patch.object(m,'private_write') as write: + try: m.finish(pathlib.Path('/offline'),copy.deepcopy(state)); assert False + except RuntimeError as error: assert str(error).startswith('RESTORED_') + write.assert_not_called() +with patch.object(m,'inspect',return_value=container),patch.object(m.Path,'exists',return_value=False),patch.object(m,'private_write'): + restored=m.finish(pathlib.Path('/offline'),copy.deepcopy(state)) +assert restored['proxyStopped'] and restored['currentFingerprint']==fingerprint and restored['originalImageId']==restored['currentImageId'] +with patch.object(m,'inspect',return_value=container),patch.object(m.Path,'exists',return_value=True),patch.object(m.Path,'read_bytes',return_value=b'python3\x00'+str(pathlib.Path('/offline')/'proxy.py').encode()+b'\x00'),patch.object(m.os,'kill'),patch.object(m.time,'sleep'),patch.object(m,'private_write') as write: + try: m.finish(pathlib.Path('/offline'),copy.deepcopy(state)); assert False + except RuntimeError as error: assert str(error)=='PROXY_NOT_STOPPED' + write.assert_not_called() +`], { encoding: 'utf8', cwd: new URL('..', import.meta.url), timeout: 10000 }) + assert.equal(result.status, 0, result.stderr || result.error?.message) +}) +test('only original scoped event is recovered and restoration precedes a successful proof', async () => { + const calls = [], executor = createTestFaultExecutor(environment, transport(calls)) + await executor.prepare(scope) + await executor.prepare({ ...scope, loadingPlanId: '100' }) + const proof = await executor.readProof({ containerBookingId: '66', loadingPlanId: '99', customerId: '7', marker: 'TEST-ROOT' }) + assert.equal(proof.restored, true) + await executor.cleanup() + assert.deepEqual(calls, ['prepare', 'proxy-ready', 'activate', 'health', 'status', 'recover', + 'status', 'proof', 'restore', 'health', 'finish']) +}) +test('failed activation retains task ownership for explicit restoration, never retries activation', async () => { + const calls = [], executor = createTestFaultExecutor(environment, transport(calls, true)) + await assert.rejects(executor.prepare(scope), /DOCKER_409/) + await executor.cleanup() + assert.equal(calls.filter(name => name === 'activate').length, 1) + assert.deepEqual(calls.slice(-3), ['restore', 'health', 'finish']) +}) diff --git a/scripts/test-event-fault-runtime.py b/scripts/test-event-fault-runtime.py new file mode 100644 index 000000000..6bde84707 --- /dev/null +++ b/scripts/test-event-fault-runtime.py @@ -0,0 +1,430 @@ +"""Internal TEST executor. Receives one bounded task action from Jenkins. + +Never edits business SQL. Docker configuration is captured before the scoped +fault and restored to the same image, mounts, ports and network identity. +Only the original DEAD event can be retried through its protected HTTP API. +""" +import base64 +import copy +import hashlib +import http.client +import json +import os +import re +import secrets +import signal +import socket +import subprocess +import sys +import tempfile +import time +import urllib.parse +import urllib.request +import uuid +from pathlib import Path + + +class DockerConnection(http.client.HTTPConnection): + def connect(self): + self.sock = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM) + self.sock.settimeout(self.timeout) + self.sock.connect('/var/run/docker.sock') + + +def docker(method, path, body=None): + conn = DockerConnection('localhost', timeout=20) + data = None if body is None else json.dumps(body).encode() + conn.request(method, path, data, {'Content-Type': 'application/json'}) + response = conn.getresponse() + raw = response.read() + conn.close() + if response.status >= 300: + raise RuntimeError('DOCKER_' + str(response.status)) + return json.loads(raw) if raw else None + + +def inspect(name): + return docker('GET', '/containers/' + urllib.parse.quote(name, safe='') + '/json') + + +def env_of(container): + return dict(item.split('=', 1) for item in container['Config']['Env']) + + +def require_test(container, name): + if container['Name'] != '/' + name or container['Config']['Labels'].get('com.docker.compose.project') != 'firstleg-test': + raise RuntimeError('NOT_TEST_CONTAINER') + + +def recreation(container, environment, full=False): + body = copy.deepcopy(container['Config']) + body['Env'] = [key + '=' + value for key, value in environment.items()] + body['HostConfig'] = copy.deepcopy(container['HostConfig']) + if body['HostConfig'].get('AutoRemove') or any(mount['Type'] not in ('bind', 'volume') for mount in container['Mounts']): + raise RuntimeError('UNSUPPORTED_CONTAINER_STORAGE') + # Preserve named volumes explicitly, never create a replacement anonymous volume. + if any(mount['Type'] == 'volume' and not mount.get('Name') for mount in container['Mounts']): + raise RuntimeError('UNNAMED_VOLUME') + if body['HostConfig'].get('VolumesFrom'): + raise RuntimeError('UNSUPPORTED_VOLUME_REFERENCE') + endpoints = {} + for name, network in container['NetworkSettings']['Networks'].items(): + aliases = [alias for alias in network.get('Aliases') or [] if alias != container['Id'][:12]] + endpoints[name] = {'Aliases': aliases, 'IPAMConfig': {'IPv4Address': network['IPAddress']}} + if full: + ipam = copy.deepcopy(network.get('IPAMConfig') or {}) + ipam['IPv4Address'] = network['IPAddress'] + if network.get('GlobalIPv6Address'): + ipam['IPv6Address'] = network['GlobalIPv6Address'] + endpoints[name]['IPAMConfig'] = ipam + for key in ('Links', 'DriverOpts', 'GwPriority', 'MacAddress'): + if network.get(key) is not None: + endpoints[name][key] = copy.deepcopy(network[key]) + body['NetworkingConfig'] = {'EndpointsConfig': endpoints} + return body + + +def replace(container, body): + name = container['Name'].lstrip('/') + current = inspect(name) + if current['Id'] != container['Id']: + raise RuntimeError('CONCURRENT_CONTAINER_CHANGE') + docker('POST', '/containers/' + current['Id'] + '/stop?t=10') + docker('DELETE', '/containers/' + current['Id'] + '?v=false') + created = docker('POST', '/containers/create?name=' + name, body) + docker('POST', '/containers/' + created['Id'] + '/start') + return created['Id'] + + +def private_write(path, value): + fd = os.open(path, os.O_WRONLY | os.O_CREAT | os.O_TRUNC, 0o600) + with os.fdopen(fd, 'w') as file: + json.dump(value, file) + + +def scope_directory(scope, direction='LOADING'): + if direction not in ('LOADING', 'UNLOAD'): + 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}: + 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 + return Path(tempfile.gettempdir()) / ('chongchong-event-' + hashlib.sha256(identity.encode()).hexdigest()[:24]) + + +def unload(state): + return state.get('direction', 'LOADING') == 'UNLOAD' + + +def producer_name(state): + return 'firstleg-test-lastleg' if unload(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') + + +def fingerprint(container): + body = recreation(container, env_of(container), full=True) + # Docker may add the legacy Config.MacAddress after explicit per-network MAC + # restoration. It is redundant only when it equals an actual preserved MAC; + # every network MAC remains in the fingerprint and unrelated values still fail. + network_macs = {endpoint.get('MacAddress') for endpoint in body['NetworkingConfig']['EndpointsConfig'].values()} + if body.get('MacAddress') and body['MacAddress'] in network_macs: + body.pop('MacAddress') + body['Env'].sort() + for endpoint in body['NetworkingConfig']['EndpointsConfig'].values(): + endpoint['Aliases'].sort() + body['ResolvedMounts'] = sorted(container['Mounts'], key=lambda mount: mount['Destination']) + body['NetworkIds'] = {name: network.get('NetworkID') for name, network in container['NetworkSettings']['Networks'].items()} + return hashlib.sha256(json.dumps(body, sort_keys=True, separators=(',', ':')).encode()).hexdigest() + + +def control(state, mode=None): + url = 'http://127.0.0.1:' + str(state['controlPort']) + ('/mode' if mode else '/status') + request = urllib.request.Request(url, None if mode is None else json.dumps({'mode': mode}).encode(), + {'X-Event-Fault-Control-Token': state['controlToken'], 'Content-Type': 'application/json'}) + with urllib.request.urlopen(request, timeout=5) as response: + return json.load(response) + + +def read_sql(container, sql): + # SQL is generated here from fixed templates and decimal/UUID/hash identities. + 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' + 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;' + result = subprocess.run(['mysql', '--host=' + host, '--port=' + str(url.port or 3306), + '--user=' + e[prefix + '_DB_USERNAME'], '--batch', '--raw', '--skip-column-names', url.path.lstrip('/')], + input=transaction, text=True, capture_output=True, timeout=20, + env=dict(os.environ, MYSQL_PWD=e[prefix + '_DB_PASSWORD'])) + if result.returncode: + code = re.search(r'ERROR (\d+)', result.stderr) + raise RuntimeError('MYSQL_' + (code.group(1) if code else 'UNKNOWN')) + return [json.loads(line) for line in result.stdout.splitlines() if line.strip()] + + +def producer(scope, direction='LOADING'): + if 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'] + "';") + else: + rows = read_sql(inspect('firstleg-test-business'), + "SELECT JSON_OBJECT('entryId',CAST(id AS CHAR),'eventId',JSON_UNQUOTE(JSON_EXTRACT(payload_json,'$.eventId')),'payloadHash',payload_hash,'eventType',event_type,'publishStatus',publish_status,'failedAttempts',retry_count) FROM fl_inbound_outbox WHERE tenant_id=" + scope['customerId'] + + " AND aggregate_id=" + scope['loadingPlanId'] + " AND event_type='FirstLegLoadingPlanCompleted';") + if len(rows) > 1: + raise RuntimeError('AMBIGUOUS_EVENT') + return rows[0] if rows else None + + +def container_facts(scope): + rows = read_sql(inspect('firstleg-test-business'), + "SELECT JSON_OBJECT('id',CAST(id AS CHAR),'containerNo',container_no,'version',CAST(version AS CHAR),'unloadingCompletedAt',CAST(unloading_completed_at AS CHAR),'unloadingPhotoUploadedAt',CAST(unloading_photo_uploaded_at AS CHAR),'estimatedEmptyReportedAt',CAST(estimated_empty_reported_at AS CHAR),'actualEmptyReportedAt',CAST(actual_empty_reported_at AS CHAR),'emptyReportSourceVersion',CAST(empty_report_source_version AS CHAR),'updatedAt',CAST(updated_at AS CHAR),'updatedBy',CAST(updated_by AS CHAR),'updatedByName',updated_by_name) FROM fl_container_booking WHERE id=" + scope['containerBookingId'] + ";") + if len(rows) != 1: + raise RuntimeError('CONTAINER_SCOPE_MISMATCH') + return rows[0] + + +def business_ids(scope): + rows = read_sql(inspect('firstleg-test-lastleg'), "SELECT JSON_OBJECT('id',CAST(id AS CHAR)) FROM ll_order WHERE customer_id='" + + scope['customerId'] + "' AND source_container_booking_id='" + scope['containerBookingId'] + "' ORDER BY id;") + return [row['id'] for row in rows] + + +def prepare(scope, directory, config): + 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 = 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)) + 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)] + for sock in sockets: + sock.bind(('127.0.0.1', 0)) + ports = [sock.getsockname()[1] for sock in sockets] + for sock in sockets: + sock.close() + proxy = directory / 'proxy.py' + 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']] + 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')], + env=dict(os.environ, EVENT_FAULT_CONTROL_TOKEN=token), stdout=log, stderr=log, start_new_session=True) + state = {'scope': scope, 'originalBody': original_body, 'originalId': original['Id'], 'image': original['Config']['Image'], + 'direction': config.get('direction', 'LOADING'), 'originalImageId': original['Image'], 'originalFingerprint': fingerprint(original), + 'proxyPid': process.pid, 'proxyPort': ports[0], 'controlPort': ports[1], 'controlToken': token, + 'operationsRequestId': str(uuid.uuid4()), 'prepared': False, 'restored': False} + private_write(directory / 'state.json', state) + return {'created': True, 'originalImageId': state['originalImageId'], 'originalFingerprint': state['originalFingerprint']} + + +def activate(scope, directory, state): + snapshot = control(state, 'FAULT') + if snapshot.get('eventId'): + raise RuntimeError('EVENT_ALREADY_PINNED') + original = inspect(producer_name(state)) + require_test(original, producer_name(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']): + raise RuntimeError('CONCURRENT_PRODUCER_CHANGE') + body = copy.deepcopy(state['originalBody']) + e = env_of(original) + e[setting] = 'http://host.docker.internal:' + str(state['proxyPort']) + e.update(EVENT_MAX_ATTEMPTS='12', EVENT_BASE_DELAY_SECONDS='5', EVENT_MAX_DELAY_SECONDS='5', EVENT_JITTER='0') + if len(e.get('EVENT_OPERATIONS_TOKEN', '')) < 32: + e['EVENT_OPERATIONS_TOKEN'] = secrets.token_urlsafe(32) + body['Env'] = [key + '=' + value for key, value in e.items()] + state['prepared'] = True + private_write(directory / 'state.json', state) + replace(original, body) + return {'activated': True} + + +def recover(scope, directory, state): + row = producer(scope, state.get('direction', 'LOADING')) + if not row or row['publishStatus'] != 'DEAD' or row['failedAttempts'] != 12: + raise RuntimeError('ORIGINAL_EVENT_NOT_DEAD') + 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)})} + private_write(directory / 'state.json', state) + control(state, 'PASS') + container = inspect(producer_name(state)) + e = env_of(container) + 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/' + 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'}) + with urllib.request.urlopen(request, timeout=15) as response: + result = json.load(response) + if result.get('code') != '200': + raise RuntimeError('OPERATIONS_RETRY_REJECTED') + return {'retrySubmitted': True} + + +def restore(directory, state): + if state['restored']: + return {'restored': True} + try: + control(state, 'PASS') + except Exception: + if not unload(state): + raise + # A dead proxy must not prevent restoring the producer's direct target. + if state['prepared']: + try: + current = inspect(producer_name(state)) + except RuntimeError as error: + if str(error) != 'DOCKER_404': + raise + created = docker('POST', '/containers/create?name=' + producer_name(state), state['originalBody']) + docker('POST', '/containers/' + created['Id'] + '/start') + 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'): + docker('POST', '/containers/' + current['Id'] + '/start') + state['prepared'] = False + private_write(directory / 'state.json', state) + return {'restoreStarted': True} + if current['Config']['Image'] != state['image'] or env_of(current).get(target_setting(state)[0]) != 'http://host.docker.internal:' + str(state['proxyPort']): + raise RuntimeError('CONCURRENT_PRODUCER_CHANGE') + replace(current, state['originalBody']) + state['prepared'] = False + private_write(directory / 'state.json', state) + return {'restoreStarted': True} + + +def proof(scope, state): + row = producer(scope, state.get('direction', 'LOADING')) + if not row or row['publishStatus'] != 'PUBLISHED' or not state.get('beforeRetry'): + raise RuntimeError('RECOVERY_NOT_PUBLISHED') + 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' + 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): + 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: + consumers = read_sql(inspect('firstleg-test-lastleg'), + "SELECT JSON_OBJECT('eventId',event_id,'payloadHash',payload_hash,'status',status) FROM ll_loading_event_inbox WHERE event_id='" + + event_id + "' AND customer_id='" + scope['customerId'] + "';") + 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)})}, + '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 + + +def finish(directory, state): + if state['restored']: + return state.get('restoration', {'restored': True}) + current = inspect(producer_name(state)) + require_test(current, producer_name(state)) + if current['State'].get('Health', {}).get('Status') != 'healthy' or current['Config']['Image'] != state['image']: + 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']): + 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): + for _ in range(50): + if not cmdline.exists() or not cmdline.read_bytes(): + break + time.sleep(0.1) + else: + raise RuntimeError('PROXY_NOT_STOPPED') + result = {'restored': True} + if unload(state): + result.update(healthy=True, proxyStopped=True, producerName=producer_name(state), + originalImageId=state['originalImageId'], currentImageId=current['Image'], + originalFingerprint=state['originalFingerprint'], currentFingerprint=fingerprint(current)) + state['restoration'] = result + state['restored'] = True + # Proof is already captured. Remove credentials from the retained task metadata. + state.pop('originalBody', None) + state.pop('controlToken', None) + private_write(directory / 'state.json', state) + return result + + +def main(config): + scope, action = config['scope'], config['action'] + direction = config.get('direction', 'LOADING') + directory = scope_directory(scope, direction) + if action == 'prepare': + return prepare(scope, directory, config) + state = json.loads((directory / 'state.json').read_text()) + if state['scope'] != scope or state.get('direction', 'LOADING') != direction: + raise RuntimeError('TASK_SCOPE_MISMATCH') + if action == 'activate': + return activate(scope, directory, state) + if action == 'health': + current = inspect(producer_name(state)) + return {'healthy': current['State'].get('Health', {}).get('Status') == 'healthy'} + if action == 'proxy-ready': + return {'ready': control(state)['mode'] == 'PASS'} + if action == 'status': + return {'producer': producer(scope, direction), 'proxy': control(state)} + if action == 'recover': + return recover(scope, directory, state) + if action == 'proof': + return proof(scope, state) + if action == 'restore': + return restore(directory, state) + if action == 'finish': + return finish(directory, state) + raise RuntimeError('INVALID_ACTION') + + +if __name__ == '__main__': + try: + value = main(json.loads(base64.b64decode(sys.argv[1]))) + print('TEST_EVENT_JSON=' + json.dumps({'ok': True, 'data': value})) + except Exception as error: + message = str(error) + safe = message if re.fullmatch(r'[A-Z_0-9]{1,80}', message) else 'TEST_EVENT_ACTION_FAILED' + print('TEST_EVENT_JSON=' + json.dumps({'ok': False, 'error': safe}))