604 lines
39 KiB
Python
604 lines
39 KiB
Python
"""Internal TEST executor. Receives one bounded task action from Jenkins.
|
|
|
|
Never edits business SQL. Docker configuration is captured before the scoped
|
|
fault and restored to the same image, mounts, ports and network identity.
|
|
Only the original DEAD event can be retried through its protected HTTP API.
|
|
"""
|
|
import base64
|
|
import copy
|
|
import decimal
|
|
import hashlib
|
|
import http.client
|
|
import json
|
|
import os
|
|
import re
|
|
import secrets
|
|
import signal
|
|
import socket
|
|
import subprocess
|
|
import sys
|
|
import tempfile
|
|
import time
|
|
import urllib.parse
|
|
import urllib.request
|
|
import uuid
|
|
from pathlib import Path
|
|
|
|
|
|
class DockerConnection(http.client.HTTPConnection):
|
|
def connect(self):
|
|
self.sock = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
|
|
self.sock.settimeout(self.timeout)
|
|
self.sock.connect('/var/run/docker.sock')
|
|
|
|
|
|
def docker(method, path, body=None):
|
|
conn = DockerConnection('localhost', timeout=20)
|
|
data = None if body is None else json.dumps(body).encode()
|
|
conn.request(method, path, data, {'Content-Type': 'application/json'})
|
|
response = conn.getresponse()
|
|
raw = response.read()
|
|
conn.close()
|
|
if response.status >= 300:
|
|
raise RuntimeError('DOCKER_' + str(response.status))
|
|
return json.loads(raw) if raw else None
|
|
|
|
|
|
def inspect(name):
|
|
return docker('GET', '/containers/' + urllib.parse.quote(name, safe='') + '/json')
|
|
|
|
|
|
def env_of(container):
|
|
return dict(item.split('=', 1) for item in container['Config']['Env'])
|
|
|
|
|
|
def require_test(container, name):
|
|
if container['Name'] != '/' + name or container['Config']['Labels'].get('com.docker.compose.project') != 'firstleg-test':
|
|
raise RuntimeError('NOT_TEST_CONTAINER')
|
|
|
|
|
|
def recreation(container, environment, full=False):
|
|
body = copy.deepcopy(container['Config'])
|
|
body['Env'] = [key + '=' + value for key, value in environment.items()]
|
|
body['HostConfig'] = copy.deepcopy(container['HostConfig'])
|
|
if body['HostConfig'].get('AutoRemove') or any(mount['Type'] not in ('bind', 'volume') for mount in container['Mounts']):
|
|
raise RuntimeError('UNSUPPORTED_CONTAINER_STORAGE')
|
|
# Preserve named volumes explicitly, never create a replacement anonymous volume.
|
|
if any(mount['Type'] == 'volume' and not mount.get('Name') for mount in container['Mounts']):
|
|
raise RuntimeError('UNNAMED_VOLUME')
|
|
if body['HostConfig'].get('VolumesFrom'):
|
|
raise RuntimeError('UNSUPPORTED_VOLUME_REFERENCE')
|
|
endpoints = {}
|
|
for name, network in container['NetworkSettings']['Networks'].items():
|
|
aliases = [alias for alias in network.get('Aliases') or [] if alias != container['Id'][:12]]
|
|
endpoints[name] = {'Aliases': aliases, 'IPAMConfig': {'IPv4Address': network['IPAddress']}}
|
|
if full:
|
|
ipam = copy.deepcopy(network.get('IPAMConfig') or {})
|
|
ipam['IPv4Address'] = network['IPAddress']
|
|
if network.get('GlobalIPv6Address'):
|
|
ipam['IPv6Address'] = network['GlobalIPv6Address']
|
|
endpoints[name]['IPAMConfig'] = ipam
|
|
for key in ('Links', 'DriverOpts', 'GwPriority', 'MacAddress'):
|
|
if network.get(key) is not None:
|
|
endpoints[name][key] = copy.deepcopy(network[key])
|
|
body['NetworkingConfig'] = {'EndpointsConfig': endpoints}
|
|
return body
|
|
|
|
|
|
def replace(container, body):
|
|
name = container['Name'].lstrip('/')
|
|
current = inspect(name)
|
|
if current['Id'] != container['Id']:
|
|
raise RuntimeError('CONCURRENT_CONTAINER_CHANGE')
|
|
docker('POST', '/containers/' + current['Id'] + '/stop?t=10')
|
|
docker('DELETE', '/containers/' + current['Id'] + '?v=false')
|
|
created = docker('POST', '/containers/create?name=' + name, body)
|
|
docker('POST', '/containers/' + created['Id'] + '/start')
|
|
return created['Id']
|
|
|
|
|
|
def private_write(path, value):
|
|
fd = os.open(path, os.O_WRONLY | os.O_CREAT | os.O_TRUNC, 0o600)
|
|
with os.fdopen(fd, 'w') as file:
|
|
json.dump(value, file)
|
|
|
|
|
|
def scope_directory(scope, direction='LOADING'):
|
|
if direction not in ('LOADING', 'UNLOAD', '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', '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'
|
|
|
|
|
|
def delivery(state):
|
|
return state.get('direction') == 'DELIVERY'
|
|
|
|
|
|
def reverse_delivery(state):
|
|
return state.get('direction') in ('UNLOAD', 'DELIVERY', 'FEE_VOID')
|
|
|
|
|
|
def producer_name(state):
|
|
return 'firstleg-test-lastleg' if reverse_delivery(state) else 'firstleg-test-business'
|
|
|
|
|
|
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')
|
|
|
|
|
|
def fingerprint(container):
|
|
body = recreation(container, env_of(container), full=True)
|
|
# Docker may add the legacy Config.MacAddress after explicit per-network MAC
|
|
# restoration. It is redundant only when it equals an actual preserved MAC;
|
|
# every network MAC remains in the fingerprint and unrelated values still fail.
|
|
network_macs = {endpoint.get('MacAddress') for endpoint in body['NetworkingConfig']['EndpointsConfig'].values()}
|
|
if body.get('MacAddress') and body['MacAddress'] in network_macs:
|
|
body.pop('MacAddress')
|
|
body['Env'].sort()
|
|
for endpoint in body['NetworkingConfig']['EndpointsConfig'].values():
|
|
endpoint['Aliases'].sort()
|
|
body['ResolvedMounts'] = sorted(container['Mounts'], key=lambda mount: mount['Destination'])
|
|
body['NetworkIds'] = {name: network.get('NetworkID') for name, network in container['NetworkSettings']['Networks'].items()}
|
|
return hashlib.sha256(json.dumps(body, sort_keys=True, separators=(',', ':')).encode()).hexdigest()
|
|
|
|
|
|
def control(state, mode=None):
|
|
url = 'http://127.0.0.1:' + str(state['controlPort']) + ('/mode' if mode else '/status')
|
|
request = urllib.request.Request(url, None if mode is None else json.dumps({'mode': mode}).encode(),
|
|
{'X-Event-Fault-Control-Token': state['controlToken'], 'Content-Type': 'application/json'})
|
|
with urllib.request.urlopen(request, timeout=5) as response:
|
|
return json.load(response)
|
|
|
|
|
|
def read_sql(container, sql):
|
|
# SQL is generated here from fixed templates and decimal/UUID/hash identities.
|
|
if re.search(r'\b(INSERT|UPDATE|DELETE|ALTER|DROP|TRUNCATE|CALL|INTO|OUTFILE)\b', sql, re.I):
|
|
raise RuntimeError('NON_READ_ONLY_SQL')
|
|
e = env_of(container)
|
|
prefix = {'/firstleg-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;'
|
|
result = subprocess.run(['mysql', '--host=' + host, '--port=' + str(url.port or 3306),
|
|
'--user=' + e[prefix + '_DB_USERNAME'], '--batch', '--raw', '--skip-column-names', url.path.lstrip('/')],
|
|
input=transaction, text=True, capture_output=True, timeout=20,
|
|
env=dict(os.environ, MYSQL_PWD=e[prefix + '_DB_PASSWORD']))
|
|
if result.returncode:
|
|
code = re.search(r'ERROR (\d+)', result.stderr)
|
|
raise RuntimeError('MYSQL_' + (code.group(1) if code else 'UNKNOWN'))
|
|
return [json.loads(line) for line in result.stdout.splitlines() if line.strip()]
|
|
|
|
|
|
def producer(scope, direction='LOADING'):
|
|
if direction == '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'] + "';")
|
|
elif direction == 'UNLOAD':
|
|
rows = read_sql(inspect('firstleg-test-lastleg'),
|
|
"SELECT JSON_OBJECT('entryId',CAST(id AS CHAR),'eventId',JSON_UNQUOTE(JSON_EXTRACT(payload_json,'$.eventId')),'payloadHash',JSON_UNQUOTE(JSON_EXTRACT(payload_json,'$.payloadHash')),'eventType',event_type,'customerId',customer_id,'aggregateId',aggregate_id,'aggregateType',aggregate_type,'eventKey',event_key,'envelope',JSON_EXTRACT(payload_json,'$'),'publishStatus',status,'failedAttempts',retry_count) FROM ll_event_outbox WHERE customer_id='0' AND aggregate_id='"
|
|
+ scope['containerBookingId'] + "' AND event_type='OverseasContainerUnloaded' AND event_key='OverseasContainerUnloaded:" + scope['containerBookingId'] + "';")
|
|
else:
|
|
rows = read_sql(inspect('firstleg-test-business'),
|
|
"SELECT JSON_OBJECT('entryId',CAST(id AS CHAR),'eventId',JSON_UNQUOTE(JSON_EXTRACT(payload_json,'$.eventId')),'payloadHash',payload_hash,'eventType',event_type,'publishStatus',publish_status,'failedAttempts',retry_count) FROM fl_inbound_outbox WHERE tenant_id=" + scope['customerId']
|
|
+ " AND aggregate_id=" + scope['loadingPlanId'] + " AND event_type='FirstLegLoadingPlanCompleted';")
|
|
if len(rows) > 1:
|
|
raise RuntimeError('AMBIGUOUS_EVENT')
|
|
return rows[0] if rows else None
|
|
|
|
|
|
def container_facts(scope):
|
|
rows = read_sql(inspect('firstleg-test-business'),
|
|
"SELECT JSON_OBJECT('id',CAST(id AS CHAR),'containerNo',container_no,'version',CAST(version AS CHAR),'unloadingCompletedAt',CAST(unloading_completed_at AS CHAR),'unloadingPhotoUploadedAt',CAST(unloading_photo_uploaded_at AS CHAR),'estimatedEmptyReportedAt',CAST(estimated_empty_reported_at AS CHAR),'actualEmptyReportedAt',CAST(actual_empty_reported_at AS CHAR),'emptyReportSourceVersion',CAST(empty_report_source_version AS CHAR),'updatedAt',CAST(updated_at AS CHAR),'updatedBy',CAST(updated_by AS CHAR),'updatedByName',updated_by_name) FROM fl_container_booking WHERE id=" + scope['containerBookingId'] + ";")
|
|
if len(rows) != 1:
|
|
raise RuntimeError('CONTAINER_SCOPE_MISMATCH')
|
|
return rows[0]
|
|
|
|
|
|
def business_ids(scope):
|
|
rows = read_sql(inspect('firstleg-test-lastleg'), "SELECT JSON_OBJECT('id',CAST(id AS CHAR)) FROM ll_order WHERE customer_id='"
|
|
+ scope['customerId'] + "' AND source_container_booking_id='" + scope['containerBookingId'] + "' ORDER BY id;")
|
|
return [row['id'] for row in rows]
|
|
|
|
|
|
def 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)}
|
|
|
|
|
|
def delivery_facts(scope, event_id):
|
|
if not re.fullmatch(r'[0-9a-fA-F-]{36}|[0-9A-HJKMNP-TV-Z]{26}', event_id):
|
|
raise RuntimeError('INVALID_EVENT_IDENTITY')
|
|
customer, source, root = scope['customerId'], scope['sourceOrderId'], scope['rootOrderId']
|
|
firstleg = inspect('firstleg-test-business')
|
|
order = inspect('firstleg-test-order')
|
|
require_test(firstleg, 'firstleg-test-business')
|
|
require_test(order, 'firstleg-test-order')
|
|
# Each service is read in its own consistent read-only transaction; no cross-schema join.
|
|
inbound_sql = "tenant_id=" + customer + " AND deleted=0 AND order_id IN (" + source + "," + root + ")"
|
|
first_rows = read_sql(firstleg,
|
|
"SELECT JSON_OBJECT('kind','inbound','id',CAST(id AS CHAR),'customerId',CAST(tenant_id AS CHAR),'orderId',CAST(order_id AS CHAR),'orderNo',order_no,'inboundCode',inbound_code,'processStatus',process_status,'orderStatusSnapshot',order_status_snapshot,'version',version,'updatedAt',CAST(updated_at AS CHAR)) FROM fl_inbound_order WHERE " + inbound_sql + " ORDER BY id;\n"
|
|
+ "SELECT JSON_OBJECT('kind','inboundHistory','id',CAST(id AS CHAR),'inboundId',CAST(inbound_id AS CHAR),'action',action,'fromStatus',from_status,'toStatus',to_status,'occurredAt',CAST(occurred_at AS CHAR)) FROM fl_inbound_status_history WHERE tenant_id=" + customer + " AND inbound_id=" + scope['sourceInboundId'] + " ORDER BY id;\n"
|
|
+ "SELECT JSON_OBJECT('kind','inbox','eventId',event_id,'eventType',event_type,'aggregateId',aggregate_id,'aggregateVersion',aggregate_version,'payloadHash',payload_hash,'payload',JSON_EXTRACT(payload_json,'$'),'status',status,'receivedAt',CAST(received_at AS CHAR),'processedAt',CAST(processed_at AS CHAR)) FROM fl_container_event_inbox WHERE event_id='" + event_id + "';")
|
|
tree = "tenant_id=" + customer + " AND deleted=0 AND (id=" + root + " OR root_order_id=" + root + ")"
|
|
ids = "SELECT id FROM ord_order WHERE " + tree
|
|
order_rows = read_sql(order,
|
|
"SELECT JSON_OBJECT('kind','order','id',CAST(id AS CHAR),'customerId',CAST(tenant_id AS CHAR),'rootOrderId',CAST(root_order_id AS CHAR),'orderRole',order_role,'orderNo',order_no,'inboundCode',inbound_code,'status',status,'version',version,'completedAt',CAST(completed_at AS CHAR),'updatedAt',CAST(updated_at AS CHAR)) FROM ord_order WHERE " + tree + " ORDER BY id;\n"
|
|
+ "SELECT JSON_OBJECT('kind','orderHistory','id',CAST(id AS CHAR),'orderId',CAST(order_id AS CHAR),'action',action,'fromStatus',from_status,'toStatus',to_status,'occurredAt',CAST(occurred_at AS CHAR)) FROM ord_order_status_history WHERE tenant_id=" + customer + " AND order_id IN (" + ids + ") ORDER BY id;\n"
|
|
+ "SELECT JSON_OBJECT('kind','completionEvent','id',CAST(id AS CHAR),'orderId',CAST(aggregate_id AS CHAR),'eventType',event_type,'status',publish_status,'payload',JSON_EXTRACT(payload_json,'$')) FROM ord_outbox_event WHERE tenant_id=" + customer + " AND aggregate_id IN (" + ids + ") AND event_type='FIRST_LEG_ORDER_FULFILLMENT_COMPLETED' ORDER BY id;")
|
|
return {kind: [row for row in first_rows + order_rows if row['kind'] == kind] for kind in
|
|
('inbound', 'inboundHistory', 'inbox', 'order', 'orderHistory', 'completionEvent')}
|
|
|
|
|
|
def prepare(scope, directory, config):
|
|
if delivery(config) and producer(scope, 'DELIVERY') is not None:
|
|
raise RuntimeError('DELIVERY_EVENT_ALREADY_EXISTS')
|
|
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-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)
|
|
setting, direct = target_setting(config)
|
|
if old_env.get(setting) != direct:
|
|
raise RuntimeError('PRODUCER_NOT_DIRECT')
|
|
original_body = recreation(original, old_env, full=reverse_delivery(config))
|
|
token = secrets.token_urlsafe(32)
|
|
# Reserve task-specific ports and let the existing proxy select exactly this plan.
|
|
sockets = [socket.socket() for _ in range(2)]
|
|
for sock in sockets:
|
|
sock.bind(('127.0.0.1', 0))
|
|
ports = [sock.getsockname()[1] for sock in sockets]
|
|
for sock in sockets:
|
|
sock.close()
|
|
proxy = directory / 'proxy.py'
|
|
proxy.write_bytes(base64.b64decode(config['proxySource']))
|
|
ip = consumer['NetworkSettings']['Networks']['firstleg-test-network']['IPAddress']
|
|
log = open(directory / 'proxy-output.log', 'wb')
|
|
upstream = ':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')],
|
|
env=dict(os.environ, EVENT_FAULT_CONTROL_TOKEN=token), stdout=log, stderr=log, start_new_session=True)
|
|
state = {'scope': scope, 'originalBody': original_body, 'originalId': original['Id'], 'image': original['Config']['Image'],
|
|
'direction': config.get('direction', 'LOADING'), 'originalImageId': original['Image'], 'originalFingerprint': fingerprint(original),
|
|
'proxyPid': process.pid, 'proxyPort': ports[0], 'controlPort': ports[1], 'controlToken': token,
|
|
'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'], **({'prerequisite': prerequisite} if fee_void(config) else {})}
|
|
|
|
|
|
def activate(scope, directory, state):
|
|
snapshot = control(state, 'FAULT')
|
|
if snapshot.get('eventId'):
|
|
raise RuntimeError('EVENT_ALREADY_PINNED')
|
|
original = inspect(producer_name(state))
|
|
require_test(original, producer_name(state))
|
|
setting, direct = target_setting(state)
|
|
if original['Id'] != state['originalId'] or original['Config']['Image'] != state['image'] or env_of(original).get(setting) != direct:
|
|
raise RuntimeError('CONCURRENT_PRODUCER_CHANGE')
|
|
if reverse_delivery(state) and (original['Image'] != state['originalImageId'] or fingerprint(original) != state['originalFingerprint']):
|
|
raise RuntimeError('CONCURRENT_PRODUCER_CHANGE')
|
|
body = copy.deepcopy(state['originalBody'])
|
|
e = env_of(original)
|
|
e[setting] = 'http://host.docker.internal:' + str(state['proxyPort'])
|
|
e.update(EVENT_MAX_ATTEMPTS='12', EVENT_BASE_DELAY_SECONDS='5', EVENT_MAX_DELAY_SECONDS='5', EVENT_JITTER='0')
|
|
if len(e.get('EVENT_OPERATIONS_TOKEN', '')) < 32:
|
|
e['EVENT_OPERATIONS_TOKEN'] = secrets.token_urlsafe(32)
|
|
body['Env'] = [key + '=' + value for key, value in e.items()]
|
|
state['prepared'] = True
|
|
private_write(directory / 'state.json', state)
|
|
replace(original, body)
|
|
return {'activated': True}
|
|
|
|
|
|
def recover(scope, directory, state):
|
|
row = producer(scope, state.get('direction', 'LOADING'))
|
|
if not row or row['publishStatus'] != 'DEAD' or row['failedAttempts'] != 12:
|
|
raise RuntimeError('ORIGINAL_EVENT_NOT_DEAD')
|
|
pinned = control(state)
|
|
if pinned['eventId'] != row['eventId'] or pinned['payloadHash'] != row['payloadHash']:
|
|
raise RuntimeError('PINNED_EVENT_MISMATCH')
|
|
state['beforeRetry'] = {**row, **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))
|
|
e = env_of(container)
|
|
ip = container['NetworkSettings']['Networks']['firstleg-test-network']['IPAddress']
|
|
body = {'requestId': state['operationsRequestId'], 'operator': 'admin-full-test',
|
|
'reason': scope['marker'] + ': scoped ACK-loss recovery', 'expectedPayloadHash': row['payloadHash']}
|
|
endpoint = ':9303/internal/lastleg/event-operations/' if reverse_delivery(state) else ':9202/internal/firstleg/event-operations/'
|
|
url = 'http://' + ip + endpoint + row['entryId'] + '/retry'
|
|
request = urllib.request.Request(url, json.dumps(body).encode(),
|
|
{'X-Event-Operations-Token': e['EVENT_OPERATIONS_TOKEN'], 'Content-Type': 'application/json'})
|
|
with urllib.request.urlopen(request, timeout=15) as response:
|
|
result = json.load(response)
|
|
if result.get('code') != '200':
|
|
raise RuntimeError('OPERATIONS_RETRY_REJECTED')
|
|
return {'retrySubmitted': True}
|
|
|
|
|
|
def restore(directory, state):
|
|
if state['restored']:
|
|
return {'restored': True}
|
|
try:
|
|
control(state, 'PASS')
|
|
except Exception:
|
|
if not reverse_delivery(state):
|
|
raise
|
|
# A dead proxy must not prevent restoring the producer's direct target.
|
|
if state['prepared']:
|
|
try:
|
|
current = inspect(producer_name(state))
|
|
except RuntimeError as error:
|
|
if str(error) != 'DOCKER_404':
|
|
raise
|
|
created = docker('POST', '/containers/create?name=' + producer_name(state), state['originalBody'])
|
|
docker('POST', '/containers/' + created['Id'] + '/start')
|
|
state['prepared'] = False
|
|
private_write(directory / 'state.json', state)
|
|
return {'restoreStarted': True}
|
|
if current['Id'] == state['originalId'] or (reverse_delivery(state) and current['Image'] == state['originalImageId'] and fingerprint(current) == state['originalFingerprint']):
|
|
if reverse_delivery(state) and not current['State'].get('Running'):
|
|
docker('POST', '/containers/' + current['Id'] + '/start')
|
|
state['prepared'] = False
|
|
private_write(directory / 'state.json', state)
|
|
return {'restoreStarted': True}
|
|
if current['Config']['Image'] != state['image'] or env_of(current).get(target_setting(state)[0]) != 'http://host.docker.internal:' + str(state['proxyPort']):
|
|
raise RuntimeError('CONCURRENT_PRODUCER_CHANGE')
|
|
replace(current, state['originalBody'])
|
|
state['prepared'] = False
|
|
private_write(directory / 'state.json', state)
|
|
return {'restoreStarted': True}
|
|
|
|
|
|
def proof(scope, state):
|
|
row = producer(scope, state.get('direction', 'LOADING'))
|
|
if not row or row['publishStatus'] != 'PUBLISHED' or not state.get('beforeRetry'):
|
|
raise RuntimeError('RECOVERY_NOT_PUBLISHED')
|
|
event_id, entry_id = row['eventId'], row['entryId']
|
|
if not re.fullmatch(r'[0-9a-fA-F-]{36}|[0-9A-HJKMNP-TV-Z]{26}', event_id) or not re.fullmatch(r'[1-9][0-9]{0,19}', entry_id):
|
|
raise RuntimeError('INVALID_EVENT_IDENTITY')
|
|
event_type = '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;")
|
|
audit = read_sql(inspect(producer_name(state)),
|
|
"SELECT JSON_OBJECT('requestId',request_id,'entryId',entry_id,'eventId',event_id,'payloadHash',payload_hash,'previousFailures',previous_failures) FROM event_delivery_retry_audit WHERE request_id='"
|
|
+ state['operationsRequestId'] + "';" if not 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 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:
|
|
consumers = read_sql(inspect('firstleg-test-lastleg'),
|
|
"SELECT JSON_OBJECT('eventId',event_id,'payloadHash',payload_hash,'status',status) FROM ll_loading_event_inbox WHERE event_id='"
|
|
+ event_id + "' AND customer_id='" + scope['customerId'] + "';")
|
|
consumer = {**(consumers[0] if len(consumers) == 1 else {}), 'count': len(consumers)}
|
|
result = {**scope, 'readOnly': True, 'producer': row, 'consumer': consumer, 'attempts': attempts,
|
|
'proxyRecords': control(state)['records'], 'beforeRetry': state['beforeRetry'],
|
|
'afterRetry': {**row, **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
|
|
|
|
|
|
def finish(directory, state):
|
|
if state['restored']:
|
|
return state.get('restoration', {'restored': True})
|
|
current = inspect(producer_name(state))
|
|
require_test(current, producer_name(state))
|
|
if current['State'].get('Health', {}).get('Status') != 'healthy' or current['Config']['Image'] != state['image']:
|
|
raise RuntimeError('RESTORED_SERVICE_NOT_HEALTHY')
|
|
if sorted(current['Config']['Env']) != sorted(state['originalBody']['Env']):
|
|
raise RuntimeError('RESTORED_CONFIGURATION_MISMATCH')
|
|
if reverse_delivery(state) and (current['Image'] != state['originalImageId'] or fingerprint(current) != state['originalFingerprint']):
|
|
raise RuntimeError('RESTORED_CONFIGURATION_MISMATCH')
|
|
cmdline = Path('/proc/' + str(state['proxyPid']) + '/cmdline')
|
|
if cmdline.exists():
|
|
if str(directory / 'proxy.py').encode() not in cmdline.read_bytes().split(b'\x00'):
|
|
raise RuntimeError('PROXY_PROCESS_OWNERSHIP_MISMATCH')
|
|
os.kill(state['proxyPid'], signal.SIGTERM)
|
|
if reverse_delivery(state):
|
|
for _ in range(50):
|
|
if not cmdline.exists() or not cmdline.read_bytes():
|
|
break
|
|
time.sleep(0.1)
|
|
else:
|
|
raise RuntimeError('PROXY_NOT_STOPPED')
|
|
result = {'restored': True}
|
|
if reverse_delivery(state):
|
|
result.update(healthy=True, proxyStopped=True, producerName=producer_name(state),
|
|
originalImageId=state['originalImageId'], currentImageId=current['Image'],
|
|
originalFingerprint=state['originalFingerprint'], currentFingerprint=fingerprint(current))
|
|
state['restoration'] = result
|
|
state['restored'] = True
|
|
# Proof is already captured. Remove credentials from the retained task metadata.
|
|
state.pop('originalBody', None)
|
|
state.pop('controlToken', None)
|
|
private_write(directory / 'state.json', state)
|
|
return result
|
|
|
|
|
|
def main(config):
|
|
scope, action = config['scope'], config['action']
|
|
direction = config.get('direction', 'LOADING')
|
|
directory = scope_directory(scope, direction)
|
|
if action == '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())
|
|
if state['scope'] != scope or state.get('direction', 'LOADING') != direction:
|
|
raise RuntimeError('TASK_SCOPE_MISMATCH')
|
|
if action == 'activate':
|
|
return activate(scope, directory, state)
|
|
if action == 'health':
|
|
current = inspect(producer_name(state))
|
|
return {'healthy': current['State'].get('Health', {}).get('Status') == 'healthy'}
|
|
if action == 'proxy-ready':
|
|
return {'ready': control(state)['mode'] == 'PASS'}
|
|
if action == 'status':
|
|
return {'producer': producer(scope, direction), 'proxy': control(state)}
|
|
if action == 'recover':
|
|
return recover(scope, directory, state)
|
|
if action == 'proof':
|
|
return proof(scope, state)
|
|
if action == 'restore':
|
|
return restore(directory, state)
|
|
if action == 'finish':
|
|
return finish(directory, state)
|
|
raise RuntimeError('INVALID_ACTION')
|
|
|
|
|
|
if __name__ == '__main__':
|
|
try:
|
|
value = main(json.loads(base64.b64decode(sys.argv[1])))
|
|
print('TEST_EVENT_JSON=' + json.dumps({'ok': True, 'data': value}))
|
|
except Exception as error:
|
|
message = str(error)
|
|
safe = message if re.fullmatch(r'[A-Z_0-9]{1,80}', message) else 'TEST_EVENT_ACTION_FAILED'
|
|
print('TEST_EVENT_JSON=' + json.dumps({'ok': False, 'error': safe}))
|