From 49687457644c1016de02a913bfb8b8d7fa80fd31 Mon Sep 17 00:00:00 2001 From: ouyangxq <775153912@qq.com> Date: Sat, 3 Oct 2026 09:02:25 +0800 Subject: [PATCH] test: verify fee void ACK loss with original event recovery --- scripts/README.md | 6 + scripts/fee-void-recovery-test-env.mjs | 170 ++++++++++++++++++ scripts/fee-void-recovery-test-env.test.mjs | 72 ++++++++ scripts/test-event-fault-executor.mjs | 26 ++- scripts/test-event-fault-runtime.py | 150 ++++++++++++++-- scripts/test-fee-void-fault-proof.mjs | 91 ++++++++++ scripts/test-fee-void-fault-proof.test.mjs | 189 ++++++++++++++++++++ 7 files changed, 688 insertions(+), 16 deletions(-) create mode 100644 scripts/fee-void-recovery-test-env.mjs create mode 100644 scripts/fee-void-recovery-test-env.test.mjs create mode 100644 scripts/test-fee-void-fault-proof.mjs create mode 100644 scripts/test-fee-void-fault-proof.test.mjs diff --git a/scripts/README.md b/scripts/README.md index d2e1cd427..5c7a20302 100644 --- a/scripts/README.md +++ b/scripts/README.md @@ -1,5 +1,11 @@ # 后端脚本管理与全量 HTTP 验收索引 +`fee-void-recovery-test-env.mjs` 是费用作废确认响应丢失的 TEST 专项入口，默认 `--plan` 零网络。执行须显式传入 `--execute --environment TEST --run-id <新批次> --customer-id <测试客户ID> --control-customer-id <另一测试客户ID> --warehouse-id <海外仓ID>`，外部 TEST/Jenkins 配置及 `FBA_CORE_JENKINS_AGENT` 提供凭据与取证节点。创建一条新仓储服务应收，经提交审核和原发布确认后，对该来源版本的 VOIDED 事件注入 ACK 丢失；原事件 DEAD 后受保护重试，核对最终作废、同命令重放、另一客户费用与完整库存不变。不写业务SQL，不使用旧费用作废；故障窗口会临时重建同镜像 Lastleg，禁止并行发布或其他共享环境写入。报告在 `reports/fee-void-<新批次>/`，请求前保存意图，失败保留样本，不自动重跑或覆盖；finally 恢复原服务、停止代理并退出会话，恢复未证明则失败。`fee-void-recovery-test-env.test.mjs` 是离线编排与失败清理检查，不能代替真实环境结果。 + +`test-fee-void-fault-proof.mjs` 是上述流程的内部只读证据校验器，无独立 CLI；配合 `test-event-fault-executor.mjs` 与 `test-event-fault-runtime.py` 检查原费用发布确认、精确作废事件、Finance 消费及 Lastleg 确认状态。实际运行结论只以该批次证据为准。 + +费用作废代理使用 `event-delivery-fault-proxy.py --fee-source-id <来源ID> --source-version <版本>`，两参数必须成对且不能与其他选择器混用，事件类型固定 `BusinessFeeSourceChanged`，仅匹配正式 action `VOIDED`。指向正式费用消费路径时，同源 `/internal/finance/v1/wallet-command-events` 保留原请求与响应直通，不进入费用故障选择；其他未配置路径仍拒绝。此边界由代理离线测试覆盖。 + `test-delivery-fault-proof.mjs` 是签收回写故障的内部证据校验器，无CLI、网络或写入。按精确客户/派送/来源入仓/来源订单/根订单核对原事件及载荷摘要、13次投递与代理记录、唯一收件箱和人工重试审计，并比较重试前后的Firstleg与Order事实。首片签收仍有有效余量时允许根单继续等待；不把根单必须立即COMPLETED当成成功条件。 F40 专项可用 `--signature-mode ROOT_FIRST|CHILD_FIRST|ROOT_SIGNED_REMAINDER_CANCELLED` 只运行显式指定的一种顺序，省略仍串行执行三种；报告和成功范围仅覆盖所列模式。`--verify-delivery-event-recovery` 必须同时显式指定一种顺序，且环境为TEST；仅在该模式第一张真实FBA派送签收前开启定向ACK丢失，不能使用普通海外转运的签收替代国内回写。故障窗口会重建同镜像Lastleg并临时调整事件目标和退避配置，禁止与其他部署或共享环境写入并行；原事件DEAD恢复后必须验证原镜像/配置/网络身份、健康及代理退出。入口事先留存精确故障scope，finally兜底恢复，缺恢复证据或恢复校验失败一律失败。完整库存取证接入既有只读器，报告保留于该批次 `evidence/`，不能把某一模式或事件通过解读为全部反向事件完成。 diff --git a/scripts/fee-void-recovery-test-env.mjs b/scripts/fee-void-recovery-test-env.mjs new file mode 100644 index 000000000..3271ae95b --- /dev/null +++ b/scripts/fee-void-recovery-test-env.mjs @@ -0,0 +1,170 @@ +import { mkdirSync, writeFileSync, openSync, closeSync, unlinkSync, existsSync } from 'node:fs' +import { dirname, resolve, join } from 'node:path' +import { fileURLToPath, pathToFileURL } from 'node:url' +import { isDeepStrictEqual } from 'node:util' +import { createHash } from 'node:crypto' +import { AcceptanceRun, parseArguments, validateExecutionOptions } from './admin-test-env-acceptance.mjs' +import { createTestEvidenceReaders } from './fba-core-regression-evidence.mjs' +import { redactEvidence } from './fba-core-regression-test-env.mjs' + +const ROOT = resolve(dirname(fileURLToPath(import.meta.url)), '..') +const FEES = '/api/platform/v1/last-leg/finance-fee-sources' +const TYPE = 'OVERSEAS_WAREHOUSE_FEE' +const id = value => String(value ?? '') +const check = (value, message) => { if (!value) throw Error(message) } +export function parseFeeVoidArguments(args) { + const config = { execute: false, environment: 'TEST', runId: 'plan' }, seen = new Set() + for (let i = 0; i < args.length; i++) { + const arg = args[i] + if (['--plan', '--execute'].includes(arg)) { + check(!seen.has('action'), 'Choose one execution action'); seen.add('action'); config.execute = arg === '--execute'; continue + } + const field = { '--environment': 'environment', '--run-id': 'runId', '--customer-id': 'customerId', '--control-customer-id': 'controlCustomerId', '--warehouse-id': 'warehouseId' }[arg] + check(field && !seen.has(arg) && args[i + 1] && !args[i + 1].startsWith('--'), 'Invalid or duplicate argument') + seen.add(arg); config[field] = args[++i] + } + validateConfig(config) + if (config.execute) check(seen.has('--environment'), 'Execution requires explicit TEST') + return config +} +function validateConfig(config) { + check(config.environment === 'TEST' && /^[A-Za-z0-9][A-Za-z0-9_-]{0,47}$/.test(config.runId), 'TEST and safe run-id required') + if (config.execute) check(config.runId !== 'plan' && ['customerId', 'controlCustomerId', 'warehouseId'].every(key => typeof config[key] === 'string' && /^[1-9][0-9]{0,19}$/.test(config[key])) + && config.customerId !== config.controlCustomerId, 'Two explicit distinct customers and warehouse required') +} +export function feeVoidPlan(config) { + return { environment: 'TEST', execute: config.execute, runId: config.runId, customerId: config.customerId, controlCustomerId: config.controlCustomerId, + warehouseId: config.warehouseId, output: `reports/fee-void-${config.runId}`, sourceType: TYPE, amount: 17.31, currency: 'CNY', + stages: ['read customer/warehouse and isolation baselines', 'create/submit/approve new warehouse service fee', 'wait original publication', + 'inject only VOIDED ACK loss', 'original DEAD retry and full runtime restore', 'same-command replay and isolation readback'], + limits: 'Only a new unbilled TEST receivable. Retain fixtures and failures, never automatically restart. No business SQL writes. Exclusive Lastleg fault window; no overlapping deployments or writers.' } +} +export async function runFeeVoid(config, { env = process.env, outputRoot = ROOT, readers, faultFactory, + contextFactory = options => new AcceptanceRun(options), wait = ms => new Promise(resolve => setTimeout(resolve, ms)) } = {}) { + validateConfig(config) + if (!config.execute) return feeVoidPlan(config) + check(env.CHONGCHONG_ENV === 'test' && env.CHONGCHONG_ADMIN_USERNAME && env.CHONGCHONG_ADMIN_PASSWORD, 'Explicit TEST credentials required') + const base = new URL(env.CHONGCHONG_ADMIN_API_BASE_URL) + check(['http:', 'https:'].includes(base.protocol) && ['/api', '/api/'].includes(base.pathname) && !base.username && !base.password && !base.search && !base.hash, 'Invalid TEST gateway') + readers ||= createTestEvidenceReaders(env) + faultFactory ||= (await import('./test-event-fault-executor.mjs')).createTestFeeVoidFaultExecutor + check(typeof faultFactory === 'function' && typeof readers.readFleetInventory === 'function', 'Fee fault executor and complete inventory reader required') + const reports = resolve(outputRoot, 'reports'), output = join(reports, `fee-void-${config.runId}`) + mkdirSync(reports, { recursive: true }) + for (const name of ['.fba-core-regression.lock', '.fleet-acceptance.lock']) check(!existsSync(join(reports, name)), 'Another shared-environment acceptance is running or needs review') + const lockPath = join(reports, '.test-event-recovery.lock'), lock = openSync(lockPath, 'wx') + const report = { success: false, stage: 'PREFLIGHT', requests: [], inventory: [], goalComplete: false } + let context, fault, initialized = false, refreshToken, attempted = false, prepared = false + const secrets = [env.CHONGCHONG_ADMIN_PASSWORD, env.JENKINS_API_TOKEN] + const save = () => { if (initialized) writeFileSync(join(output, 'report.json'), JSON.stringify(redactEvidence(report, secrets), null, 2)) } + try { + writeFileSync(lock, JSON.stringify({ pid: process.pid, runId: config.runId })) + mkdirSync(output); initialized = true + writeFileSync(join(output, 'manifest.json'), JSON.stringify(feeVoidPlan(config), null, 2)) + await readers.preflight() + context = contextFactory(validateExecutionOptions({ ...parseArguments(['--modules', 'merchant-qualification-fixture', '--quiet', 'true']), + environment: 'test', gateway: base.origin, username: env.CHONGCHONG_ADMIN_USERNAME, password: env.CHONGCHONG_ADMIN_PASSWORD, + markerNamespace: createHash('sha256').update(config.runId).digest('hex').slice(0, 12) })) + report.marker = context.marker + const request = async (method, path, body, headers = {}) => { + report.pending = { method, path, body }; save() + const result = await context.request(method, path, body, context.token, headers) + report.requests.push({ method, path, http: result.http, code: result.body?.code, requestId: result.requestId, traceId: result.traceId }) + delete report.pending; save() + check(result.http === 200 && id(result.body?.code) === '200', `Fee recovery request failed: ${method} ${path} (${result.http}/${result.body?.code})`) + return result.body.data + } + const login = await request('POST', '/api/auth/login', { loginId: env.CHONGCHONG_ADMIN_USERNAME, password: env.CHONGCHONG_ADMIN_PASSWORD }) + check(login?.accessToken && login.refreshToken, 'Complete administrator session required') + context.token = login.accessToken; refreshToken = login.refreshToken; secrets.push(context.token, refreshToken) + const customers = [] + for (const customerId of [config.customerId, config.controlCustomerId]) { + const customer = await request('GET', `/api/platform/v1/iam/tenants/${customerId}`) + check(id(customer.id) === customerId && customer.status === 'ACTIVE' && customer.tenantName, 'Exact active customer required') + customers.push(customer) + } + const warehouse = await request('GET', `/api/platform/v1/warehouse-admin/${config.warehouseId}?tenantId=0`) + check(id(warehouse.warehouse?.id) === config.warehouseId && id(warehouse.warehouse.tenantId) === '0' && warehouse.warehouse.status === 'ACTIVE' + && warehouse.warehouse.warehouseType === 'OVERSEAS', 'Exact active overseas warehouse required') + const feeList = customerId => request('GET', `${FEES}?${new URLSearchParams({ customerId })}`) + const pool = customerId => request('GET', `/api/platform/finance/v1/fee-pool?${new URLSearchParams({ customerId, direction: 'RECEIVABLE' })}`) + const snapshot = async () => { + const value = [] + for (const customerId of [config.customerId, config.controlCustomerId]) { + const sources = await feeList(customerId), finance = await pool(customerId) + check(Array.isArray(sources) && sources.every(row => id(row.customerId) === customerId) && Array.isArray(finance) && finance.every(row => id(row.customerId) === customerId), 'Customer fee scope mismatch') + const inventory = await readers.readFleetInventory({ marker: context.marker, customerId, warehouseId: config.warehouseId }) + value.push({ customerId, sources, finance, inventory }) + } + return value + } + report.before = await snapshot() + report.stage = 'CREATE_NEW_FEE' + const body = { customerId: config.customerId, customerName: customers[0].tenantName, sourceType: TYPE, businessType: 'WAREHOUSE_SERVICE', + businessId: null, businessNo: `${context.marker}-VOID-TEST`, warehouseId: config.warehouseId, + feeItemCode: 'TEST-VOID-ACK', feeItemName: '受控测试作废响应丢失费用', direction: 'RECEIVABLE', businessDate: new Date().toISOString().slice(0, 10), + currency: 'CNY', amount: 17.31, paymentVoucherFileIds: [] } + report.createBody = body + let fee = await request('POST', FEES, body, { 'Idempotency-Key': `${context.marker}-CREATE` }) + report.sourceFeeId = id(fee.sourceFeeId) + check(/^[1-9][0-9]+$/.test(report.sourceFeeId) && fee.reviewStatus === 'DRAFT' && id(fee.customerId) === config.customerId && Number(fee.sourceVersion) === 1, 'New isolated fee draft required') + const path = `${FEES}/${report.sourceFeeId}` + fee = await request('POST', `${path}/submit`, { idempotencyKey: `${context.marker}-SUBMIT` }) + check(fee.reviewStatus === 'PENDING', 'New fee must await review') + fee = await request('POST', `${path}/review`, { decision: 'APPROVE', reason: context.marker, idempotencyKey: `${context.marker}-REVIEW` }) + check(fee.reviewStatus === 'APPROVED' && fee.publicationStatus === 'PUBLISHED' && fee.sourceStatus === 'ACTIVE', 'Approved active source required') + report.approved = fee + let published = [] + for (let i = 0; i < 30; i++) { + published = (await pool(config.customerId)).filter(row => id(row.sourceId) === report.sourceFeeId && Number(row.sourceVersion) === 1) + if (published.length) break + await wait(1000) + } + check(published.length === 1 && Number(published[0].amount) === 17.31 && published[0].currency === 'CNY', 'Original approved fee must exist exactly once in Finance') + report.originalFinance = published + report.scope = { marker: context.marker, customerId: config.customerId, sourceFeeId: report.sourceFeeId, sourceVersion: '1', sourceType: TYPE } + fault = faultFactory(env) + report.stage = 'PREPARE_INTENT'; attempted = true; save() + await fault.prepare(report.scope) + prepared = true + report.stage = 'VOID_REQUEST' + report.voidRequest = { expectedVersion: 1, reason: '受控TEST作废确认响应丢失验收', idempotencyKey: `${context.marker}-VOID` } + const pending = await request('POST', `${path}/void`, report.voidRequest) + check(pending.sourceStatus === 'ACTIVE' && pending.voidStatus === 'PENDING_FINANCE', 'Sending void request must not confirm business-side cancellation') + report.pendingSource = pending; report.stage = 'WAIT_DEAD_AND_RECOVER'; save() + report.eventProof = await fault.readProof(report.scope) + report.stage = 'RESTORED_READBACK'; save() + const confirmed = (await feeList(config.customerId)).filter(row => id(row.sourceFeeId) === report.sourceFeeId) + check(confirmed.length === 1 && confirmed[0].sourceStatus === 'VOIDED' && confirmed[0].voidStatus === 'CONFIRMED' && Number(confirmed[0].sourceVersion) === 1, 'Matching Finance ACK must confirm only the original source version') + report.confirmedSource = confirmed[0] + const replay = await request('POST', `${path}/void`, report.voidRequest) + check(isDeepStrictEqual(replay, confirmed[0]), 'Same void command replay changed the confirmed source') + report.after = await snapshot() + check(isDeepStrictEqual(report.before[1], report.after[1]), 'Other customer fees or inventory changed') + check(isDeepStrictEqual(report.before[0].inventory, report.after[0].inventory), 'Fee cancellation changed inventory') + check(isDeepStrictEqual(report.before[0].sources, report.after[0].sources.filter(row => id(row.sourceFeeId) !== report.sourceFeeId)), 'Unrelated original source changed') + check(isDeepStrictEqual(report.before[0].finance, report.after[0].finance), 'Voided fee remains in pool or unrelated finance row changed') + report.success = true; report.stage = 'VERIFIED' + } catch (error) { report.error = error.message; report.success = false; report.stage = 'FAILED' } + finally { + if (fault) try { + report.restoration = await fault.cleanup() + const noMutation = !prepared && report.restoration?.attempted === false + if (attempted && !noMutation && !(report.restoration?.restored && report.restoration?.healthy && report.restoration?.proxyStopped)) throw Error('Fee void restoration not verified') + } catch (error) { report.success = false; report.stage = 'RESTORATION_FAILED'; report.restorationError = error.message } + if (context?.token) try { + const result = await context.request('POST', '/api/auth/logout', { refreshToken }, context.token) + report.sessionClosed = result.http === 200 && id(result.body?.code) === '200' + if (!report.sessionClosed) report.success = false + } catch (error) { report.success = false; report.sessionClosed = false; report.logoutError = error.message } + try { save() } finally { closeSync(lock); unlinkSync(lockPath) } + } + return report +} +if (process.argv[1] && import.meta.url === pathToFileURL(resolve(process.argv[1])).href) { + try { + const config = parseFeeVoidArguments(process.argv.slice(2)), result = await runFeeVoid(config) + console.log(JSON.stringify(config.execute ? { success: result.success, stage: result.stage, error: result.error, sourceFeeId: result.sourceFeeId, restored: result.restoration?.restored, output: feeVoidPlan(config).output } : result, null, 2)) + if (config.execute && !result.success) process.exitCode = 1 + } catch (error) { console.error(error.message); process.exitCode = 1 } +} diff --git a/scripts/fee-void-recovery-test-env.test.mjs b/scripts/fee-void-recovery-test-env.test.mjs new file mode 100644 index 000000000..4ebe8b729 --- /dev/null +++ b/scripts/fee-void-recovery-test-env.test.mjs @@ -0,0 +1,72 @@ +import test from 'node:test' +import assert from 'node:assert/strict' +import { mkdtempSync, existsSync, readFileSync, rmSync } from 'node:fs' +import { tmpdir } from 'node:os' +import { join, dirname, basename, resolve } from 'node:path' +import { parseFeeVoidArguments, runFeeVoid } from './fee-void-recovery-test-env.mjs' + +const config = () => parseFeeVoidArguments(['--execute', '--environment', 'TEST', '--run-id', 'offline', + '--customer-id', '41', '--control-customer-id', '42', '--warehouse-id', '900']) +const env = { CHONGCHONG_ENV: 'test', CHONGCHONG_ADMIN_USERNAME: 'offline-admin', + CHONGCHONG_ADMIN_PASSWORD: 'private-password', CHONGCHONG_ADMIN_API_BASE_URL: 'https://test.example.invalid/api' } +function harness(outputRoot, failure) { + const calls = []; let fee, finished = false, cleanup = 0 + const context = { marker: 'FEE-OFFLINE', async request(method, path, body) { + calls.push({ method, path }) + let data + if (path === '/api/auth/login') data = { accessToken: 'private-access', refreshToken: 'private-refresh' } + else if (path === '/api/auth/logout') { if (failure === 'logout') throw Error('LOGOUT_TRANSPORT'); data = {} } + else if (path.includes('/iam/tenants/')) data = { id: path.split('/').at(-1), tenantName: 'Offline customer', status: 'ACTIVE' } + else if (path.includes('/warehouse-admin/')) data = { warehouse: { id: '900', tenantId: '0', status: 'ACTIVE', warehouseType: 'OVERSEAS' } } + else if (path.includes('/fee-pool?')) data = path.includes('customerId=41') && fee?.reviewStatus === 'APPROVED' && (!finished || failure === 'pool') + ? [{ customerId: '41', sourceId: '81', sourceVersion: 1, amount: 17.31, currency: 'CNY' }] : [] + else if (method === 'GET') data = path.includes('customerId=41') && fee ? [structuredClone(fee)] + : finished && failure === 'isolation' ? [{ customerId: '42', sourceFeeId: '99' }] : [] + else if (path.endsWith('/submit')) { fee.reviewStatus = 'PENDING'; data = fee } + else if (path.endsWith('/review')) { Object.assign(fee, { reviewStatus: 'APPROVED', publicationStatus: 'PUBLISHED', sourceStatus: 'ACTIVE' }); data = fee } + else if (path.endsWith('/void')) { + fee.voidStatus = finished ? 'CONFIRMED' : 'PENDING_FINANCE' + if (failure === 'early-confirm') fee.sourceStatus = 'VOIDED' + data = fee + } else { fee = { sourceFeeId: '81', sourceVersion: 1, customerId: body.customerId, reviewStatus: 'DRAFT' }; data = fee } + return { http: 200, body: { code: '200', data: structuredClone(data) } } + } } + return { calls, get cleanup() { return cleanup }, options: { env, outputRoot, contextFactory: () => context, + wait: async () => {}, readers: { preflight: async () => {}, readFleetInventory: async scope => ({ scope, balances: [], positions: [], ledger: [] }) }, + faultFactory: () => ({ prepare: async () => { if (['prepare', 'preflight'].includes(failure)) throw Error('PREPARE_FAILED') }, + readProof: async () => { finished = true; Object.assign(fee, { sourceStatus: 'VOIDED', voidStatus: 'CONFIRMED' }); return { status: 'VERIFIED' } }, + cleanup: async () => { cleanup++; return failure === 'preflight' || failure === 'missing-restoration' + ? { attempted: false } : { restored: failure !== 'restore', healthy: true, proxyStopped: true } } }) } } +} +test('fee void plan has no network and refuses implicit environment or ambiguous customer scope', async () => { + assert.equal((await runFeeVoid(parseFeeVoidArguments([]), { env: {} })).execute, false) + for (const args of [['--execute'], ['--environment', 'PROD'], ['--run-id', '../escape'], ['--plan', '--execute']]) + assert.throws(() => parseFeeVoidArguments(args)) + await assert.rejects(runFeeVoid({ ...config(), controlCustomerId: '41' }, { env }), /distinct/) + await assert.rejects(runFeeVoid(config(), { env: { ...env, CHONGCHONG_ENV: 'prod' } }), /TEST/) +}) +for (const failure of [undefined, 'prepare', 'preflight', 'early-confirm', 'isolation', 'pool', 'restore', 'missing-restoration', 'logout']) { + test(`fee void ${failure || 'success'} retains evidence and releases its lock after cleanup`, async () => { + const root = mkdtempSync(join(tmpdir(), 'fee-void-contract-')) + try { + const h = harness(root, failure), report = await runFeeVoid(config(), h.options) + assert.equal(report.success, !failure, JSON.stringify(report)) + assert.equal(h.cleanup, 1) + if (failure === 'preflight') assert.equal(report.stage, 'FAILED') + if (failure === 'missing-restoration') assert.equal(report.stage, 'RESTORATION_FAILED') + assert.equal(h.calls.filter(c => c.path === '/api/auth/logout').length, 1) + assert.equal(existsSync(join(root, 'reports/.test-event-recovery.lock')), false) + const path = join(root, 'reports/fee-void-offline/report.json'), saved = readFileSync(path, 'utf8') + assert.doesNotMatch(saved, /private-password|private-access|private-refresh/) + if (failure === 'logout') assert.equal(JSON.parse(saved).logoutError, 'LOGOUT_TRANSPORT') + const previous = h.calls.length + assert.equal((await runFeeVoid(config(), h.options)).success, false) + assert.equal(h.calls.length, previous) + assert.equal(readFileSync(path, 'utf8'), saved) + } finally { + assert.equal(dirname(resolve(root)), resolve(tmpdir())) + assert.ok(basename(root).startsWith('fee-void-contract-')) + rmSync(root, { recursive: true, force: true }) + } + }) +} diff --git a/scripts/test-event-fault-executor.mjs b/scripts/test-event-fault-executor.mjs index 2f587be62..c3fbab69e 100644 --- a/scripts/test-event-fault-executor.mjs +++ b/scripts/test-event-fault-executor.mjs @@ -4,6 +4,8 @@ import { readFileSync } from 'node:fs' import { createHash } from 'node:crypto' import { isDeepStrictEqual } from 'node:util' import { assertTestDeliveryFaultProof, deliveryFaultScopeKeys } from './test-delivery-fault-proof.mjs' +import { assertTestFeeVoidFaultProof, feeVoidScopeKeys, feeSourceTypes } from './test-fee-void-fault-proof.mjs' +export { assertTestFeeVoidFaultProof } from './test-fee-void-fault-proof.mjs' export { assertTestDeliveryFaultProof } from './test-delivery-fault-proof.mjs' import { validateEvidenceConfig } from './fba-core-regression-evidence.mjs' @@ -19,10 +21,12 @@ export function renderTestUnloadFaultAction(action, scope, agent) { return renderAction(action,scope,agent,'UNLOAD') } function renderAction(action,scope,agent,direction) { - const keys=direction==='DELIVERY'?deliveryFaultScopeKeys:direction==='UNLOAD'?['marker','customerId','containerBookingId']:['marker','customerId','loadingPlanId','containerBookingId'] - if (!['prepare', 'activate', 'proxy-ready', 'health', 'status', 'recover', 'proof', 'restore', 'finish'].includes(action) + const keys=direction==='FEE_VOID'?feeVoidScopeKeys:direction==='DELIVERY'?deliveryFaultScopeKeys:direction==='UNLOAD'?['marker','customerId','containerBookingId']:['marker','customerId','loadingPlanId','containerBookingId'] + if (!(action==='preflight'&&direction==='FEE_VOID')&&!['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] || '')) + || keys.filter(key=>!['marker','sourceType'].includes(key)).some(key => !/^[1-9][0-9]{0,19}$/.test(scope?.[key] || '')) + || (direction==='FEE_VOID'&&(!feeSourceTypes.includes(scope?.sourceType)||Number(scope?.sourceVersion)>2147483647)) + || (direction==='FEE_VOID'&&keys.some(key=>typeof scope?.[key]!=='string')) || (direction!=='LOADING'&&Object.keys(scope).some(key=>!keys.includes(key)))) { throw Error('Invalid TEST fault action or scope') } @@ -37,6 +41,9 @@ export function createTestFaultExecutor(env, fetcher = globalThis.fetch) { export function createTestDeliveryFaultExecutor(env, fetcher = globalThis.fetch) { return createExecutor(env,fetcher,'DELIVERY') } +export function createTestFeeVoidFaultExecutor(env, fetcher = globalThis.fetch) { + return createExecutor(env,fetcher,'FEE_VOID') +} export function createTestUnloadFaultExecutor(env, fetcher = globalThis.fetch) { return createExecutor(env,fetcher,'UNLOAD') } @@ -93,7 +100,7 @@ export function assertTestUnloadFaultProof(proof,scope) { 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 + let scope = null, finished = false, preparedEvidence=null, restoration=null, preparationAttempted=false, ready=false 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(/\/?$/, '/')), { @@ -118,6 +125,7 @@ function createExecutor(env,fetcher,direction) { throw Error('TEST fault observation timed out: ' + name) } async function cleanup() { + if(direction==='FEE_VOID'&&!preparationAttempted)return {attempted:false} if (!scope || finished) return restoration await action('restore') await wait('health', row => row.healthy) @@ -127,7 +135,7 @@ function createExecutor(env,fetcher,direction) { &&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((direction==='DELIVERY'?'Delivery':'Unload')+' restoration evidence mismatch') + &&restoration.currentFingerprint===preparedEvidence?.originalFingerprint))throw Error((direction==='FEE_VOID'?'Fee void':direction==='DELIVERY'?'Delivery':'Unload')+' restoration evidence mismatch') finished = true return restoration } @@ -135,15 +143,19 @@ function createExecutor(env,fetcher,direction) { async prepare(next) { if (scope) { if(direction!=='LOADING'&&!isDeepStrictEqual(scope,next))throw Error('TEST unload fault scope drift') + if(direction==='FEE_VOID'&&!ready)throw Error('Fee void preparation did not complete') return // Pin only the first actual current-run plan; other events pass through. } if(direction!=='LOADING')renderAction('prepare',next,env.FBA_CORE_JENKINS_AGENT,direction) scope = direction!=='LOADING'?{...next}:{ marker: next.marker, customerId: next.customerId, loadingPlanId: next.loadingPlanId, containerBookingId: next.containerBookingId } + if(direction==='FEE_VOID')await wait('preflight', row=>row.ready===true,60) + preparationAttempted=true preparedEvidence=await action('prepare') await wait('proxy-ready', row => row.ready, 10, true) await action('activate') await wait('health', row => row.healthy) + ready=true }, async readProof(expected) { if (!scope || (direction!=='LOADING'?!isDeepStrictEqual(scope,expected):Object.keys(scope).some(key => expected?.[key] !== scope[key]))) throw Error('TEST fault scope drift') @@ -154,6 +166,10 @@ function createExecutor(env,fetcher,direction) { const proof = await action('proof') if(direction==='UNLOAD')assertTestUnloadFaultProof(proof,scope) if(direction==='DELIVERY')assertTestDeliveryFaultProof(proof,scope) + if(direction==='FEE_VOID') { + if(!isDeepStrictEqual(proof.prerequisite,preparedEvidence?.prerequisite))throw Error('Fee prerequisite evidence drift') + assertTestFeeVoidFaultProof(proof,scope) + } await cleanup() return { ...proof, restored: true, ...(direction!=='LOADING'?{restoration}:{}) } } finally { if(direction!=='LOADING')await cleanup() } diff --git a/scripts/test-event-fault-runtime.py b/scripts/test-event-fault-runtime.py index 7a4662f5d..de8b5c469 100644 --- a/scripts/test-event-fault-runtime.py +++ b/scripts/test-event-fault-runtime.py @@ -6,6 +6,7 @@ Only the original DEAD event can be retried through its protected HTTP API. """ import base64 import copy +import decimal import hashlib import http.client import json @@ -103,21 +104,34 @@ def private_write(path, value): def scope_directory(scope, direction='LOADING'): - if direction not in ('LOADING', 'UNLOAD', 'DELIVERY'): + if direction not in ('LOADING', 'UNLOAD', 'DELIVERY', 'FEE_VOID'): 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', 'deliveryId', 'sourceInboundId', 'sourceOrderId', 'rootOrderId') if direction == 'DELIVERY' else ('customerId', 'containerBookingId') if direction == 'UNLOAD' else ('customerId', 'loadingPlanId', 'containerBookingId') + keys = ('customerId', 'sourceFeeId', 'sourceVersion', 'sourceType') if direction == 'FEE_VOID' else ('customerId', 'deliveryId', 'sourceInboundId', 'sourceOrderId', 'rootOrderId') if direction == 'DELIVERY' else ('customerId', 'containerBookingId') if direction == 'UNLOAD' else ('customerId', 'loadingPlanId', 'containerBookingId') if direction != 'LOADING' and set(scope) != {'marker', *keys}: raise RuntimeError('INVALID_SCOPE') + if direction == 'FEE_VOID' and scope.get('sourceType') not in FEE_SOURCE_TYPES: + raise RuntimeError('INVALID_FEE_SOURCE_TYPE') + if direction == 'FEE_VOID' and (not str(scope.get('sourceVersion', '')).isdigit() or not 0 < int(scope['sourceVersion']) <= 2147483647): + raise RuntimeError('INVALID_SOURCE_VERSION') for key in keys: + if key == 'sourceType': + continue if not re.fullmatch(r'[1-9][0-9]{0,19}', scope.get(key, '')): raise RuntimeError('INVALID_SCOPE') identity = direction + ':' + marker if direction != 'LOADING' else marker return Path(tempfile.gettempdir()) / ('chongchong-event-' + hashlib.sha256(identity.encode()).hexdigest()[:24]) +FEE_SOURCE_TYPES = ('OVERSEAS_INBOUND_FEE', 'OVERSEAS_PROCUREMENT_FEE', 'OVERSEAS_TRUCK_FEE', 'OVERSEAS_LOOSEBOARD_FEE', 'OVERSEAS_WAREHOUSE_FEE', 'OVERSEAS_CONTAINER_FEE', 'OVERSEAS_ORDER_FEE', 'OVERSEAS_DELIVERY_FEE', 'OVERSEAS_RELABEL_FEE') + + +def fee_void(state): + return state.get('direction') == 'FEE_VOID' + + def unload(state): return state.get('direction', 'LOADING') == 'UNLOAD' @@ -127,7 +141,7 @@ def delivery(state): def reverse_delivery(state): - return state.get('direction') in ('UNLOAD', 'DELIVERY') + return state.get('direction') in ('UNLOAD', 'DELIVERY', 'FEE_VOID') def producer_name(state): @@ -135,6 +149,8 @@ def producer_name(state): def target_setting(state): + if fee_void(state): + return ('FINANCE_SERVICE_URL', 'http://finance:9203') return ('FIRSTLEG_SERVICE_URL', 'http://firstleg:9202') if reverse_delivery(state) else ('LASTLEG_SERVICE_URL', 'http://lastleg-service:9303') @@ -167,7 +183,7 @@ def read_sql(container, sql): if re.search(r'\b(INSERT|UPDATE|DELETE|ALTER|DROP|TRUNCATE|CALL|INTO|OUTFILE)\b', sql, re.I): raise RuntimeError('NON_READ_ONLY_SQL') e = env_of(container) - prefix = {'/firstleg-test-business': 'FIRSTLEG', '/firstleg-test-lastleg': 'LASTLEG', '/firstleg-test-order': 'ORDER'}[container['Name']] + prefix = {'/firstleg-test-business': 'FIRSTLEG', '/firstleg-test-lastleg': 'LASTLEG', '/firstleg-test-order': 'ORDER', '/firstleg-test-finance': 'FINANCE'}[container['Name']] url = urllib.parse.urlsplit(e[prefix + '_DB_URL'].removeprefix('jdbc:')) host = '127.0.0.1' if url.hostname == 'host.docker.internal' else url.hostname transaction = 'SET SESSION TRANSACTION ISOLATION LEVEL REPEATABLE READ; START TRANSACTION WITH CONSISTENT SNAPSHOT, READ ONLY;\n' + sql + '\nCOMMIT;' @@ -182,6 +198,8 @@ def read_sql(container, sql): def producer(scope, direction='LOADING'): + if direction == 'FEE_VOID': + return fee_event(scope, 'VOIDED') if direction == 'DELIVERY': rows = read_sql(inspect('firstleg-test-lastleg'), "SELECT JSON_OBJECT('entryId',CAST(id AS CHAR),'eventId',JSON_UNQUOTE(JSON_EXTRACT(payload_json,'$.eventId')),'payloadHash',JSON_UNQUOTE(JSON_EXTRACT(payload_json,'$.payloadHash')),'eventType',event_type,'customerId',customer_id,'aggregateId',aggregate_id,'aggregateType',aggregate_type,'eventKey',event_key,'envelope',JSON_EXTRACT(payload_json,'$'),'publishStatus',status,'failedAttempts',retry_count) FROM ll_event_outbox WHERE customer_id='" + scope['customerId'] + "' AND event_type='OverseasDeliveryCompleted' AND JSON_UNQUOTE(JSON_EXTRACT(payload_json,'$.payload.deliveryOrderId'))='" + scope['deliveryId'] + "';") @@ -212,7 +230,101 @@ def business_ids(scope): return [row['id'] for row in rows] +def fee_canonical(value): + """Match EventPayloadCanonicalizer's sorted JSON and stripped BigDecimal.toString.""" + if isinstance(value, dict): + return '{' + ','.join(json.dumps(key, ensure_ascii=False) + ':' + fee_canonical(value[key]) for key in sorted(value)) + '}' + if isinstance(value, list): + return '[' + ','.join(fee_canonical(item) for item in value) + ']' + if isinstance(value, decimal.Decimal): + with decimal.localcontext() as context: + context.prec = max(28, len(value.as_tuple().digits)) + number = value.normalize() + if not number: + return '0' + return format(number, 'f') if number.as_tuple().exponent <= 0 and number.adjusted() >= -6 else str(number) + return json.dumps(value, ensure_ascii=False, separators=(',', ':')) + + +def fee_event(scope, action): + if action not in ('PUBLISHED', 'ADJUSTED', 'VOIDED'): + raise RuntimeError('INVALID_FEE_ACTION') + event_key = 'BusinessFeeSourceChanged:LASTLEG:' + scope['sourceType'] + ':' + scope['sourceFeeId'] + ':' + scope['sourceVersion'] + ':' + action + 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,'$'),'envelopeJson',payload_json,'publishStatus',status,'failedAttempts',retry_count) FROM ll_event_outbox WHERE event_type='BusinessFeeSourceChanged' AND event_key='" + event_key + "';") + if len(rows) > 1: + raise RuntimeError('AMBIGUOUS_FEE_EVENT') + if not rows: + return None + row = rows[0] + payload = json.loads(row['envelopeJson'], parse_float=decimal.Decimal, parse_int=decimal.Decimal)['payload'] + row['payloadCanonicalJson'] = fee_canonical(payload) + unsigned = {**payload, 'action': None, 'sourceStatus': 'ACTIVE', 'sourceHash': None} + row['sourceCanonicalJson'] = fee_canonical(unsigned) + if hashlib.sha256(row['payloadCanonicalJson'].encode()).hexdigest() != row['payloadHash'].removeprefix('sha256:') or payload.get('sourceHash') != 'sha256:' + hashlib.sha256(row['sourceCanonicalJson'].encode()).hexdigest(): + raise RuntimeError('FEE_PAYLOAD_HASH_MISMATCH') + if row['customerId'] != scope['customerId'] or row['aggregateId'] != scope['sourceFeeId'] or row['aggregateType'] != scope['sourceType']: + raise RuntimeError('FEE_EVENT_SCOPE_MISMATCH') + if payload.get('sourceSystem') != 'LASTLEG' or payload.get('customerId') != scope['customerId'] or payload.get('sourceFeeId') != scope['sourceFeeId'] or str(payload.get('sourceVersion')) != scope['sourceVersion'] or payload.get('sourceType') != scope['sourceType'] or payload.get('action') != action: + raise RuntimeError('FEE_PAYLOAD_SCOPE_MISMATCH') + return row + + +def fee_inbox(scope, event_id): + if not re.fullmatch(r'[0-9a-fA-F-]{36}|[0-9A-HJKMNP-TV-Z]{26}', event_id): + raise RuntimeError('INVALID_EVENT_IDENTITY') + return read_sql(inspect('firstleg-test-finance'), + "SELECT JSON_OBJECT('id',CAST(id AS CHAR),'eventId',event_id,'sourceSystem',source_system,'eventType',event_type,'envelope',JSON_EXTRACT(payload,'$'),'payloadHash',payload_hash,'resultResourceId',CAST(result_resource_id AS CHAR),'status',status,'errorCode',error_code,'receivedAt',CAST(received_at AS CHAR),'processedAt',CAST(processed_at AS CHAR)) FROM fin_inbox_event WHERE source_system='LASTLEG' AND event_id='" + event_id + "';") + + +def fee_facts(scope, event_id): + lastleg, finance = inspect('firstleg-test-lastleg'), inspect('firstleg-test-finance') + require_test(lastleg, 'firstleg-test-lastleg') + require_test(finance, 'firstleg-test-finance') + # Scalar decimals remain strings. Original event decimals are retained in envelopeJson. + def columns(names): + return ','.join("'" + name + "',CAST(" + re.sub(r'([A-Z])', lambda match: '_' + match[1].lower(), name) + " AS CHAR)" for name in names.split()) + local_names = 'id sourceFeeId sourceVersion customerId customerName sourceType sourceNo businessType businessId businessNo orderId inboundId inboundCode containerId containerBookingId containerNo warehouseId feeItemCode feeItemName direction supplierId supplierName supplierCategory settlementMode paidAt paymentMethod paymentReferenceNo businessDate currency amount originalCurrency originalAmount exchangeRate exchangeRateMonth chargeableQuantity quantityUnit unitPrice periodStart periodEnd reviewStatus publicationStatus sourceStatus reviewReason reviewerId reviewedAt createdBy updatedBy createdAt updatedAt' + local = read_sql(lastleg, "SELECT JSON_OBJECT(" + columns(local_names) + ",'paymentVoucherFileIds',payment_voucher_file_ids) FROM ll_finance_fee_source WHERE source_fee_id=" + scope['sourceFeeId'] + " ORDER BY source_version,id;") + where = "source_system='LASTLEG' AND source_type='" + scope['sourceType'] + "' AND source_fee_id=" + scope['sourceFeeId'] + finance_names = 'id sourceSystem sourceFeeId sourceFeeVersion customerId sourceType sourceNo businessType businessId businessNo inboundCode warehouseId direction orderId inboundId containerId containerBookingId containerNo feeItemCode feeItemName amount currencyCode settlementMode paidAt paymentMethod paymentReferenceNo sourceHash reviewStatus reviewReason operationVersion walletHoldId status createdAt updatedAt' + ids = 'SELECT id FROM fin_fee_source WHERE ' + where + rows = read_sql(finance, + "SELECT JSON_OBJECT('kind','source'," + columns(finance_names) + ",'sourceSnapshot',source_snapshot,'paymentVoucherFileIds',payment_voucher_file_ids) FROM fin_fee_source WHERE " + where + " ORDER BY source_fee_version,id;\n" + + "SELECT JSON_OBJECT('kind','bill'," + columns('id tenantId billId sourceId sourceVersion amount currencyCode createdAt') + ") FROM fin_bill_line WHERE source_id IN (" + ids + ") ORDER BY id;\n" + + "SELECT JSON_OBJECT('kind','settlement'," + columns('id tenantId settlementId feeSourceId amount currencyCode createdAt') + ") FROM fin_settlement_line WHERE fee_source_id IN (" + ids + ") ORDER BY id;\n" + + "SELECT JSON_OBJECT('kind','cost'," + columns('id tenantId sourceSystem sourceType sourceFeeId sourceFeeVersion sourceHash status version costAmount createdAt voidedAt') + ") FROM fin_cost WHERE " + where + " ORDER BY id;") + return {'lastleg': local, 'finance': {**{kind: [row for row in rows if row['kind'] == kind] for kind in ('source', 'bill', 'settlement', 'cost')}, 'inbox': fee_inbox(scope, event_id)}} + + +def fee_prerequisite(scope, wait=False): + if fee_event(scope, 'VOIDED') is not None: + raise RuntimeError('FEE_VOID_EVENT_ALREADY_EXISTS') + publication = fee_event(scope, 'PUBLISHED' if scope['sourceVersion'] == '1' else 'ADJUSTED') + if not publication or publication['publishStatus'] not in ('PENDING', 'PROCESSING', 'PUBLISHED'): + raise RuntimeError('FEE_PUBLICATION_NOT_READY') + facts = fee_facts(scope, publication['eventId']) + current = [row for row in facts['lastleg'] if row['sourceVersion'] == scope['sourceVersion']] + financial = [row for row in facts['finance']['source'] if row['sourceFeeVersion'] == scope['sourceVersion']] + if len(current) != 1 or len(financial) != 1 or any(row['customerId'] != scope['customerId'] or row['sourceType'] != scope['sourceType'] for row in facts['lastleg'] + facts['finance']['source']): + raise RuntimeError('FEE_SOURCE_SCOPE_MISMATCH') + if max(int(row['sourceVersion']) for row in facts['lastleg']) != int(scope['sourceVersion']) or current[0]['sourceStatus'] != 'ACTIVE' or current[0]['reviewStatus'] != 'APPROVED' or current[0]['publicationStatus'] != 'PUBLISHED' or current[0]['direction'] != 'RECEIVABLE': + raise RuntimeError('FEE_SOURCE_NOT_ACTIVE_RECEIVABLE') + if financial[0]['status'] != 'AVAILABLE' or financial[0]['walletHoldId'] is not None or any(facts['finance'][key] for key in ('bill', 'settlement', 'cost')): + raise RuntimeError('FEE_SOURCE_ALREADY_FINANCIALLY_LINKED') + inbox = facts['finance']['inbox'] + if len(inbox) != 1 or inbox[0]['status'] != 'PROCESSED' or inbox[0]['payloadHash'] != publication['payloadHash'] or inbox[0]['resultResourceId'] != financial[0]['id']: + raise RuntimeError('FEE_PUBLICATION_INBOX_MISMATCH') + attempts = read_sql(inspect('firstleg-test-lastleg'), "SELECT JSON_OBJECT('entryId',entry_id,'eventId',event_id,'outcome',outcome) FROM event_delivery_attempt WHERE entry_id='" + publication['entryId'] + "' AND event_type='BusinessFeeSourceChanged' ORDER BY started_at,attempt_id;") + ready = publication['publishStatus'] == 'PUBLISHED' and any(row['eventId'] == publication['eventId'] and row['outcome'] in ('APPLIED', 'DUPLICATE') for row in attempts) + if not ready and not wait: + raise RuntimeError('FEE_PUBLICATION_ACK_NOT_READY') + return {'ready': ready, 'publication': publication, 'business': facts, 'attempts': attempts} + + def business_snapshot(scope, state, row): + if fee_void(state): + return {'fee': fee_facts(scope, row['eventId'])} if delivery(state): return {'business': delivery_facts(scope, row['eventId'])} return {'container': container_facts(scope)} if unload(state) else {'businessOrderIds': business_ids(scope)} @@ -245,10 +357,11 @@ def delivery_facts(scope, event_id): def prepare(scope, directory, config): if delivery(config) and producer(scope, 'DELIVERY') is not None: raise RuntimeError('DELIVERY_EVENT_ALREADY_EXISTS') + prerequisite = fee_prerequisite(scope) if fee_void(config) else None directory.mkdir(mode=0o700) original = inspect(producer_name(config)) require_test(original, producer_name(config)) - consumer_name = 'firstleg-test-business' if reverse_delivery(config) else 'firstleg-test-lastleg' + consumer_name = 'firstleg-test-finance' if fee_void(config) else 'firstleg-test-business' if reverse_delivery(config) else 'firstleg-test-lastleg' consumer = inspect(consumer_name) require_test(consumer, consumer_name) old_env = env_of(original) @@ -268,10 +381,12 @@ def prepare(scope, directory, config): proxy.write_bytes(base64.b64decode(config['proxySource'])) ip = consumer['NetworkSettings']['Networks']['firstleg-test-network']['IPAddress'] log = open(directory / 'proxy-output.log', 'wb') - upstream = ':9202/internal/firstleg/v1/container-events' if reverse_delivery(config) else ':9303/internal/lastleg/events' - selector = [] if delivery(config) else ['--event-type', 'OverseasContainerUnloaded', '--container-booking-id', scope['containerBookingId']] if unload(config) else ['--event-type', 'FirstLegLoadingPlanCompleted', '--loading-plan-id', scope['loadingPlanId']] + upstream = ':9203/internal/finance/v1/business-fee-source-events' if fee_void(config) else ':9202/internal/firstleg/v1/container-events' if reverse_delivery(config) else ':9303/internal/lastleg/events' + selector = [] if (delivery(config) or fee_void(config)) else ['--event-type', 'OverseasContainerUnloaded', '--container-booking-id', scope['containerBookingId']] if unload(config) else ['--event-type', 'FirstLegLoadingPlanCompleted', '--loading-plan-id', scope['loadingPlanId']] if delivery(config): selector = ['--event-type', 'OverseasDeliveryCompleted', '--delivery-id', scope['deliveryId']] + if fee_void(config): + selector = ['--event-type', 'BusinessFeeSourceChanged', '--fee-source-id', scope['sourceFeeId'], '--source-version', scope['sourceVersion']] 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')], @@ -279,9 +394,9 @@ def prepare(scope, directory, config): 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} + 'prerequisite': prerequisite, 'operationsRequestId': str(uuid.uuid4()), 'prepared': False, 'restored': False} private_write(directory / 'state.json', state) - return {'created': True, 'originalImageId': state['originalImageId'], 'originalFingerprint': state['originalFingerprint']} + return {'created': True, 'originalImageId': state['originalImageId'], 'originalFingerprint': state['originalFingerprint'], **({'prerequisite': prerequisite} if fee_void(config) else {})} def activate(scope, directory, state): @@ -316,6 +431,12 @@ def recover(scope, directory, state): if pinned['eventId'] != row['eventId'] or pinned['payloadHash'] != row['payloadHash']: raise RuntimeError('PINNED_EVENT_MISMATCH') state['beforeRetry'] = {**row, **business_snapshot(scope, state, row)} + if fee_void(state): + facts = state['beforeRetry']['fee'] + local = [value for value in facts['lastleg'] if value['sourceVersion'] == scope['sourceVersion']] + financial = [value for value in facts['finance']['source'] if value['sourceFeeVersion'] == scope['sourceVersion']] + if len(local) != 1 or local[0]['sourceStatus'] != 'ACTIVE' or len(financial) != 1 or financial[0]['status'] != 'VOIDED': + raise RuntimeError('FEE_VOID_DEAD_PHASE_MISMATCH') private_write(directory / 'state.json', state) control(state, 'PASS') container = inspect(producer_name(state)) @@ -375,7 +496,7 @@ def proof(scope, state): event_id, entry_id = row['eventId'], row['entryId'] if not re.fullmatch(r'[0-9a-fA-F-]{36}|[0-9A-HJKMNP-TV-Z]{26}', event_id) or not re.fullmatch(r'[1-9][0-9]{0,19}', entry_id): raise RuntimeError('INVALID_EVENT_IDENTITY') - event_type = 'OverseasDeliveryCompleted' if delivery(state) else 'OverseasContainerUnloaded' if unload(state) else 'FirstLegLoadingPlanCompleted' + event_type = 'BusinessFeeSourceChanged' if fee_void(state) else 'OverseasDeliveryCompleted' if delivery(state) else 'OverseasContainerUnloaded' if unload(state) else 'FirstLegLoadingPlanCompleted' attempts = read_sql(inspect(producer_name(state)), "SELECT JSON_OBJECT('attemptId',attempt_id,'serviceName',service_name,'entryId',entry_id,'eventId',event_id,'targetName',target_name,'outcome',outcome) FROM event_delivery_attempt WHERE entry_id='" + entry_id + "' AND event_type='" + event_type + "' ORDER BY started_at,attempt_id;") @@ -383,7 +504,9 @@ def proof(scope, 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 reverse_delivery(state) else "SELECT JSON_OBJECT('requestId',request_id,'entryId',entry_id,'eventId',event_id,'payloadHash',payload_hash,'previousFailures',previous_failures) FROM event_delivery_retry_audit WHERE entry_id='" + entry_id + "' AND event_type='" + event_type + "';") - if reverse_delivery(state): + if fee_void(state): + consumers = fee_inbox(scope, row['eventId']) + elif reverse_delivery(state): consumers = read_sql(inspect('firstleg-test-business'), "SELECT JSON_OBJECT('eventId',event_id,'payloadHash',payload_hash,'status',status,'eventType',event_type,'aggregateId',aggregate_id,'payload',JSON_EXTRACT(payload_json,'$')) FROM fl_container_event_inbox WHERE event_id='" + event_id + "';") else: @@ -395,6 +518,9 @@ def proof(scope, state): 'proxyRecords': control(state)['records'], 'beforeRetry': state['beforeRetry'], 'afterRetry': {**row, **business_snapshot(scope, state, row)}, 'retryAudit': audit[0] if len(audit) == 1 else None, 'retryAuditCount': len(audit), 'retryRequestId': state['operationsRequestId'], 'restored': False} + if fee_void(state): + result['prerequisite'] = state['prerequisite'] + result['publicationNow'] = fee_event(scope, 'PUBLISHED' if scope['sourceVersion'] == '1' else 'ADJUSTED') private_write(scope_directory(scope, state.get('direction', 'LOADING')) / 'proof.json', result) return result @@ -440,6 +566,8 @@ def main(config): scope, action = config['scope'], config['action'] direction = config.get('direction', 'LOADING') directory = scope_directory(scope, direction) + if action == 'preflight' and direction == 'FEE_VOID': + return fee_prerequisite(scope, wait=True) if action == 'prepare': return prepare(scope, directory, config) state = json.loads((directory / 'state.json').read_text()) diff --git a/scripts/test-fee-void-fault-proof.mjs b/scripts/test-fee-void-fault-proof.mjs new file mode 100644 index 000000000..e2f035fe7 --- /dev/null +++ b/scripts/test-fee-void-fault-proof.mjs @@ -0,0 +1,91 @@ +import { createHash } from 'node:crypto' +import { isDeepStrictEqual as equal } from 'node:util' + +export const feeVoidScopeKeys=['marker','customerId','sourceFeeId','sourceVersion','sourceType'] +export const feeSourceTypes=['OVERSEAS_INBOUND_FEE','OVERSEAS_PROCUREMENT_FEE','OVERSEAS_TRUCK_FEE','OVERSEAS_LOOSEBOARD_FEE', + 'OVERSEAS_WAREHOUSE_FEE','OVERSEAS_CONTAINER_FEE','OVERSEAS_ORDER_FEE','OVERSEAS_DELIVERY_FEE','OVERSEAS_RELABEL_FEE'] +const hash=value=>'sha256:'+createHash('sha256').update(value,'utf8').digest('hex') +const without=(row,keys)=>Object.fromEntries(Object.entries(row).filter(([key])=>!keys.includes(key))) +const date=value=>Date.parse(String(value).replace(' ','T').replace(/Z$/,'')+'Z') + +export function assertTestFeeVoidFaultProof(proof,scope) { + const need=(passed,code)=>{if(!passed)throw Error('Invalid fee void fault proof: '+code)} + need(proof?.readOnly===true && feeVoidScopeKeys.every(key=>typeof scope[key]==='string'&&proof[key]===scope[key]) + && Object.keys(scope).length===feeVoidScopeKeys.length && feeSourceTypes.includes(scope.sourceType) + && /^[A-Za-z0-9_-]{1,96}$/.test(scope.marker) && ['customerId','sourceFeeId','sourceVersion'].every(key=>/^[1-9][0-9]{0,19}$/.test(scope[key])) + && Number(scope.sourceVersion)<=2147483647, 'SCOPE') + const event=(row,action)=>{ + need(row && /^[1-9][0-9]{0,19}$/.test(row.entryId) && /^[0-9a-fA-F-]{36}$/.test(row.eventId), 'EVENT_IDENTITY') + const env=row.envelope,p=env?.payload + need(row.eventType==='BusinessFeeSourceChanged' && row.customerId===scope.customerId && row.aggregateId===scope.sourceFeeId + && row.aggregateType===scope.sourceType && row.eventKey===`BusinessFeeSourceChanged:LASTLEG:${scope.sourceType}:${scope.sourceFeeId}:${scope.sourceVersion}:${action}` + && env?.eventId===row.eventId && env.eventType===row.eventType && env.tenantId===scope.customerId && Number(env.eventVersion)===1 + && env.aggregateType===scope.sourceType && env.aggregateId===scope.sourceFeeId && String(env.aggregateVersion)===scope.sourceVersion + && p?.action===action && p.sourceSystem==='LASTLEG' && p.customerId===scope.customerId && p.sourceFeeId===scope.sourceFeeId + && String(p.sourceVersion)===scope.sourceVersion && p.sourceType===scope.sourceType && p.direction==='RECEIVABLE' + && p.currency==='CNY' && p.reviewStatus==='APPROVED' && p.publicationStatus==='PUBLISHED' + && p.sourceStatus===(action==='VOIDED'?'VOIDED':'ACTIVE'), 'EVENT_SCOPE') + // Runtime preserves exact decimal lexemes and applies Java BigDecimal canonicalization. + let parsed,canonical,unsigned + try {parsed=JSON.parse(row.envelopeJson);canonical=JSON.parse(row.payloadCanonicalJson);unsigned=JSON.parse(row.sourceCanonicalJson)} + catch {need(false,'CANONICAL_JSON')} + need(equal(parsed,env) && equal(canonical,p) && equal(unsigned,{...p,action:null,sourceStatus:'ACTIVE',sourceHash:null}) + && hash(row.payloadCanonicalJson)===row.payloadHash && env.payloadHash===row.payloadHash + && hash(row.sourceCanonicalJson)===p.sourceHash, 'PAYLOAD_HASH') + return p + } + const row=proof.producer,payload=event(row,'VOIDED'),prior=proof.prerequisite,publication=prior?.publication + const original=event(publication,scope.sourceVersion==='1'?'PUBLISHED':'ADJUSTED') + need(prior.ready===true && publication.publishStatus==='PUBLISHED' && equal(publication,proof.publicationNow) + && publication.eventId!==row.eventId && publication.entryId!==row.entryId + && publication.sourceCanonicalJson===row.sourceCanonicalJson && original.sourceHash===payload.sourceHash + && prior.attempts?.some(a=>a.entryId===publication.entryId&&a.eventId===publication.eventId&&['APPLIED','DUPLICATE'].includes(a.outcome)), 'PUBLICATION_PREREQUISITE') + const before=proof.beforeRetry,after=proof.afterRetry + need(before?.publishStatus==='DEAD' && before.failedAttempts===12 && after?.publishStatus==='PUBLISHED' && row.publishStatus==='PUBLISHED' + && [before,after].every(value=>value.entryId===row.entryId && value.eventId===row.eventId && value.payloadHash===row.payloadHash + && equal(value.envelope,row.envelope) && value.envelopeJson===row.envelopeJson + && value.payloadCanonicalJson===row.payloadCanonicalJson && value.sourceCanonicalJson===row.sourceCanonicalJson), 'ORIGINAL_RETRY') + const attempts=proof.attempts,records=proof.proxyRecords?.filter(value=>value.matched===true) + need(attempts?.length===13 && new Set(attempts.map(a=>a.attemptId)).size===13 && attempts.every((a,i)=>a.entryId===row.entryId + && a.eventId===row.eventId && a.serviceName==='lastleg' && a.outcome===(i<11?'RETRY':i===11?'DEAD':'DUPLICATE')), 'ATTEMPTS') + need(records?.length===13 && records.every((a,i)=>a.eventId===row.eventId && a.payloadHash===row.payloadHash && a.attempt===i+1 + && a.outcome===(i===0?'APPLIED':'DUPLICATE') && (i<12?a.errorCode==='ACK_LOSS_INJECTED':a.errorCode===null)), 'REAL_ACK_LOSS') + const audit=proof.retryAudit + need(proof.retryAuditCount===1 && audit?.requestId===proof.retryRequestId && audit.entryId===row.entryId && audit.eventId===row.eventId + && audit.payloadHash===row.payloadHash && audit.previousFailures===12, 'RETRY_AUDIT') + const initial=prior.business,dead=before.fee,done=after.fee + need(initial && dead && done && equal(dead.finance,done.finance), 'FINANCE_CHANGED_AFTER_DUPLICATE') + const selected=(facts,local)=>{ + const rows=local?facts.lastleg:facts.finance?.source + need(Array.isArray(rows)&&rows.length>0&&new Set(rows.map(a=>a.id)).size===rows.length, 'SOURCE_ROWS') + need(rows.every(a=>a.customerId===scope.customerId && a.sourceType===scope.sourceType && a.sourceFeeId===scope.sourceFeeId), 'SOURCE_SCOPE') + const result=rows.filter(a=>String(local?a.sourceVersion:a.sourceFeeVersion)===scope.sourceVersion) + need(result.length===1, 'UNIQUE_SOURCE_VERSION') + need(Math.max(...rows.map(a=>Number(local?a.sourceVersion:a.sourceFeeVersion)))===Number(scope.sourceVersion), 'LATEST_VERSION') + return result[0] + } + const l0=selected(initial,true),ld=selected(dead,true),la=selected(done,true) + const f0=selected(initial,false),fd=selected(dead,false) + need(l0.sourceStatus==='ACTIVE' && ld.sourceStatus==='ACTIVE' && la.sourceStatus==='VOIDED' + && [l0,ld,la].every(a=>a.reviewStatus==='APPROVED'&&a.publicationStatus==='PUBLISHED'&&a.direction==='RECEIVABLE'), 'LASTLEG_ACK_BOUNDARY') + need(equal(without(l0,['reviewReason','updatedBy','updatedAt']),without(ld,['reviewReason','updatedBy','updatedAt'])) + && equal(without(ld,['sourceStatus','updatedAt']),without(la,['sourceStatus','updatedAt'])) + && typeof ld.reviewReason==='string' && ld.reviewReason.trim() && date(la.updatedAt)>=date(ld.updatedAt), 'LASTLEG_UNEXPECTED_CHANGE') + need(equal(initial.lastleg.filter(a=>a.id!==l0.id),dead.lastleg.filter(a=>a.id!==ld.id)) + && equal(dead.lastleg.filter(a=>a.id!==ld.id),done.lastleg.filter(a=>a.id!==la.id)), 'OTHER_LOCAL_VERSIONS_CHANGED') + need(f0.status==='AVAILABLE' && fd.status==='VOIDED' && Number(fd.operationVersion)===Number(f0.operationVersion)+1 + && equal(without(f0,['status','operationVersion','updatedAt']),without(fd,['status','operationVersion','updatedAt'])) + && f0.sourceHash===payload.sourceHash && f0.direction==='RECEIVABLE' && f0.currencyCode==='CNY' + && f0.amount===l0.amount && f0.walletHoldId===null && equal(f0.sourceSnapshot,original), 'FINANCE_VOID_FACT') + need(equal(initial.finance.source.filter(a=>a.id!==f0.id),dead.finance.source.filter(a=>a.id!==fd.id)), 'OTHER_FINANCE_VERSIONS_CHANGED') + for(const facts of [initial,dead,done]) for(const key of ['bill','settlement','cost']) need(Array.isArray(facts.finance[key])&&facts.finance[key].length===0,'FINANCIAL_LINKS') + const checkInbox=(rows,producer,resource)=>{ + need(rows?.length===1 && rows[0].sourceSystem==='LASTLEG' && rows[0].eventType==='BusinessFeeSourceChanged' + && rows[0].eventId===producer.eventId && rows[0].payloadHash===producer.payloadHash && rows[0].status==='PROCESSED' + && rows[0].resultResourceId===resource.id && rows[0].processedAt && rows[0].errorCode===null + && equal(rows[0].envelope,producer.envelope), 'FINANCE_INBOX') + } + checkInbox(initial.finance.inbox,publication,f0); checkInbox(dead.finance.inbox,row,fd) + need(proof.consumer?.count===1 && equal(without(proof.consumer,['count']),dead.finance.inbox[0]), 'CONSUMER_IDENTITY') + return proof +} diff --git a/scripts/test-fee-void-fault-proof.test.mjs b/scripts/test-fee-void-fault-proof.test.mjs new file mode 100644 index 000000000..0f6c8b42a --- /dev/null +++ b/scripts/test-fee-void-fault-proof.test.mjs @@ -0,0 +1,189 @@ +import assert from 'node:assert/strict' +import test from 'node:test' +import { createHash } from 'node:crypto' +import { spawnSync } from 'node:child_process' +import { createTestFeeVoidFaultExecutor } from './test-event-fault-executor.mjs' +import { assertTestFeeVoidFaultProof } from './test-fee-void-fault-proof.mjs' + +const scope={marker:'FEE-VOID-OFFLINE',customerId:'7',sourceFeeId:'123',sourceVersion:'1',sourceType:'OVERSEAS_WAREHOUSE_FEE'} +const env={CHONGCHONG_ENV:'test',JENKINS_URL:'https://jenkins.example.invalid/',JENKINS_USER:'offline',JENKINS_API_TOKEN:'offline',FBA_CORE_JENKINS_AGENT:'java-test-agent'} +const sort=value=>Array.isArray(value)?value.map(sort):value&&typeof value==='object'?Object.fromEntries(Object.keys(value).sort().map(key=>[key,sort(value[key])])):value +const text=value=>JSON.stringify(sort(value)), hash=value=>'sha256:'+createHash('sha256').update(value).digest('hex') +const clone=value=>JSON.parse(JSON.stringify(value)) +const imageId='sha256:'+'a'.repeat(64),fingerprint='b'.repeat(64) +const restoration={restored:true,healthy:true,proxyStopped:true,producerName:'firstleg-test-lastleg',originalImageId:imageId,currentImageId:imageId,originalFingerprint:fingerprint,currentFingerprint:fingerprint} +function fixture() { + const base={action:null,sourceSystem:'LASTLEG',sourceType:scope.sourceType,sourceFeeId:'123',sourceVersion:1,customerId:'7',customerName:'Offline', + sourceNo:'WH-OFFLINE',businessType:'WAREHOUSE_SERVICE',businessId:'200',businessNo:'WH-OFFLINE',orderId:null,inboundId:null,inboundCode:null, + containerId:null,containerBookingId:null,containerNo:null,warehouseId:'9',feeItemCode:'TEST',feeItemName:'Offline fee',direction:'RECEIVABLE', + supplierId:null,supplierName:null,supplierCategory:null,settlementMode:null,paidAt:null,paymentMethod:null,paymentReferenceNo:null,paymentVoucherFileIds:[], + businessDate:'2026-10-03',currency:'CNY',totalAmount:7.25,originalCurrency:'CNY',originalAmount:7.25,exchangeRate:1,exchangeRateMonth:'2026-10', + chargeableQuantity:null,quantityUnit:null,unitPrice:null,periodStart:null,periodEnd:null,reviewStatus:'APPROVED',publicationStatus:'PUBLISHED',sourceStatus:'ACTIVE',sourceHash:null} + const sourceCanonicalJson=text(base),sourceHash=hash(sourceCanonicalJson) + const event=(action,index)=>{ + const payload={...base,action,sourceStatus:action==='VOIDED'?'VOIDED':'ACTIVE',sourceHash} + const payloadCanonicalJson=text(payload),payloadHash=hash(payloadCanonicalJson) + const eventId=`00000000-0000-4000-8000-00000000000${index}` + const envelope={eventId,eventType:'BusinessFeeSourceChanged',eventVersion:1,tenantId:'7',aggregateType:scope.sourceType,aggregateId:'123',aggregateVersion:1,payload,payloadHash} + return {entryId:String(index),eventId,payloadHash,eventType:envelope.eventType,customerId:'7',aggregateId:'123',aggregateType:scope.sourceType, + eventKey:`BusinessFeeSourceChanged:LASTLEG:${scope.sourceType}:123:1:${action}`,envelope,envelopeJson:JSON.stringify(envelope),payloadCanonicalJson,sourceCanonicalJson,publishStatus:'PUBLISHED',failedAttempts:0} + } + const publication=event('PUBLISHED',1),producer=event('VOIDED',2) + const local={id:'500',sourceFeeId:'123',sourceVersion:'1',customerId:'7',sourceType:scope.sourceType,sourceStatus:'ACTIVE',reviewStatus:'APPROVED',publicationStatus:'PUBLISHED',direction:'RECEIVABLE',amount:'7.250000',businessType:'WAREHOUSE_SERVICE',businessId:'200',reviewReason:'approved',updatedBy:'7',updatedAt:'2026-10-03 01:00:00'} + const finance={id:'600',sourceSystem:'LASTLEG',sourceFeeId:'123',sourceFeeVersion:'1',customerId:'7',sourceType:scope.sourceType,direction:'RECEIVABLE',currencyCode:'CNY',amount:'7.250000',walletHoldId:null,operationVersion:'0',status:'AVAILABLE',sourceHash,sourceSnapshot:publication.envelope.payload,updatedAt:'2026-10-03 01:00:01'} + const inbox=(row,id)=>({id,eventId:row.eventId,sourceSystem:'LASTLEG',eventType:'BusinessFeeSourceChanged',envelope:row.envelope,payloadHash:row.payloadHash,resultResourceId:'600',status:'PROCESSED',errorCode:null,receivedAt:'2026-10-03 01:00:01',processedAt:'2026-10-03 01:00:01'}) + const initial={lastleg:[local],finance:{source:[finance],bill:[],settlement:[],cost:[],inbox:[inbox(publication,'700')]}} + const dead=clone(initial);dead.lastleg[0].reviewReason='void reason';dead.lastleg[0].updatedAt='2026-10-03 01:01:00' + Object.assign(dead.finance.source[0],{status:'VOIDED',operationVersion:'1',updatedAt:'2026-10-03 01:01:01'});dead.finance.inbox=[inbox(producer,'701')] + const after=clone(dead);Object.assign(after.lastleg[0],{sourceStatus:'VOIDED',updatedAt:'2026-10-03 01:02:00'}) + const prior={ready:true,publication,business:initial,attempts:[{entryId:'1',eventId:publication.eventId,outcome:'APPLIED'}]} + const requestId='00000000-0000-4000-8000-000000000003' + return clone({...scope,readOnly:true,producer,prerequisite:prior,publicationNow:publication, + beforeRetry:{...producer,publishStatus:'DEAD',failedAttempts:12,fee:dead},afterRetry:{...producer,fee:after}, + consumer:{...dead.finance.inbox[0],count:1}, + attempts:Array.from({length:13},(_,i)=>({attemptId:`a${i}`,entryId:'2',eventId:producer.eventId,serviceName:'lastleg',outcome:i<11?'RETRY':i===11?'DEAD':'DUPLICATE'})), + proxyRecords:Array.from({length:13},(_,i)=>({matched:true,attempt:i+1,eventId:producer.eventId,payloadHash:producer.payloadHash,outcome:i===0?'APPLIED':'DUPLICATE',errorCode:i<12?'ACK_LOSS_INJECTED':null})), + retryAuditCount:1,retryRequestId:requestId,retryAudit:{requestId,entryId:'2',eventId:producer.eventId,payloadHash:producer.payloadHash,previousFailures:12}}) +} +function config(options) { + const chunks=[...options.body.get('script').matchAll(/'([A-Za-z0-9+/=]+)'/g)].map(m=>m[1]).join('') + const groovy=Buffer.from(chunks,'base64').toString('utf8'),encoded=groovy.match(/'UTF-8'\),'([A-Za-z0-9+/=]+)'\]\)/)?.[1] + return JSON.parse(Buffer.from(encoded,'base64').toString('utf8')) +} +function transport(calls,{pending=false,preflightFailure=false,proof=fixture(),finish={}}={}) { + let recovered=false,preflightCount=0 + return async(_url,options)=>{ + const request=config(options);assert.equal(request.direction,'FEE_VOID');assert.deepEqual(request.scope,scope) + const action=request.action;calls.push(action) + if(action==='recover')recovered=true + if(action==='preflight')preflightCount++ + const data=action==='preflight'?{ready:!pending||preflightCount>1}:action==='prepare'?{originalImageId:imageId,originalFingerprint:fingerprint,prerequisite:proof.prerequisite} + :action==='proxy-ready'?{ready:true}:action==='health'?{healthy:true}:action==='status'?{producer:{publishStatus:recovered?'PUBLISHED':'DEAD',failedAttempts:12}} + :action==='proof'?proof:action==='finish'?{...restoration,...finish}:{} + const result=preflightFailure&&action==='preflight'?{ok:false,error:'FEE_PUBLICATION_NOT_READY'}:{ok:true,data} + return {ok:true,text:async()=> 'TEST_EVENT_JSON='+JSON.stringify(result)} + } +} +test('VOID proof accepts Finance VOIDED at DEAD while Lastleg remains ACTIVE until matched retry ACK',()=>{ + const proof=fixture();assertTestFeeVoidFaultProof(proof,scope) + assert.equal(proof.beforeRetry.fee.lastleg[0].sourceStatus,'ACTIVE') + assert.equal(proof.beforeRetry.fee.finance.source[0].status,'VOIDED') + assert.equal(proof.afterRetry.fee.lastleg[0].sourceStatus,'VOIDED') +}) +test('VOID proof rejects premature local confirmation, extra financial writes, altered source and lost original event evidence',()=>{ + const mutations=[p=>p.beforeRetry.fee.lastleg[0].sourceStatus='VOIDED',p=>p.afterRetry.fee.lastleg[0].sourceStatus='ACTIVE', + p=>p.afterRetry.fee.lastleg[0].amount='8.250000',p=>p.afterRetry.fee.lastleg[0].sourceVersion='2', + p=>p.afterRetry.fee.finance.source[0].operationVersion='2',p=>p.beforeRetry.fee.finance.source[0].status='AVAILABLE', + p=>p.prerequisite.publication.publishStatus='PENDING',p=>p.prerequisite.business.finance.source[0].walletHoldId='99', + p=>p.afterRetry.fee.finance.bill.push({id:'123'}),p=>p.consumer.resultResourceId='601', + p=>p.producer.envelope.payload.sourceFeeId='124',p=>p.beforeRetry.entryId='9',p=>p.retryAuditCount=2, + p=>p.proxyRecords[0].outcome='DUPLICATE',p=>p.attempts.pop(),p=>p.producer.sourceCanonicalJson='{}', + p=>p.prerequisite.business.lastleg.push({...p.prerequisite.business.lastleg[0],id:'501',sourceVersion:'2'})] + for(const mutate of mutations){const proof=fixture();mutate(proof);assert.throws(()=>assertTestFeeVoidFaultProof(proof,scope),/Invalid fee void/)} +}) +test('preflight waits for original publish ACK before a single prepare; restoration precedes successful proof',async()=>{ + const calls=[],executor=createTestFeeVoidFaultExecutor(env,transport(calls,{pending:true})) + for(const invalid of [{...scope,sourceVersion:1},{...scope,sourceType:'UNKNOWN'},{...scope,sourceVersion:'2147483648'},{...scope,action:'VOIDED'}]) + await assert.rejects(executor.prepare(invalid),/Invalid/) + await executor.prepare(scope) + const proof=await executor.readProof(scope);assert.equal(proof.restored,true) + assert.deepEqual(calls,['preflight','preflight','prepare','proxy-ready','activate','health','status','recover','status','proof','restore','health','finish']) + await executor.cleanup();assert.equal(calls.filter(a=>a==='prepare').length,1) +}) +test('preflight failure never mutates environment and cannot silently reuse an unprepared executor',async()=>{ + const calls=[],executor=createTestFeeVoidFaultExecutor(env,transport(calls,{preflightFailure:true})) + await assert.rejects(executor.prepare(scope),/FEE_PUBLICATION_NOT_READY/) + assert.deepEqual(await executor.cleanup(),{attempted:false});await assert.rejects(executor.prepare(scope),/did not complete/) + assert.deepEqual(calls,['preflight']) +}) +test('proof failure restores; missing original configuration or proxy stop remains a blocker',async()=>{ + const proof=fixture();proof.afterRetry.fee.finance.source[0].operationVersion='2' + const calls=[],executor=createTestFeeVoidFaultExecutor(env,transport(calls,{proof})) + await executor.prepare(scope);await assert.rejects(executor.readProof(scope),/FINANCE_CHANGED/) + assert.deepEqual(calls.slice(-3),['restore','health','finish']) + for(const finish of [{proxyStopped:false},{currentFingerprint:'c'.repeat(64)}]) { + const failed=createTestFeeVoidFaultExecutor(env,transport([],{finish})) + await failed.prepare(scope);await assert.rejects(failed.readProof(scope),/Fee void restoration/) + } +}) + +test('unknown prepare result is never reissued and cleanup still attempts original restoration',async()=>{ + const calls=[],base=transport(calls) + const executor=createTestFeeVoidFaultExecutor(env,async(url,options)=>{ + if(config(options).action==='prepare'){calls.push('prepare');throw Error('response lost')} + return base(url,options) + }) + await assert.rejects(executor.prepare(scope),/response lost/) + await assert.rejects(executor.prepare(scope),/did not complete/) + await assert.rejects(executor.cleanup(),/Fee void restoration evidence mismatch/) + assert.deepEqual(calls,['preflight','prepare','restore','health','finish']) +}) + +test('Python FEE_VOID routing, exact decimal hashes, preflight gating and restore reentry are offline',()=>{ + const result=spawnSync('python',['-c',String.raw` +import sys,importlib.util,pathlib,copy,decimal,ast,io +from unittest.mock import patch,mock_open +sys.dont_write_bytecode=True +p=pathlib.Path('scripts/test-event-fault-runtime.py');ast.parse(p.read_text(encoding='utf-8')) +spec=importlib.util.spec_from_file_location('fault',p);m=importlib.util.module_from_spec(spec);spec.loader.exec_module(m) +s={'marker':'FEE-VOID-OFFLINE','customerId':'7','sourceFeeId':'123','sourceVersion':'1','sourceType':'OVERSEAS_WAREHOUSE_FEE'} +assert m.target_setting({'direction':'FEE_VOID'})==('FINANCE_SERVICE_URL','http://finance:9203') +assert m.target_setting({'direction':'DELIVERY'})==('FIRSTLEG_SERVICE_URL','http://firstleg:9202') +assert m.producer_name({'direction':'FEE_VOID'})=='firstleg-test-lastleg' +assert m.fee_canonical(decimal.Decimal('100.000000'))=='1E+2' +assert m.fee_canonical(decimal.Decimal('0.00000010'))=='1E-7' +assert m.fee_canonical(decimal.Decimal('12345678901234.123456'))=='12345678901234.123456' +assert m.fee_canonical(decimal.Decimal('-0.000'))=='0' +queries=[] +with patch.object(m,'inspect',side_effect=lambda name:{'Name':'/'+name}),patch.object(m,'require_test'),patch.object(m,'read_sql',side_effect=lambda c,sql:queries.append((c['Name'],sql)) or []): + assert m.producer(s,'FEE_VOID') is None + m.fee_facts(s,'00000000-0000-4000-8000-000000000002') +assert [r[0] for r in queries]==['/firstleg-test-lastleg','/firstleg-test-lastleg','/firstleg-test-finance','/firstleg-test-finance'] +assert 'BusinessFeeSourceChanged:LASTLEG:OVERSEAS_WAREHOUSE_FEE:123:1:VOIDED' in queries[0][1] +assert 'void_status' not in queries[1][1] and 'source_status' in queries[1][1] +assert all(t in queries[2][1] for t in ['fin_fee_source','fin_bill_line','fin_settlement_line','fin_cost','operation_version']) +assert 'fin_inbox_event' in queries[3][1] and 'result_resource_id' in queries[3][1] +publication={'entryId':'1','eventId':'00000000-0000-4000-8000-000000000001','publishStatus':'PENDING','payloadHash':'sha256:'+'a'*64} +facts={'lastleg':[{'sourceVersion':'1','sourceType':s['sourceType'],'customerId':'7','sourceStatus':'ACTIVE','reviewStatus':'APPROVED','publicationStatus':'PUBLISHED','direction':'RECEIVABLE'}], + 'finance':{'source':[{'id':'600','sourceFeeVersion':'1','sourceType':s['sourceType'],'customerId':'7','status':'AVAILABLE','walletHoldId':None}], + 'bill':[],'settlement':[],'cost':[],'inbox':[{'status':'PROCESSED','payloadHash':publication['payloadHash'],'resultResourceId':'600'}]}} +with patch.object(m,'fee_event',side_effect=lambda scope,action:None if action=='VOIDED' else publication),patch.object(m,'fee_facts',return_value=facts),patch.object(m,'inspect'),patch.object(m,'read_sql',return_value=[{'eventId':publication['eventId'],'outcome':'APPLIED'}]): + assert m.fee_prerequisite(s,wait=True)['ready'] is False + publication['publishStatus']='PUBLISHED';assert m.fee_prerequisite(s,wait=True)['ready'] is True + publication['publishStatus']='DEAD' + try:m.fee_prerequisite(s,wait=True);assert False + except RuntimeError as e:assert str(e)=='FEE_PUBLICATION_NOT_READY' +original={'Name':'/firstleg-test-lastleg','Id':'original','Image':'sha256:'+'a'*64,'Config':{'Image':'lastleg:test','Env':['FINANCE_SERVICE_URL=http://finance:9203','FIRSTLEG_SERVICE_URL=http://firstleg:9202'],'Labels':{'com.docker.compose.project':'firstleg-test'}},'HostConfig':{},'Mounts':[], + 'NetworkSettings':{'Networks':{'firstleg-test-network':{'IPAddress':'172.20.0.3','Aliases':['lastleg-service']}}},'State':{'Running':True,'Health':{'Status':'healthy'}}} +finance=copy.deepcopy(original);finance['Name']='/firstleg-test-finance';finance['NetworkSettings']['Networks']['firstleg-test-network']['IPAddress']='172.20.0.9' +with patch.object(m,'fee_prerequisite',return_value={'ready':True}),patch.object(m,'inspect',side_effect=lambda name:finance if name=='firstleg-test-finance' else original),patch.object(m.Path,'mkdir'),patch.object(m.Path,'write_bytes'),patch('builtins.open',mock_open()),patch.object(m.socket,'socket') as sock,patch.object(m.subprocess,'Popen') as process,patch.object(m,'private_write'): + sock.return_value.getsockname.return_value=('127.0.0.1',18001) + m.prepare(s,pathlib.Path('/offline'),{'direction':'FEE_VOID','proxySource':''}) + args=process.call_args.args[0] + assert args[args.index('--upstream')+1]=='http://172.20.0.9:9203/internal/finance/v1/business-fee-source-events' + assert args[args.index('--fee-source-id')+1]=='123' and args[args.index('--source-version')+1]=='1' + assert '--delivery-id' not in args and '--container-booking-id' not in args +state={'direction':'FEE_VOID','restored':False,'prepared':True,'originalId':'original','image':'lastleg:test','originalImageId':original['Image'],'originalFingerprint':m.fingerprint(original),'originalBody':m.recreation(original,m.env_of(original),full=True),'proxyPort':18001} +with patch.object(m,'control',return_value={}),patch.object(m,'inspect',return_value=original),patch.object(m,'private_write'),patch.object(m,'replace') as replace: + m.activate(s,pathlib.Path('/offline'),copy.deepcopy(state)) + changed=dict(item.split('=',1) for item in replace.call_args.args[1]['Env']) + assert changed['FINANCE_SERVICE_URL']=='http://host.docker.internal:18001' + assert changed['FIRSTLEG_SERVICE_URL']=='http://firstleg:9202' and changed['EVENT_MAX_ATTEMPTS']=='12' +row={'entryId':'2','eventId':'00000000-0000-4000-8000-000000000002','payloadHash':'sha256:'+'b'*64,'publishStatus':'DEAD','failedAttempts':12} +retry_state={**state,'operationsRequestId':'00000000-0000-4000-8000-000000000003'} +runtime=copy.deepcopy(original);runtime['Config']['Env'].append('EVENT_OPERATIONS_TOKEN=offline') +dead={'lastleg':[{'sourceVersion':'1','sourceStatus':'ACTIVE'}],'finance':{'source':[{'sourceFeeVersion':'1','status':'VOIDED'}]}} +with patch.object(m,'producer',return_value=row),patch.object(m,'control',return_value=row),patch.object(m,'fee_facts',return_value=dead),patch.object(m,'private_write'),patch.object(m,'inspect',return_value=runtime),patch.object(m.urllib.request,'urlopen',return_value=io.BytesIO(b'{"code":"200"}')) as http: + assert m.recover(s,pathlib.Path('/offline'),retry_state)['retrySubmitted'] + assert http.call_args.args[0].full_url=='http://172.20.0.3:9303/internal/lastleg/event-operations/2/retry' + assert retry_state['beforeRetry']['fee']['lastleg'][0]['sourceStatus']=='ACTIVE' +restored=copy.deepcopy(original);restored['Id']='restored' +with patch.object(m,'control',side_effect=OSError('stopped')),patch.object(m,'inspect',return_value=restored),patch.object(m,'private_write'),patch.object(m,'replace') as replace: + m.restore(pathlib.Path('/offline'),state);assert not state['prepared'];replace.assert_not_called() +state['prepared']=True;restored['Config']['Env'].append('UNEXPECTED=1') +with patch.object(m,'control'),patch.object(m,'inspect',return_value=restored): + try:m.restore(pathlib.Path('/offline'),state);assert False + except RuntimeError as e:assert str(e)=='CONCURRENT_PRODUCER_CHANGE' +`],{cwd:new URL('..',import.meta.url),encoding:'utf8',timeout:10000}) + assert.equal(result.status,0,result.stderr||result.error?.message) +})