java-project/baiyue-firstleg
4
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

test: isolate fee void ACK loss and preserve wallet delivery

Browse Source
This commit is contained in:
ouyangxq
2026-10-03 08:54:52 +08:00
parent 933dd4cdef
commit 912e69aeb2
2 changed files with 84 additions and 11 deletions
Download Patch File Download Diff File Expand all files Collapse all files
scripts/event-delivery-fault-proxy.py
+28 -5
View File
@@ -32,12 +32,15 @@ def encoded(value):
class State:
def __init__(self, event_type, event_id, loading_plan_id, log_path=None, container_booking_id=None, delivery_id=None):
def __init__(self, event_type, event_id, loading_plan_id, log_path=None, container_booking_id=None, delivery_id=None,
fee_source_id=None, source_version=None):
self.event_type = event_type
self.selector_id = event_id
self.plan_id = loading_plan_id
self.booking_id = container_booking_id
self.delivery_id = delivery_id
self.fee_source_id = fee_source_id
self.source_version = source_version
self.log_path = log_path
self.lock = threading.Lock()
self.mode = "PASS"
@@ -73,6 +76,9 @@ class State:
selected = isinstance(payload, dict) and str(payload.get("sourceContainerBookingId")) == self.booking_id
elif self.delivery_id:
selected = isinstance(payload, dict) and str(payload.get("deliveryOrderId")) == self.delivery_id
elif self.fee_source_id:
selected = (isinstance(payload, dict) and str(payload.get("sourceFeeId")) == self.fee_source_id
and str(payload.get("sourceVersion")) == self.source_version and payload.get("action") == "VOIDED")
else:
selected = isinstance(payload, dict) and str(payload.get("loadingPlanId")) == self.plan_id
qualifies = event.get("eventType") == self.event_type and selected
@@ -146,16 +152,22 @@ class QuietHandler(BaseHTTPRequestHandler):
def data_handler(state, upstream, timeout):
parts = urllib.parse.urlsplit(upstream)
endpoint = urllib.parse.urlunsplit(("", "", parts.path or "/", parts.query, ""))
# Both consumers share FINANCE_SERVICE_URL. Preserve wallet delivery while
# targeting one fee event; never select wallet bodies for fault injection.
wallet_endpoint = "/internal/finance/v1/wallet-command-events"
wallet_passthrough = (state.fee_source_id is not None
and endpoint == "/internal/finance/v1/business-fee-source-events")
class DataHandler(QuietHandler):
def do_POST(self):
if self.path != endpoint:
passthrough = wallet_passthrough and self.path == wallet_endpoint
if self.path != endpoint and not passthrough:
self.safe_error(404, "UNEXPECTED_EVENT_ENDPOINT")
return
body = self.read_body()
if body is None:
return
identity, matched, error_code, attempt, mode = state.select(body)
identity, matched, error_code, attempt, mode = (None, False, None, 0, "PASS") if passthrough else state.select(body)
if error_code:
state.record(identity, matched, attempt, None, error_code)
self.safe_error(409, error_code)
@@ -164,7 +176,8 @@ def data_handler(state, upstream, timeout):
connection_headers = set(self.headers.get("Connection", "").lower().replace(" ", "").split(","))
headers = {name: value for name, value in self.headers.items()
if name.lower() not in HOP_HEADERS | connection_headers}
request = urllib.request.Request(upstream, data=body, headers=headers, method="POST")
destination = urllib.parse.urlunsplit((parts.scheme, parts.netloc, wallet_endpoint, "", "")) if passthrough else upstream
request = urllib.request.Request(destination, data=body, headers=headers, method="POST")
try:
# Redirects are forbidden: a redirect could expose a service identity token.
class NoRedirect(urllib.request.HTTPRedirectHandler):
@@ -264,6 +277,8 @@ def main():
selector.add_argument("--loading-plan-id")
selector.add_argument("--container-booking-id")
selector.add_argument("--delivery-id")
selector.add_argument("--fee-source-id")
parser.add_argument("--source-version")
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)
@@ -284,9 +299,17 @@ def main():
parser.error("container-booking-id must be a positive decimal identifier")
if args.delivery_id and not re.fullmatch(r"[1-9][0-9]{0,19}", args.delivery_id):
parser.error("delivery-id must be a positive decimal identifier")
if bool(args.fee_source_id) != bool(args.source_version):
parser.error("fee-source-id and source-version are required together")
if args.fee_source_id and (not re.fullmatch(r"[1-9][0-9]{0,19}", args.fee_source_id)
or not re.fullmatch(r"[1-9][0-9]{0,9}", args.source_version)
or int(args.source_version) > 2147483647
or args.event_type != "BusinessFeeSourceChanged"):
parser.error("fee void requires valid source identity, version and BusinessFeeSourceChanged")
if not re.fullmatch(r"[A-Za-z][A-Za-z0-9]{0,79}", args.event_type) or not 0 < args.timeout <= 120:
parser.error("invalid event type or timeout")
state = State(args.event_type, args.event_id, args.loading_plan_id, args.log, args.container_booking_id, args.delivery_id)
state = State(args.event_type, args.event_id, args.loading_plan_id, args.log, args.container_booking_id, args.delivery_id,
args.fee_source_id, args.source_version)
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):
scripts/event-delivery-fault-proxy.test.mjs
+56 -6
View File
@@ -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'], eventType = 'LoadingCompleted') {
async function fixture(selector = ['--loading-plan-id', '2100000000000000001'], eventType = 'LoadingCompleted', endpoint = '/internal/events') {
const calls = []
const applied = new Set()
let invalidAck = false
@@ -28,7 +28,7 @@ async function fixture(selector = ['--loading-plan-id', '2100000000000000001'],
for await (const chunk of request) chunks.push(chunk)
const raw = Buffer.concat(chunks).toString()
const envelope = JSON.parse(raw)
calls.push({ raw, authorization: request.headers.authorization, eventId: envelope.eventId })
calls.push({ raw, authorization: request.headers.authorization, eventId: envelope.eventId, path: request.url })
const outcome = applied.has(envelope.eventId) ? 'DUPLICATE' : 'APPLIED'
applied.add(envelope.eventId)
const body = JSON.stringify({ code: '200', data: { eventId: invalidAck ? otherId : envelope.eventId,
@@ -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', eventType,
`http://127.0.0.1:${consumer.address().port}${endpoint}`, '--event-type', eventType,
...selector, '--log', log], { env: { ...process.env, EVENT_FAULT_CONTROL_TOKEN: controlToken },
stdio: ['ignore', 'pipe', 'pipe'] })
let output = ''; let errors = ''
@@ -58,12 +58,12 @@ async function fixture(selector = ['--loading-plan-id', '2100000000000000001'],
child.once('error', error => { clearTimeout(timer); reject(error) })
child.once('exit', code => { if (!output) { clearTimeout(timer); reject(new Error(`Proxy exit ${code}`)) } })
})
const dataUrl = `http://127.0.0.1:${ready.dataPort}/internal/events`
const dataUrl = `http://127.0.0.1:${ready.dataPort}${endpoint}`
const controlUrl = `http://127.0.0.1:${ready.controlPort}`
return { calls, applied, log, invalidAck: value => { invalidAck = value },
send: async envelope => {
send: async (envelope, path = endpoint) => {
const sent = JSON.stringify(envelope)
const response = await fetch(dataUrl, { method: 'POST', body: sent,
const response = await fetch(new URL(path, dataUrl), { method: 'POST', body: sent,
headers: { 'Content-Type': 'application/json', Authorization: `Bearer ${secret}` } })
return { status: response.status, raw: await response.text(), headers: response.headers, sent }
},
@@ -85,6 +85,26 @@ async function fixture(selector = ['--loading-plan-id', '2100000000000000001'],
}
}
test('fee fault preserves wallet endpoint bytes and auth without selecting its body', async t => {
const proxy = await fixture(['--fee-source-id', '2100000000000000001', '--source-version', '1'],
'BusinessFeeSourceChanged', '/internal/finance/v1/business-fee-source-events')
t.after(() => proxy.stop())
await proxy.control('/mode', { mode: 'FAULT' })
const envelope = { ...event(), eventType: 'BusinessFeeSourceChanged',
payload: { sourceFeeId: '2100000000000000001', sourceVersion: 1, action: 'VOIDED' } }
const walletPath = '/internal/finance/v1/wallet-command-events'
const wallet = await proxy.send(envelope, walletPath)
assert.equal(wallet.status, 200)
assert.equal(proxy.calls[0].path, walletPath)
assert.equal(proxy.calls[0].raw, wallet.sent)
assert.equal(proxy.calls[0].authorization, `Bearer ${secret}`)
assert.equal((await (await proxy.control('/status')).json()).attempt, 0)
assert.equal((await proxy.send(envelope, '/internal/finance/v1/unknown')).status, 404)
assert.equal(proxy.calls.length, 1)
assert.equal((await proxy.send(envelope)).status, 503)
assert.equal((await (await proxy.control('/status')).json()).attempt, 1)
})
test('ACK-loss proxy only targets one event and forwards committed upstream results before dropping ACK', async t => {
const proxy = await fixture()
t.after(() => proxy.stop())
@@ -260,3 +280,33 @@ test('delivery selection rejects malformed IDs or a second selector before openi
assert.ok(!result.stdout.includes('"ready"'))
}
})
test('fee void selector leaves publication, other sources and versions unchanged while recovering exact VOIDED ACK', async t => {
const sourceFeeId='2100000000000000011'
const proxy=await fixture(['--fee-source-id',sourceFeeId,'--source-version','2'],'BusinessFeeSourceChanged')
t.after(()=>proxy.stop())
await proxy.control('/mode',{mode:'FAULT'})
const voided={...event(),eventType:'BusinessFeeSourceChanged',payload:{sourceFeeId,sourceVersion:2,action:'VOIDED',totalAmount:30}}
for(const payload of [{...voided.payload,action:'PUBLISHED'},{...voided.payload,sourceVersion:1},{...voided.payload,sourceFeeId:'2100000000000000012'},
{...voided.payload,action:'VOID'}]) assert.equal((await proxy.send({...voided,eventId:otherId,payload})).status,200)
assert.equal((await proxy.send(voided)).status,503)
assert.equal((await proxy.send(voided)).status,503)
assert.equal((await proxy.send({...voided,eventId:samePlanOtherId})).status,200)
const before=proxy.calls.length
assert.equal((await proxy.send({...voided,payload:{...voided.payload,sourceVersion:3}})).status,409)
assert.equal(proxy.calls.length,before)
await proxy.control('/mode',{mode:'PASS'})
assert.equal(JSON.parse((await proxy.send(voided)).raw).data.outcome,'DUPLICATE')
const state=await (await proxy.control('/status')).json()
assert.deepEqual(state.records.filter(r=>r.errorCode!=='TARGET_EVENT_HASH_CONFLICT').map(r=>r.outcome),['APPLIED','DUPLICATE','DUPLICATE'])
})
test('fee void selector requires a bounded version paired with only its own source selector',()=>{
for(const args of [['--fee-source-id','11'],['--event-id',targetId,'--source-version','1'],['--fee-source-id','0','--source-version','1'],
['--fee-source-id','11','--source-version','0'],['--fee-source-id','11','--source-version','2147483648'],
['--fee-source-id','../11','--source-version','1'],['--fee-source-id','11','--source-version','1','--delivery-id','88']]) {
const result=spawnSync(process.env.PYTHON||'python',[script,'--upstream','http://127.0.0.1:1/internal/events','--event-type','BusinessFeeSourceChanged',...args],
{encoding:'utf8',timeout:5000,env:{...process.env,EVENT_FAULT_CONTROL_TOKEN:controlToken}})
assert.equal(result.status,2);assert.ok(!result.stdout.includes('"ready"'))
}
})