test(fleet): journal prerequisite writes and stop on unknown results

2026-10-03 19:07:58 +08:00
parent 499354ffd2
commit 3bf2612470
2 changed files with 150 additions and 2 deletions
@@ -23,6 +23,63 @@ const id = value => value == null ? '' : String(value)
 const success = response => response?.http === 200 && id(response?.body?.code) === '200'
 const key = () => `fleet-${randomUUID()}`

+/** Install before login/prerequisites. A caught unknown write must not permit another request. */
+export function guardFleetWrites(context, checkpoint) {
+  let stopped = null, sequence = 0
+  for (const name of ['request', 'requestWithBody', 'requestBinary']) {
+    const original = context[name]
+    if (typeof original !== 'function') continue
+    context[name] = async function (method, path, body, token, headers) {
+      if (stopped) throw stopped
+      if (['GET', 'HEAD', 'OPTIONS'].includes(String(method).toUpperCase()))
+        return original.apply(this, arguments)
+      const entry = { sequence: ++sequence, method, path, status: 'SENDING', startedAt: new Date().toISOString(),
+        headers, body }
+      // Multipart delegates to requestWithBody, so it is journaled exactly once, after construction.
+      if (body instanceof FormData) {
+        entry.body = { multipart: [] }
+        for (const [field, value] of body.entries()) entry.body.multipart.push(typeof value === 'string'
+          ? { field, value } : { field, name: value.name, type: value.type,
+            base64: Buffer.from(await value.arrayBuffer()).toString('base64') })
+      }
+      try { await checkpoint(entry) } catch (error) { stopped = error; throw error }
+      try {
+        const response = await original.apply(this, arguments)
+        const http = response?.http, payload = response?.body, code = id(payload?.code)
+        const definite = Number.isInteger(http) && http >= 200 && http < 500
+          && payload && typeof payload === 'object' && !Array.isArray(payload) && response.requestId && response.traceId
+          && (http === 200 && code === '200' || /^[A-Z]\d{4}$/.test(code) && (http === 200 || http >= 400))
+        Object.assign(entry, { status: definite ? 'RESPONDED' : 'UNKNOWN_RESULT', http, code,
+          requestId: response?.requestId, traceId: response?.traceId, finishedAt: new Date().toISOString() })
+        if (!definite) stopped = Error(`Unknown write outcome: ${method} ${path}; preserve original request, do not resend`)
+        await checkpoint(entry)
+        if (stopped) throw stopped
+        return response
+      } catch (error) {
+        stopped ||= error
+        if (entry.status === 'SENDING') {
+          entry.status = 'UNKNOWN_RESULT'; entry.finishedAt = new Date().toISOString()
+          await checkpoint(entry)
+        }
+        throw stopped
+      }
+    }
+  }
+}
+
+export function redactFleetWrite(value, secrets = []) {
+  const sensitive = name => name !== 'draftToken' && /password|authorization|cookie|privateKey|token$|^(?:loginId|username)$/i.test(name)
+  if (Array.isArray(value)) return value.map(item => redactFleetWrite(item, secrets))
+  if (value && typeof value === 'object' && typeof value.field === 'string' && sensitive(value.field))
+    return { field: value.field, value: '[REDACTED]' }
+  if (value && typeof value === 'object') return Object.fromEntries(Object.entries(value).map(([name, item]) => [name,
+    sensitive(name)
+      ? '[REDACTED]' : redactFleetWrite(item, secrets)]))
+  if (typeof value === 'string') return secrets.filter(secret => typeof secret === 'string' && secret.length)
+    .reduce((text, secret) => text.split(secret).join('[REDACTED]'), value)
+  return value
+}
+
 export function parseFleetArguments(args) {
  const config = { execute: false, environment: 'TEST', runId: 'plan', scenario: 'handover' }
  let action = null
@@ -498,6 +555,14 @@ export async function runFleetAcceptance(config, { env = process.env, readers, c
      pdaUsername: env.ADMIN_PDA_USERNAME, pdaPassword: env.ADMIN_PDA_PASSWORD, merchantUsername: '', merchantPassword: '',
      markerNamespace: createHash('sha256').update(config.runId).digest('hex').slice(0, 12) })
    context = createContext ? createContext(options) : new AcceptanceRun(options)
+    let writeCheckpoint = 0
+    guardFleetWrites(context, entry => {
+      const payload = JSON.stringify(redactFleetWrite({ marker: context.marker, runId: config.runId, request: entry },
+        [...secrets, context.token, context.pdaToken, context.refreshToken]), null, 2) + '\n'
+      const descriptor = openSync(join(output, 'evidence', `write-${String(++writeCheckpoint).padStart(6, '0')}.json`), 'wx')
+      try { writeFileSync(descriptor, payload); fsyncSync(descriptor) }
+      finally { closeSync(descriptor) }
+    })
    context.context.fleetRunId = config.runId
    context.context.fleetScenario = config.scenario || 'handover'
    context.context.saveFleetHandoverFixture = fixture => save('handover-fixture.json', fixture)
@@ -3,7 +3,66 @@ import assert from 'node:assert/strict'
 import { mkdtempSync, mkdirSync, readFileSync, existsSync, rmSync } from 'node:fs'
 import { tmpdir } from 'node:os'
 import { join, dirname, resolve } from 'node:path'
-import { parseFleetArguments, fleetExecutionPlan, runFleetAcceptance, reviewedFullCheckpoint, verifyFleetTerminalState, runFleetHandoverStage, fleetCostShares, prepareFleetCostActors } from './fleet-test-env-acceptance.mjs'
+import { parseFleetArguments, fleetExecutionPlan, runFleetAcceptance, reviewedFullCheckpoint, verifyFleetTerminalState, runFleetHandoverStage, fleetCostShares, prepareFleetCostActors, guardFleetWrites, redactFleetWrite } from './fleet-test-env-acceptance.mjs'
+
+for (const response of [
+  { http: 0 }, { http: 503, body: { code: 'A0500' }, requestId: 'r', traceId: 't' },
+  { http: 200, body: null }, { http: 200, body: { code: '200' } },
+  { http: 302, body: { code: '200' }, requestId: 'r', traceId: 't' },
+]) test(`unknown prerequisite write stops every subsequent request: ${JSON.stringify(response)}`, async () => {
+  const saved = [], calls = []
+  const ctx = { async request(method, path, body) {
+    assert.equal(saved.at(-1).status, 'SENDING')
+    assert.deepEqual(saved.at(-1).body, body)
+    calls.push(path); return response
+  } }
+  guardFleetWrites(ctx, entry => saved.push(structuredClone(entry)))
+  const body = { expectedVersion: 6, idempotencyKey: 'original-key', internalNotes: ['test remark'] }
+  await assert.rejects(ctx.request('PUT', '/orders/1/remarks?tenantId=2', body), /Unknown write outcome/)
+  body.expectedVersion = 99
+  assert.equal(saved[0].body.expectedVersion, 6)
+  assert.equal(saved[1].status, 'UNKNOWN_RESULT')
+  for (const method of ['POST', 'GET', 'DELETE']) await assert.rejects(ctx.request(method, '/must-not-run'), /Unknown write outcome/)
+  assert.deepEqual(calls, ['/orders/1/remarks?tenantId=2'])
+})
+
+test('formal correlated rejection is recorded and remains available for expected negative assertions', async () => {
+  const saved = [], response = { http: 200, body: { code: 'A0409' }, requestId: 'r', traceId: 't' }
+  const ctx = { async request() { return response } }
+  guardFleetWrites(ctx, entry => saved.push(structuredClone(entry)))
+  assert.equal(await ctx.request('POST', '/negative', { recordVersion: 7 }), response)
+  assert.equal(saved[1].status, 'RESPONDED')
+  await ctx.request('GET', '/readback'); assert.equal(saved.length, 2)
+})
+
+test('checkpoint failure stops before dispatch; transport throw stops even after caller catches', async () => {
+  let calls = 0
+  const ctx = { async request() { calls++; throw Error('transport failure') } }
+  guardFleetWrites(ctx, () => { throw Error('disk full') })
+  await assert.rejects(ctx.request('POST', '/write'), /disk full/)
+  await assert.rejects(ctx.request('GET', '/later'), /disk full/); assert.equal(calls, 0)
+  const saved = [], other = { async request() { calls++; throw Error('transport failure') } }
+  guardFleetWrites(other, entry => saved.push(structuredClone(entry)))
+  await assert.rejects(other.request('PUT', '/write'), /transport failure/)
+  await assert.rejects(other.request('POST', '/cleanup'), /transport failure/)
+  assert.equal(calls, 1); assert.equal(saved[1].status, 'UNKNOWN_RESULT')
+})
+
+test('multipart delegates once and retains recoverable fixture bytes, business keys and versions without credentials', async () => {
+  const saved = [], ctx = {
+    async requestMultipart(method, path, form) { return this.requestWithBody(method, path, form) },
+    async requestWithBody() { return { http: 200, body: { code: '200' }, requestId: 'r', traceId: 't' } },
+  }
+  guardFleetWrites(ctx, entry => saved.push(redactFleetWrite(structuredClone(entry))))
+  const form = new FormData(); form.append('file', new Blob(['fixture']), 'proof.txt'); form.append('password', 'hidden')
+  await ctx.requestMultipart('POST', '/upload', form)
+  assert.equal(saved.length, 2); assert.equal(Buffer.from(saved[0].body.multipart[0].base64, 'base64').toString(), 'fixture')
+  assert.ok(!JSON.stringify(saved).includes('hidden'))
+  const clean = redactFleetWrite({ draftToken: 'business-draft', idempotencyKey: 'command-key', expectedVersion: 6,
+    headers: { Authorization: 'Bearer hidden', 'X-Access-Token': 'hidden' }, body: { password: 'hidden', accessToken: 'hidden' } })
+  assert.equal(clean.draftToken, 'business-draft'); assert.equal(clean.idempotencyKey, 'command-key')
+  assert.equal(clean.expectedVersion, 6); assert.ok(!JSON.stringify(clean).includes('hidden'))
+})

 test('fleet entry defaults to a no-side-effect plan and requires explicit TEST and unique run ID', () => {
  const p = parseFleetArguments([])
@@ -118,12 +177,24 @@ async function executeFixture(failProducer = false) {
    CHONGCHONG_ADMIN_PASSWORD: secrets, ADMIN_PDA_USERNAME: 'fixture-pda', ADMIN_PDA_PASSWORD: secrets }
  const config = parseFleetArguments(['--execute', '--environment', 'TEST', '--run-id', 'fixture'])
  const createContext = options => ({ options, context: {}, createdResources: {}, steps: [], modules: [], coverageGaps: [], token: secrets,
+    async request(method, path, body) {
+      const pending = JSON.parse(readFileSync(join(root, 'reports/fleet-fixture/evidence/write-000001.json'), 'utf8'))
+      assert.equal(pending.request.status, 'SENDING'); assert.deepEqual(pending.request.body, body)
+      assert.equal(pending.request.path, path); calls.push('unknown-write'); return { http: 0 }
+    },
    async executeModule(name, handler) {
      try { await handler(); this.modules.push({ name, success: true }) }
      catch { this.modules.push({ name, success: false }); this.steps.push({ name: 'producer failed', passed: false }) }
    },
    async run() {
-      await this.executeModule('producer', async () => { calls.push('producer'); if (failProducer) throw Error('fixture failure') })
+      await this.executeModule('producer', async () => {
+        calls.push('producer')
+        if (failProducer === 'unknown') {
+          try { await this.request('PUT', '/orders/1/remarks?tenantId=2', { expectedVersion: 6, idempotencyKey: 'original-key', internalNotes: ['fixture'] }) } catch { /* Simulate a prerequisite swallowing an error. */ }
+          await this.request('POST', '/loading/must-not-run', {})
+        }
+        if (failProducer) throw Error('fixture failure')
+      })
      await this.executeModule('phase4-overseas-dispatch', async () => { calls.push('dispatch'); this.createdResources.current = { accessToken: secrets } })
      return { success: true, steps: this.steps, modules: this.modules, createdResources: this.createdResources, summary: { failed: 0 } }
    },
@@ -145,6 +216,11 @@ async function executeFixture(failProducer = false) {
      } }
    const result = await runFleetAcceptance(config, options)
    const report = JSON.parse(readFileSync(result.report, 'utf8'))
+    if (failProducer === 'unknown') {
+      const unknown = JSON.parse(readFileSync(join(root, 'reports/fleet-fixture/evidence/write-000002.json'), 'utf8'))
+      assert.equal(unknown.request.status, 'UNKNOWN_RESULT'); assert.equal(unknown.request.body.expectedVersion, 6)
+      assert.equal(unknown.request.body.idempotencyKey, 'original-key')
+    }
    assert.ok(!readFileSync(result.report, 'utf8').includes(secrets))
    assert.equal(existsSync(join(root, 'reports/.fleet-acceptance.lock')), false)
    const oldReport = readFileSync(result.report, 'utf8')
@@ -174,6 +250,13 @@ test('failed prerequisite stops before further fixture creation and retains the
  assert.equal(report.fatal, 'Producer stopped: producer')
 })

+test('real fleet entry journals prerequisite writes before sending and stops loading even if producer catches unknown outcome', async () => {
+  const { result, report, calls } = await executeFixture('unknown')
+  assert.equal(result.success, false)
+  assert.deepEqual(calls, ['preflight', 'producer', 'unknown-write'])
+  assert.equal(report.fatal, 'Producer stopped: producer')
+})
+
 test('environment mismatch rejects before invoking any readers or contexts', async () => {
  let touched = false
  await assert.rejects(() => runFleetAcceptance({ execute: true, environment: 'TEST', runId: 'new' }, {