"""Reference solution — the integration fabric: ISO 20022, outbox, log, registry, products. An AI platform in a bank is only as useful as the systems it can reach, and those systems are thirty years old, extremely reliable, and were not designed with you in mind. Five mechanisms, and each is boring in a way that matters: * **ISO 20022** — the vocabulary is standardized and the standard is large. Getting `pain.001` wrong is not a validation error, it is a rejected payment. * **The transactional outbox** — the fix for the dual-write problem, which is the single most common source of "the event never arrived" in an event-driven bank. * **A partitioned log** — where ordering holds, where it does not, and what the difference costs. * **A schema registry** — because compatibility mode *is* deploy order, and finding that out during a release is how you learn it. * **Data-product contracts** — schema, freshness, quality, ownership, checked rather than asserted. Deterministic: an injected clock, derived identifiers, money as integer minor units, sorted outputs, no network. ``python solution.py`` runs the worked example. """ from __future__ import annotations import hashlib import re import xml.etree.ElementTree as ET from dataclasses import dataclass, field, replace from decimal import Decimal, InvalidOperation from enum import Enum from typing import (Any, Callable, Dict, Iterable, List, Mapping, Optional, Sequence, Set, Tuple) # ====================================================================================== # 1. ISO 20022 — reading a message name # ====================================================================================== @dataclass(frozen=True) class MessageIdentifier: """`pain.001.001.09` — and every part of it means something. pain . 001 . 001 . 09 │ │ │ └── version │ │ └────────── variant (almost always 001) │ └────────────────── message number within the family └─────────────────────────── business area Knowing how to *read* the name is most of what a working knowledge of ISO 20022 means in an interview. The details are lookups; the families are the map. """ business_area: str message_number: str variant: str version: str @classmethod def parse(cls, name: str) -> "MessageIdentifier": match = re.fullmatch(r"([a-z]{4})\.(\d{3})\.(\d{3})\.(\d{2})", name.strip()) if not match: raise ValueError(f"not an ISO 20022 message identifier: {name!r}") return cls(*match.groups()) def __str__(self) -> str: return (f"{self.business_area}.{self.message_number}.{self.variant}." f"{self.version}") @property def family(self) -> str: return BUSINESS_AREAS.get(self.business_area, "unknown") @property def description(self) -> str: return MESSAGE_CATALOGUE.get( f"{self.business_area}.{self.message_number}", "unknown message") BUSINESS_AREAS: Mapping[str, str] = { "pain": "Payments Initiation", # customer -> bank "pacs": "Payments Clearing and Settlement", # bank -> bank "camt": "Cash Management", # statements, balances, investigations "acmt": "Account Management", "auth": "Authorities (regulatory reporting)", "reda": "Reference Data", "seev": "Securities Events", "setr": "Securities Trade", "tsmt": "Trade Services Management", } MESSAGE_CATALOGUE: Mapping[str, str] = { "pain.001": "CustomerCreditTransferInitiation", "pain.002": "CustomerPaymentStatusReport", "pain.008": "CustomerDirectDebitInitiation", "pacs.002": "FIToFIPaymentStatusReport", "pacs.004": "PaymentReturn", "pacs.008": "FIToFICustomerCreditTransfer", "pacs.009": "FinancialInstitutionCreditTransfer", "camt.052": "BankToCustomerAccountReport", "camt.053": "BankToCustomerStatement", "camt.054": "BankToCustomerDebitCreditNotification", "camt.056": "FIToFIPaymentCancellationRequest", } # ====================================================================================== # 2. pain.001 — parsing and validating # ====================================================================================== NS = "urn:iso:std:iso:20022:tech:xsd:pain.001.001.09" class Severity(str, Enum): ERROR = "error" # the message is rejected WARNING = "warning" # accepted, but somebody should look @dataclass(frozen=True) class Rejection: """A rejection names the ELEMENT, not just the problem. "Invalid message" sends a human to read 400 lines of XML. `[1]/` sends them to one line. In a bank the difference is a day. """ code: str element: str reason: str severity: Severity = Severity.ERROR def __str__(self) -> str: return f"[{self.code}] {self.element}: {self.reason}" @dataclass(frozen=True) class Party: name: str iban: Optional[str] = None bic: Optional[str] = None country: Optional[str] = None address_lines: Tuple[str, ...] = () @dataclass(frozen=True) class CreditTransfer: end_to_end_id: str amount_minor: int # integer minor units. Never a float. currency: str creditor: Party remittance_info: str = "" @dataclass(frozen=True) class PaymentInstruction: message_id: str creation_datetime: str number_of_transactions: int control_sum_minor: int initiating_party: str payment_information_id: str requested_execution_date: str debtor: Party transfers: Tuple[CreditTransfer, ...] @property def actual_sum_minor(self) -> int: return sum(t.amount_minor for t in self.transfers) #: Currencies whose minor unit is not 2 digits. Assuming 2 everywhere is the classic #: cross-border bug: JPY 1000 parsed as 100000 minor units is a payment 100× too large. CURRENCY_EXPONENT: Mapping[str, int] = { "JPY": 0, "KRW": 0, "VND": 0, "CLP": 0, "ISK": 0, "BHD": 3, "KWD": 3, "OMR": 3, "TND": 3, "JOD": 3, } DEFAULT_EXPONENT = 2 VALID_CURRENCIES = frozenset({ "AED", "USD", "EUR", "GBP", "JPY", "SAR", "KWD", "BHD", "CHF", "SGD", "INR"}) def currency_exponent(currency: str) -> int: return CURRENCY_EXPONENT.get(currency.upper(), DEFAULT_EXPONENT) def to_minor(amount: str, currency: str) -> int: """Parse a decimal amount string into integer minor units. Decimal, not float. `float("0.10") * 100` is `10.000000000000002`, and a payment system that rounds that is a payment system with a reconciliation break every day. """ try: value = Decimal(amount) except (InvalidOperation, TypeError): raise ValueError(f"not a decimal amount: {amount!r}") from None exponent = currency_exponent(currency) scaled = value.scaleb(exponent) if scaled != scaled.to_integral_value(): raise ValueError( f"{amount} has more decimal places than {currency} permits ({exponent})") return int(scaled) def from_minor(minor: int, currency: str) -> str: exponent = currency_exponent(currency) return str(Decimal(minor).scaleb(-exponent)) def valid_iban(candidate: str) -> bool: """ISO 13616 mod-97. Same check as Phase 11 — banks reuse their checksums.""" s = candidate.replace(" ", "").upper() if not re.fullmatch(r"[A-Z]{2}\d{2}[A-Z0-9]{11,30}", s): return False rearranged = s[4:] + s[:4] expanded = "".join(str(ord(c) - 55) if c.isalpha() else c for c in rearranged) return expanded.isdigit() and int(expanded) % 97 == 1 def valid_bic(candidate: str) -> bool: """ISO 9362: 4 institution + 2 country + 2 location, optionally + 3 branch.""" return bool(re.fullmatch(r"[A-Z]{4}[A-Z]{2}[A-Z0-9]{2}([A-Z0-9]{3})?", candidate.strip().upper())) class Pain001Parser: """Parse and validate a pain.001 CustomerCreditTransferInitiation. The design rule: **collect every rejection, then decide.** A parser that raises on the first problem forces the sender into one round trip per error, and a payment file with six problems takes six days to fix. """ def __init__(self, *, max_transactions: int = 1000) -> None: self.max_transactions = max_transactions def parse(self, xml: str) -> Tuple[Optional[PaymentInstruction], List[Rejection]]: rejections: List[Rejection] = [] try: root = ET.fromstring(xml) except ET.ParseError as exc: return None, [Rejection("FF01", "/", f"malformed XML: {exc}")] # The namespace pins the version. A pain.001.001.09 parser must not silently # accept an .001.03 document whose element semantics differ. namespace = root.tag[1:root.tag.index("}")] if root.tag.startswith("{") else "" if namespace != NS: rejections.append(Rejection( "FF02", "/Document", f"expected namespace {NS}, got {namespace!r}")) return None, rejections ns = {"p": NS} header = root.find(".//p:GrpHdr", ns) if header is None: return None, [Rejection("MS01", "/Document/CstmrCdtTrfInitn/GrpHdr", "mandatory element is absent")] message_id = _text(header, "p:MsgId", ns) creation = _text(header, "p:CreDtTm", ns) nb_of_txs = _text(header, "p:NbOfTxs", ns) control_sum = _text(header, "p:CtrlSum", ns) initiating = _text(header.find("p:InitgPty", ns), "p:Nm", ns) if \ header.find("p:InitgPty", ns) is not None else None for name, value, element in ( ("MsgId", message_id, "GrpHdr/MsgId"), ("CreDtTm", creation, "GrpHdr/CreDtTm"), ("NbOfTxs", nb_of_txs, "GrpHdr/NbOfTxs"), ): if not value: rejections.append(Rejection("MS01", element, "mandatory element is absent")) payment_info = root.find(".//p:PmtInf", ns) if payment_info is None: rejections.append(Rejection("MS01", "CstmrCdtTrfInitn/PmtInf", "mandatory element is absent")) return None, sorted(rejections, key=lambda r: (r.element, r.code)) pmt_inf_id = _text(payment_info, "p:PmtInfId", ns) exec_date = _text(payment_info, "p:ReqdExctnDt", ns) debtor, debtor_rejections = self._party( payment_info, "p:Dbtr", "p:DbtrAcct", "p:DbtrAgt", ns, "PmtInf/Dbtr") rejections.extend(debtor_rejections) if not pmt_inf_id: rejections.append(Rejection("MS01", "PmtInf/PmtInfId", "mandatory element is absent")) if not exec_date: rejections.append(Rejection("MS01", "PmtInf/ReqdExctnDt", "mandatory element is absent")) elif not re.fullmatch(r"\d{4}-\d{2}-\d{2}", exec_date): rejections.append(Rejection("FF03", "PmtInf/ReqdExctnDt", f"expected ISO date, got {exec_date!r}")) transfers: List[CreditTransfer] = [] nodes = payment_info.findall("p:CdtTrfTxInf", ns) if not nodes: rejections.append(Rejection("MS01", "PmtInf/CdtTrfTxInf", "at least one transaction is required")) if len(nodes) > self.max_transactions: rejections.append(Rejection( "LM01", "PmtInf/CdtTrfTxInf", f"{len(nodes)} transactions exceeds the limit of " f"{self.max_transactions}")) seen_ids: Set[str] = set() for index, node in enumerate(nodes): transfer, txn_rejections = self._transfer(node, ns, index) rejections.extend(txn_rejections) if transfer is None: continue if transfer.end_to_end_id in seen_ids: # A duplicate EndToEndId in one file is how the same payment gets made # twice: downstream idempotency keys on it. rejections.append(Rejection( "DU01", f"CdtTrfTxInf[{index}]/PmtId/EndToEndId", f"duplicate EndToEndId {transfer.end_to_end_id!r}")) continue seen_ids.add(transfer.end_to_end_id) transfers.append(transfer) # Cross-field checks — the ones a per-element schema cannot express. if nb_of_txs and nb_of_txs.isdigit() and int(nb_of_txs) != len(nodes): rejections.append(Rejection( "CS01", "GrpHdr/NbOfTxs", f"declares {nb_of_txs} transactions, found {len(nodes)}")) instruction: Optional[PaymentInstruction] = None if not any(r.severity is Severity.ERROR for r in rejections): control = 0 if control_sum and transfers: try: control = to_minor(control_sum, transfers[0].currency) except ValueError: control = 0 instruction = PaymentInstruction( message_id=message_id or "", creation_datetime=creation or "", number_of_transactions=len(transfers), control_sum_minor=control, initiating_party=initiating or "", payment_information_id=pmt_inf_id or "", requested_execution_date=exec_date or "", debtor=debtor or Party(""), transfers=tuple(transfers)) if control_sum and control != instruction.actual_sum_minor: rejections.append(Rejection( "CS02", "GrpHdr/CtrlSum", f"declares {control_sum}, transactions sum to " f"{from_minor(instruction.actual_sum_minor, transfers[0].currency)}")) instruction = None return instruction, sorted(rejections, key=lambda r: (r.element, r.code)) # -- helpers --------------------------------------------------------------------- def _transfer(self, node: ET.Element, ns: Mapping[str, str], index: int) -> Tuple[Optional[CreditTransfer], List[Rejection]]: base = f"CdtTrfTxInf[{index}]" rejections: List[Rejection] = [] pmt_id = node.find("p:PmtId", ns) end_to_end = _text(pmt_id, "p:EndToEndId", ns) if pmt_id is not None else None if not end_to_end: rejections.append(Rejection("MS01", f"{base}/PmtId/EndToEndId", "mandatory element is absent")) amount_node = node.find("p:Amt/p:InstdAmt", ns) amount_minor: Optional[int] = None currency = "" if amount_node is None or not (amount_node.text or "").strip(): rejections.append(Rejection("MS01", f"{base}/Amt/InstdAmt", "mandatory element is absent")) else: currency = (amount_node.get("Ccy") or "").upper() if not currency: rejections.append(Rejection( "MS02", f"{base}/Amt/InstdAmt@Ccy", "the currency attribute is mandatory on an amount")) elif currency not in VALID_CURRENCIES: rejections.append(Rejection("FF04", f"{base}/Amt/InstdAmt@Ccy", f"unsupported currency {currency!r}")) else: try: amount_minor = to_minor(amount_node.text.strip(), currency) except ValueError as exc: rejections.append(Rejection("FF05", f"{base}/Amt/InstdAmt", str(exc))) else: if amount_minor <= 0: rejections.append(Rejection( "AM01", f"{base}/Amt/InstdAmt", "amount must be strictly positive")) creditor, creditor_rejections = self._party( node, "p:Cdtr", "p:CdtrAcct", "p:CdtrAgt", ns, f"{base}/Cdtr") rejections.extend(creditor_rejections) remittance = "" rmt = node.find("p:RmtInf/p:Ustrd", ns) if rmt is not None and rmt.text: remittance = rmt.text.strip() if end_to_end and amount_minor is not None and creditor is not None: return CreditTransfer(end_to_end, amount_minor, currency, creditor, remittance), rejections return None, rejections def _party(self, parent: Optional[ET.Element], party_tag: str, account_tag: str, agent_tag: str, ns: Mapping[str, str], element: str) -> Tuple[Optional[Party], List[Rejection]]: rejections: List[Rejection] = [] if parent is None: return None, [Rejection("MS01", element, "mandatory element is absent")] party = parent.find(party_tag, ns) name = _text(party, "p:Nm", ns) if party is not None else None if not name: rejections.append(Rejection("MS01", f"{element}/Nm", "mandatory element is absent")) country = None address_lines: List[str] = [] if party is not None: postal = party.find("p:PstlAdr", ns) if postal is not None: country = _text(postal, "p:Ctry", ns) address_lines = [ln.text.strip() for ln in postal.findall("p:AdrLine", ns) if ln.text] if country and not re.fullmatch(r"[A-Z]{2}", country): rejections.append(Rejection( "FF06", f"{element}/PstlAdr/Ctry", f"expected an ISO 3166 alpha-2 code, got {country!r}")) iban = None account = parent.find(account_tag, ns) if account is None: rejections.append(Rejection("MS01", f"{element}Acct", "mandatory element is absent")) else: iban = _text(account.find("p:Id", ns), "p:IBAN", ns) if \ account.find("p:Id", ns) is not None else None if not iban: rejections.append(Rejection("MS01", f"{element}Acct/Id/IBAN", "mandatory element is absent")) elif not valid_iban(iban): rejections.append(Rejection("FF07", f"{element}Acct/Id/IBAN", f"IBAN {iban!r} fails the mod-97 check")) bic = None agent = parent.find(agent_tag, ns) if agent is not None: bic = _text(agent.find("p:FinInstnId", ns), "p:BICFI", ns) if \ agent.find("p:FinInstnId", ns) is not None else None if bic and not valid_bic(bic): rejections.append(Rejection("FF08", f"{element}Agt/FinInstnId/BICFI", f"{bic!r} is not a valid BIC")) if name: return Party(name, iban, bic, country, tuple(address_lines)), rejections return None, rejections def _text(node: Optional[ET.Element], path: str, ns: Mapping[str, str]) -> Optional[str]: if node is None: return None found = node.find(path, ns) if found is None or found.text is None: return None return found.text.strip() or None # ====================================================================================== # 3. Payment rails and finality # ====================================================================================== class Finality(str, Enum): """When a payment stops being reversible — a SCHEME rule, not a database property. This is the phase's link back to [Phase 10]'s side-effect classes: `irreversible` is not an engineering opinion about difficulty, it is a statement about what the scheme permits. An agent's autonomy is bounded by the rail it is using. """ REVOCABLE = "revocable" # before cut-off: cancel freely CONDITIONALLY_REVOCABLE = "conditionally_revocable" # a recall request; may be refused FINAL = "final" # settled; only a NEW payment moves it back @dataclass(frozen=True) class Rail: rail_id: str name: str currency: str cutoff_minute: int # minutes past midnight, local settles_after_minutes: int max_amount_minor: Optional[int] = None supports_recall: bool = True def finality_at(self, submitted_minute: int, now_minute: int) -> Finality: """Settlement dominates the cut-off. The ordering matters and is easy to get backwards: an *instant* payment settles in zero minutes, so it is FINAL the moment it is submitted — despite having no cut-off at all. A cut-off is only meaningful for a rail that batches, and checking it first would report the most irrevocable rail in the bank as revocable. """ if now_minute >= submitted_minute + self.settles_after_minutes: return Finality.FINAL if now_minute < self.cutoff_minute and submitted_minute < self.cutoff_minute: return Finality.REVOCABLE return (Finality.CONDITIONALLY_REVOCABLE if self.supports_recall else Finality.FINAL) RAILS: Mapping[str, Rail] = { "instant": Rail("instant", "Instant Payments (IPI)", "AED", cutoff_minute=1440, settles_after_minutes=0, max_amount_minor=50_000_00, supports_recall=False), "rtgs": Rail("rtgs", "UAEFTS (RTGS)", "AED", cutoff_minute=15 * 60, settles_after_minutes=15, supports_recall=False), "ach": Rail("ach", "WPS/ACH batch", "AED", cutoff_minute=14 * 60, settles_after_minutes=24 * 60, supports_recall=True), "swift": Rail("swift", "SWIFT cross-border", "USD", cutoff_minute=16 * 60, settles_after_minutes=48 * 60, supports_recall=True), } def choose_rail(amount_minor: int, currency: str, now_minute: int, *, urgent: bool = False) -> Tuple[Optional[str], str]: """Pick a rail, and say WHY — including why the others were excluded. An agent proposing a payment must be able to explain the rail choice, because the rail determines the finality, which determines whether a human must approve. """ candidates = [r for r in RAILS.values() if r.currency == currency] if not candidates: return None, f"no rail carries {currency}" reasons: List[str] = [] for rail in sorted(candidates, key=lambda r: r.settles_after_minutes): if rail.max_amount_minor is not None and amount_minor > rail.max_amount_minor: reasons.append(f"{rail.rail_id}: above its limit") continue if now_minute >= rail.cutoff_minute: reasons.append(f"{rail.rail_id}: past its {rail.cutoff_minute // 60}:00 cut-off") continue if urgent and rail.settles_after_minutes > 60: reasons.append(f"{rail.rail_id}: too slow for an urgent payment") continue if reasons: return rail.rail_id, f"after ruling out {', '.join(reasons)}" return rail.rail_id, "fastest rail that carries it" return None, "; ".join(reasons) # ====================================================================================== # 4. The transactional outbox # ====================================================================================== @dataclass(frozen=True) class OutboxRecord: seq: int aggregate_id: str event_type: str payload: Mapping[str, Any] created_at: int published_at: Optional[int] = None @property def published(self) -> bool: return self.published_at is not None class DualWriteStore: """A deliberately BROKEN store, to demonstrate the problem the outbox solves. The dual-write problem: writing to the database and publishing to a broker are two separate systems, so a crash between them leaves them disagreeing — and there is no ordering of the two writes that fixes it. * DB then publish -> the crash loses the event. The payment happened; nothing downstream knows. * Publish then DB -> the crash produces a phantom event. Downstream reacts to a payment that does not exist. The second is worse, and neither is acceptable. """ def __init__(self, *, broker: "MessageLog", fail_after_db: bool = False) -> None: self.rows: Dict[str, Mapping[str, Any]] = {} self.broker = broker self.fail_after_db = fail_after_db def save_and_publish(self, key: str, row: Mapping[str, Any], topic: str) -> None: self.rows[key] = dict(row) # write 1: the database if self.fail_after_db: raise RuntimeError("crashed between the database write and the publish") self.broker.append(topic, key, dict(row)) # write 2: the broker class Outbox: """Domain change and outbound event in ONE local transaction; relay afterwards. The insight is that both writes go to the *same* database, so the database's own atomicity covers them. A separate relay then reads unpublished rows and publishes them — at-least-once, because the relay can crash after publishing and before marking. Which is exactly why the consumer must be idempotent (§6). The outbox does not give you exactly-once delivery; it gives you *no lost events*, and idempotency turns that into exactly-once effects. """ def __init__(self, *, now: Callable[[], int]) -> None: self.now = now self.rows: Dict[str, Mapping[str, Any]] = {} self._outbox: List[OutboxRecord] = [] self._seq = 0 def transact(self, key: str, row: Mapping[str, Any], *, event_type: str, payload: Optional[Mapping[str, Any]] = None, fail_before_commit: bool = False) -> OutboxRecord: """Both writes, atomically. A failure before commit leaves NEITHER.""" self._seq += 1 record = OutboxRecord(self._seq, key, event_type, dict(payload or row), self.now()) if fail_before_commit: self._seq -= 1 raise RuntimeError("crashed before commit; neither write happened") self.rows[key] = dict(row) self._outbox.append(record) return record def unpublished(self) -> List[OutboxRecord]: return [r for r in self._outbox if not r.published] def mark_published(self, seq: int) -> None: for i, record in enumerate(self._outbox): if record.seq == seq: self._outbox[i] = replace(record, published_at=self.now()) return raise KeyError(f"no outbox record {seq}") def records(self) -> List[OutboxRecord]: return list(self._outbox) class OutboxRelay: """Publishes unpublished outbox rows, in sequence order. ``fail_after_publish`` models the crash that makes at-least-once real: the event reached the broker, the row was not marked, and the next run publishes it again. """ def __init__(self, *, outbox: Outbox, log: "MessageLog", topic: str) -> None: self.outbox = outbox self.log = log self.topic = topic self.published = 0 def run_once(self, *, fail_after_publish: bool = False) -> int: count = 0 for record in sorted(self.outbox.unpublished(), key=lambda r: r.seq): self.log.append(self.topic, record.aggregate_id, { "event_type": record.event_type, "seq": record.seq, **record.payload}) count += 1 self.published += 1 if fail_after_publish: # Published, not marked. The next run will publish it again. return count self.outbox.mark_published(record.seq) return count # ====================================================================================== # 5. A partitioned log # ====================================================================================== @dataclass(frozen=True) class Message: partition: int offset: int key: str value: Mapping[str, Any] timestamp: int class MessageLog: """A partitioned, append-only log — Kafka and Event Hubs are the same model. | Kafka | Event Hubs | |---|---| | topic | event hub | | partition | partition | | consumer group | consumer group | | offset | offset / sequence number | **Ordering is per partition, never across.** That is the single most important property, and the routing decision below is what makes it useful: keying by account means every event for one account lands in one partition, so per-account order holds even though global order does not. """ def __init__(self, *, partitions: int = 4, now: Optional[Callable[[], int]] = None) -> None: if partitions < 1: raise ValueError("a log needs at least one partition") self.partitions = partitions self.now = now or (lambda: 0) self._log: Dict[Tuple[str, int], List[Message]] = {} self._topics: Set[str] = set() def partition_for(self, key: str) -> int: """Derived, never ``hash()`` — Python salts string hashing per process, so ``hash()`` would route the same key to different partitions across restarts and silently break ordering.""" digest = hashlib.blake2b(key.encode(), digest_size=8).digest() return int.from_bytes(digest, "big") % self.partitions def append(self, topic: str, key: str, value: Mapping[str, Any]) -> Message: self._topics.add(topic) partition = self.partition_for(key) bucket = self._log.setdefault((topic, partition), []) message = Message(partition, len(bucket), key, dict(value), self.now()) bucket.append(message) return message def read(self, topic: str, partition: int, *, from_offset: int = 0, limit: Optional[int] = None) -> List[Message]: bucket = self._log.get((topic, partition), []) window = bucket[from_offset:] return window[:limit] if limit is not None else window def all(self, topic: str) -> List[Message]: """Every message, partition then offset. NOT a global ordering — it is a deterministic *listing*, and the distinction matters.""" out: List[Message] = [] for partition in range(self.partitions): out.extend(self._log.get((topic, partition), [])) return out def high_water_mark(self, topic: str, partition: int) -> int: return len(self._log.get((topic, partition), [])) class ConsumerGroup: """Offsets per (topic, partition), committed explicitly. Two ordering rules that follow directly from the model: * **commit AFTER processing** — commit first and a crash loses the message (at-most-once); * **at-least-once is the practical default** — so processing must be idempotent. """ def __init__(self, group_id: str, log: MessageLog) -> None: self.group_id = group_id self.log = log self._offsets: Dict[Tuple[str, int], int] = {} def committed(self, topic: str, partition: int) -> int: return self._offsets.get((topic, partition), 0) def poll(self, topic: str, *, max_records: int = 10) -> List[Message]: out: List[Message] = [] for partition in range(self.log.partitions): offset = self.committed(topic, partition) out.extend(self.log.read(topic, partition, from_offset=offset, limit=max_records - len(out))) if len(out) >= max_records: break return out def commit(self, topic: str, partition: int, offset: int) -> None: """The committed offset is the NEXT one to read, not the last one read. Off-by-one here means either re-reading the last message forever or skipping it — and which one you get depends on a convention nobody wrote down. """ current = self.committed(topic, partition) if offset < current: raise ValueError( f"cannot rewind {topic}/{partition} from {current} to {offset} " f"with commit(); use seek()") self._offsets[(topic, partition)] = offset def seek(self, topic: str, partition: int, offset: int) -> None: """Deliberate replay. Separate from ``commit`` so a rewind is never accidental.""" self._offsets[(topic, partition)] = max(0, offset) def lag(self, topic: str) -> Dict[int, int]: return {p: self.log.high_water_mark(topic, p) - self.committed(topic, p) for p in range(self.log.partitions)} def total_lag(self, topic: str) -> int: return sum(self.lag(topic).values()) # ====================================================================================== # 6. The idempotent consumer # ====================================================================================== class IdempotentConsumer: """At-least-once delivery + idempotent handling = **exactly-once effects**. Exactly-once *delivery* is impossible (two generals). Exactly-once *effects* is routine, and it is what anybody actually wants. The whole mechanism is: derive a stable id from the message, remember the ids you have processed, skip the repeats. The id must come from the message's CONTENT or a business key — never from the offset, because a replay from a different partition assignment yields different offsets for the same event. """ def __init__(self, group: ConsumerGroup, handler: Callable[[Message], Any], *, key_of: Optional[Callable[[Message], str]] = None) -> None: self.group = group self.handler = handler self.key_of = key_of or (lambda m: str(m.value.get("event_id") or f"{m.key}:{m.value.get('seq')}")) self.processed: Set[str] = set() self.effects: List[Any] = [] self.duplicates = 0 def consume(self, topic: str, *, max_records: int = 100, fail_before_commit: bool = False) -> int: handled = 0 for message in self.group.poll(topic, max_records=max_records): event_id = self.key_of(message) if event_id in self.processed: self.duplicates += 1 else: self.effects.append(self.handler(message)) self.processed.add(event_id) handled += 1 if fail_before_commit: # The effect happened; the offset was not committed. The next poll # re-delivers, and the dedup set is what makes that harmless. return handled self.group.commit(topic, message.partition, message.offset + 1) return handled # ====================================================================================== # 7. The schema registry # ====================================================================================== class Compatibility(str, Enum): """The mode IS the deploy order. This is the whole content of the section. | Mode | New schema readable by | Upgrade first | |---|---|---| | BACKWARD | the NEW reader, on OLD data | **consumers** | | FORWARD | the OLD reader, on NEW data | **producers** | | FULL | both | either | | NONE | nothing guaranteed | pray | Getting this backwards means a release where the consumers cannot read what the producers are writing, discovered in production because staging deployed everything at once. """ BACKWARD = "backward" FORWARD = "forward" FULL = "full" NONE = "none" @dataclass(frozen=True) class Field: name: str type: str required: bool = True default: Any = None @dataclass(frozen=True) class Schema: subject: str version: int fields: Tuple[Field, ...] def field(self, name: str) -> Optional[Field]: return next((f for f in self.fields if f.name == name), None) @property def names(self) -> FrozenSetLike: return frozenset(f.name for f in self.fields) FrozenSetLike = Any # keep the annotation readable without importing FrozenSet here def check_backward(old: Schema, new: Schema) -> List[str]: """Can a NEW reader read OLD data? Consumers upgrade first. A new *required* field breaks it: old data does not have it and there is no default to fall back on. Adding it with a default is fine, which is why "always give new fields a default" is such durable advice. """ problems: List[str] = [] for f in new.fields: if f.required and f.default is None and old.field(f.name) is None: problems.append( f"new required field {f.name!r} has no default; old data lacks it") for f in old.fields: matching = new.field(f.name) if matching is not None and matching.type != f.type: problems.append( f"field {f.name!r} changed type {f.type} -> {matching.type}") return sorted(problems) def check_forward(old: Schema, new: Schema) -> List[str]: """Can an OLD reader read NEW data? Producers upgrade first. Removing a required field breaks it: the old reader still demands it. """ problems: List[str] = [] for f in old.fields: if f.required and new.field(f.name) is None: problems.append( f"required field {f.name!r} was removed; old readers still expect it") for f in old.fields: matching = new.field(f.name) if matching is not None and matching.type != f.type: problems.append( f"field {f.name!r} changed type {f.type} -> {matching.type}") return sorted(problems) def check_compatibility(old: Schema, new: Schema, mode: Compatibility) -> List[str]: if mode is Compatibility.NONE: return [] if mode is Compatibility.BACKWARD: return check_backward(old, new) if mode is Compatibility.FORWARD: return check_forward(old, new) return sorted(set(check_backward(old, new)) | set(check_forward(old, new))) DEPLOY_ORDER: Mapping[Compatibility, str] = { Compatibility.BACKWARD: "consumers first, then producers", Compatibility.FORWARD: "producers first, then consumers", Compatibility.FULL: "either order", Compatibility.NONE: "coordinate manually; there is no guarantee", } class SchemaRegistry: def __init__(self, *, default_mode: Compatibility = Compatibility.BACKWARD) -> None: self.default_mode = default_mode self._subjects: Dict[str, List[Schema]] = {} self._modes: Dict[str, Compatibility] = {} def set_mode(self, subject: str, mode: Compatibility) -> None: self._modes[subject] = mode def mode(self, subject: str) -> Compatibility: return self._modes.get(subject, self.default_mode) def register(self, subject: str, fields: Sequence[Field]) -> Schema: versions = self._subjects.setdefault(subject, []) candidate = Schema(subject, len(versions) + 1, tuple(fields)) if versions: problems = check_compatibility(versions[-1], candidate, self.mode(subject)) if problems: raise ValueError( f"{subject} v{candidate.version} is not " f"{self.mode(subject).value}-compatible: {'; '.join(problems)}") versions.append(candidate) return candidate def latest(self, subject: str) -> Schema: return self._subjects[subject][-1] def version(self, subject: str, version: int) -> Schema: return self._subjects[subject][version - 1] def deploy_order(self, subject: str) -> str: return DEPLOY_ORDER[self.mode(subject)] # ====================================================================================== # 8. Data products # ====================================================================================== @dataclass(frozen=True) class QualityRule: name: str check: Callable[[Sequence[Mapping[str, Any]]], bool] description: str @dataclass(frozen=True) class DataProductContract: """Schema, semantics, quality, freshness, ownership. All five, or it is a table. The freshness SLO is the one that separates a data *product* from a dataset: it is a promise with a number, which means it can be breached, which means somebody can be told. """ product_id: str owner: str # a named human schema: Schema freshness_slo_ticks: int quality_rules: Tuple[QualityRule, ...] = () classification: str = "internal" description: str = "" @dataclass(frozen=True) class ContractResult: product_id: str passed: bool schema_problems: Tuple[str, ...] quality_failures: Tuple[str, ...] freshness_lag: int fresh: bool def report(self) -> str: if self.passed: return f"{self.product_id}: OK (lag {self.freshness_lag})" parts: List[str] = [] if self.schema_problems: parts.append(f"schema: {', '.join(self.schema_problems)}") if self.quality_failures: parts.append(f"quality: {', '.join(self.quality_failures)}") if not self.fresh: parts.append(f"stale by {self.freshness_lag}") return f"{self.product_id}: FAIL — " + "; ".join(parts) class DataProduct: def __init__(self, contract: DataProductContract, *, now: Callable[[], int]) -> None: self.contract = contract self.now = now self.rows: List[Mapping[str, Any]] = [] self.last_updated = now() def publish(self, rows: Sequence[Mapping[str, Any]]) -> None: self.rows = [dict(r) for r in rows] self.last_updated = self.now() def check(self) -> ContractResult: schema_problems: List[str] = [] for index, row in enumerate(self.rows): for f in self.contract.schema.fields: if f.required and f.name not in row: schema_problems.append(f"row {index}: {f.name} missing") elif f.name in row and not _type_ok(row[f.name], f.type): schema_problems.append( f"row {index}: {f.name} is not {f.type}") quality_failures = [rule.name for rule in self.contract.quality_rules if not rule.check(self.rows)] lag = self.now() - self.last_updated fresh = lag <= self.contract.freshness_slo_ticks return ContractResult( self.contract.product_id, not schema_problems and not quality_failures and fresh, tuple(sorted(set(schema_problems))[:5]), tuple(sorted(quality_failures)), lag, fresh) _TYPES: Mapping[str, tuple] = { "string": (str,), "int": (int,), "float": (int, float), "bool": (bool,), } def _type_ok(value: Any, type_name: str) -> bool: allowed = _TYPES.get(type_name) if allowed is None: return True if type_name == "int" and isinstance(value, bool): return False # bool is a subclass of int return isinstance(value, allowed) # ====================================================================================== # 9. Reconciliation # ====================================================================================== class BreakType(str, Enum): MISSING_IN_B = "missing_in_b" MISSING_IN_A = "missing_in_a" VALUE_MISMATCH = "value_mismatch" @dataclass(frozen=True) class Break: break_type: BreakType key: str a_value: Optional[int] b_value: Optional[int] def __str__(self) -> str: if self.break_type is BreakType.VALUE_MISMATCH: return f"{self.key}: A={self.a_value} B={self.b_value}" return f"{self.key}: {self.break_type.value}" def reconcile(a: Mapping[str, int], b: Mapping[str, int]) -> List[Break]: """Two independent records, compared. The bank habit worth adopting everywhere. A break is not an error to be swallowed — it is a *finding* with an owner and a due date. The value of reconciliation is not that it is clever; it is that it runs every day and somebody is accountable for the breaks. This is what would have caught every dual-write bug in this file. """ breaks: List[Break] = [] for key in sorted(set(a) | set(b)): left, right = a.get(key), b.get(key) if left is None: breaks.append(Break(BreakType.MISSING_IN_A, key, None, right)) elif right is None: breaks.append(Break(BreakType.MISSING_IN_B, key, left, None)) elif left != right: breaks.append(Break(BreakType.VALUE_MISMATCH, key, left, right)) return breaks # ====================================================================================== # Worked example # ====================================================================================== def _clock(start: int = 1000, step: int = 1) -> Callable[[], int]: state = {"t": start - step} def now() -> int: state["t"] += step return state["t"] return now GOOD_PAIN001 = f""" MSG-2026-0211-001 2026-02-11T09:15:00 2 18500.00 Falcon Trading LLC PMTINF-001 2026-02-12 Falcon Trading LLC AEPO Box 9021, Abu Dhabi AE070331234567890123456 NBADAEAA E2E-0001 12500.00 EBILAEAD Zenith Supplies FZEAE AE460090000000123456789 Invoice INV-2026-0044 E2E-0002 6000.00 Falcon Logistics LLC AE070331234567890123456 """ BAD_PAIN001 = f""" MSG-BAD 2026-02-11T09:15:00 3 PMTINF-BAD 12/02/2026 Falcon Trading LLC AE070331234567890123457 E2E-0001 12500.00 Zenith Supplies FZE AE460090000000123456789 E2E-0001 -5.00 """ def main() -> None: # pragma: no cover - narrative output print("=" * 78) print("1. READING AN ISO 20022 MESSAGE NAME") print("=" * 78) for name in ("pain.001.001.09", "pacs.008.001.10", "camt.053.001.08", "pacs.004.001.09"): mid = MessageIdentifier.parse(name) print(f" {name:<18} {mid.family:<34} {mid.description}") print(" -> the name is the map. pain = customer to bank, pacs = bank to bank,") print(" camt = cash management. Everything else is a lookup.") print() print("=" * 78) print("2. PARSING A VALID pain.001") print("=" * 78) parser = Pain001Parser() instruction, rejections = parser.parse(GOOD_PAIN001) print(f" rejections: {len(rejections)}") print(f" message : {instruction.message_id} ({instruction.number_of_transactions} txns)") print(f" debtor : {instruction.debtor.name} {instruction.debtor.iban}") for transfer in instruction.transfers: print(f" {transfer.end_to_end_id} " f"{from_minor(transfer.amount_minor, transfer.currency):>10} " f"{transfer.currency} -> {transfer.creditor.name}") print(f" control sum {from_minor(instruction.control_sum_minor, 'AED')} == " f"actual {from_minor(instruction.actual_sum_minor, 'AED')}") print() print("=" * 78) print("3. REJECTIONS NAME THE ELEMENT") print("=" * 78) instruction, rejections = parser.parse(BAD_PAIN001) for rejection in rejections: print(f" {rejection}") print(f" parsed: {instruction}") print(" -> every rejection at once, each naming its element. A parser that raises") print(" on the first problem makes a six-error file take six round trips, and") print(" in a bank each round trip is a day.") print() print("=" * 78) print("4. MINOR UNITS — THE CROSS-BORDER BUG") print("=" * 78) print(f" {'amount':<10} {'ccy':<5} {'exp':<4} {'minor units':>14} round-trip") for amount, ccy in (("100.00", "AED"), ("100", "JPY"), ("100.000", "KWD"), ("1.15", "USD")): minor = to_minor(amount, ccy) print(f" {amount:<10} {ccy:<5} {currency_exponent(ccy):<4} {minor:>14} " f"{from_minor(minor, ccy)}") print() print(f" the same conversion in float: 1.15 * 100 = {1.15 * 100!r}") print(f" int(1.15 * 100) = {int(1.15 * 100)} " f"<- one cent short, silently") print(f" round(2.675, 2) = {round(2.675, 2)} " f"<- not 2.68 either") print(" -> integers and Decimal, never float. And note JPY: ZERO minor digits,") print(" so assuming two everywhere makes a JPY payment 100x too large.") try: to_minor("100.001", "AED") except ValueError as exc: print(f" refusing extra precision: {exc}") print() print("=" * 78) print("5. FINALITY IS A SCHEME RULE, NOT A DATABASE PROPERTY") print("=" * 78) print(f" {'rail':<9} {'cut-off':<9} {'settles':<10} {'recall':<8} finality at 16:00") for rail in RAILS.values(): finality = rail.finality_at(submitted_minute=9 * 60, now_minute=16 * 60) cutoff = ("none" if rail.cutoff_minute >= 1440 else f"{rail.cutoff_minute // 60:02d}:00") print(f" {rail.rail_id:<9} {cutoff:<9} " f"{rail.settles_after_minutes:>4} min {str(rail.supports_recall):<8} " f"{finality.value}") print() for amount, ccy, minute, urgent in ((25_000_00, "AED", 10 * 60, True), (250_000_00, "AED", 10 * 60, True), (250_000_00, "AED", 16 * 60, False), (10_000_00, "USD", 10 * 60, False)): rail, why = choose_rail(amount, ccy, minute, urgent=urgent) print(f" {from_minor(amount, ccy):>12} {ccy} at {minute // 60:02d}:00 " f"urgent={str(urgent):<5} -> {rail or 'NO RAIL'} ({why})") print(" -> this is the link back to Phase 10's side-effect classes: 'irreversible'") print(" is not an engineering opinion about difficulty, it is what the scheme") print(" permits. The rail determines the finality, which determines whether a") print(" human must approve.") print() print("=" * 78) print("6. THE DUAL-WRITE PROBLEM") print("=" * 78) broker = MessageLog(partitions=2) broken = DualWriteStore(broker=broker, fail_after_db=True) try: broken.save_and_publish("PMT-1", {"status": "RELEASED"}, "payments") except RuntimeError as exc: print(f" crash: {exc}") print(f" database rows : {list(broken.rows)}") print(f" broker events : {[m.key for m in broker.all('payments')]}") print(" -> the payment happened and nothing downstream knows. Reversing the order") print(" is worse: a phantom event for a payment that does not exist.") print() print("=" * 78) print("7. THE TRANSACTIONAL OUTBOX") print("=" * 78) now = _clock() outbox = Outbox(now=now) log = MessageLog(partitions=3, now=now) relay = OutboxRelay(outbox=outbox, log=log, topic="payments") try: outbox.transact("PMT-9", {"status": "RELEASED"}, event_type="PaymentReleased", fail_before_commit=True) except RuntimeError as exc: print(f" crash before commit: {exc}") print(f" rows={list(outbox.rows)} outbox={len(outbox.records())} — NEITHER wrote") for pid in ("PMT-1", "PMT-2", "PMT-3"): outbox.transact(pid, {"status": "RELEASED"}, event_type="PaymentReleased", payload={"payment_id": pid, "amount_minor": 12_500_00}) print(f" 3 transactions -> rows={len(outbox.rows)} " f"unpublished={len(outbox.unpublished())}") relay.run_once(fail_after_publish=True) print(f" relay crashes after publishing 1 -> published={relay.published} " f"unpublished={len(outbox.unpublished())}") relay.run_once() print(f" relay runs again -> published={relay.published} " f"unpublished={len(outbox.unpublished())}") print(" -> PMT-1 was published TWICE. That is at-least-once, and it is the") print(" honest guarantee. The outbox promises no LOST events, not no duplicates.") print() print("=" * 78) print("8. ORDERING HOLDS WITHIN A PARTITION, NEVER ACROSS") print("=" * 78) accounts = ("ACC-A", "ACC-B", "ACC-C", "ACC-D") ordered = MessageLog(partitions=3) for i in range(3): for account in accounts: ordered.append("ledger", account, {"seq": i, "account": account}) for partition in range(ordered.partitions): messages = ordered.read("ledger", partition) keys = sorted({m.key for m in messages}) pairs = [f"{m.key[-1]}{m.value['seq']}" for m in messages] print(f" partition {partition}: keys={keys or '[]'} " f"log={pairs or '(empty)'}") print(f" routing: " + " ".join( f"{a}->p{ordered.partition_for(a)}" for a in accounts)) print() print(" read ACC-A's events in log order:") home = ordered.partition_for("ACC-A") print(f" {[m.value['seq'] for m in ordered.read('ledger', home) if m.key == 'ACC-A']}" f" <- monotone, guaranteed") print(" now read events for two accounts in DIFFERENT partitions and try to") print(" interleave them: there is no answer. Nothing in the system knows which") print(" of two events in different partitions happened first.") print(" -> keying by account is what makes per-account order hold. And note the") print(" collision: two accounts sharing a partition get a total order they did") print(" not ask for — which is fine until somebody starts relying on it and") print(" you repartition.") print() print("=" * 78) print("9. EXACTLY-ONCE EFFECTS UNDER DUPLICATE DELIVERY") print("=" * 78) applied: List[str] = [] group = ConsumerGroup("ledger-poster", log) consumer = IdempotentConsumer( group, lambda m: applied.append(f"{m.value['event_type']}:{m.value['seq']}"), key_of=lambda m: f"{m.key}:{m.value['seq']}") consumer.consume("payments") print(f" first pass : handled effects={len(applied)} duplicates={consumer.duplicates}") group.seek("payments", 0, 0) group.seek("payments", 1, 0) group.seek("payments", 2, 0) consumer.consume("payments") print(f" full replay: effects={len(applied)} duplicates={consumer.duplicates}") print(f" effects: {sorted(set(applied))}") print(" -> four deliveries (PMT-1 arrived twice), three effects. Exactly-once") print(" DELIVERY is impossible; exactly-once EFFECTS is a dict lookup.") print() print("=" * 78) print("10. COMPATIBILITY MODE IS DEPLOY ORDER") print("=" * 78) v1 = [Field("payment_id", "string"), Field("amount_minor", "int"), Field("currency", "string")] for mode in (Compatibility.BACKWARD, Compatibility.FORWARD, Compatibility.FULL): registry = SchemaRegistry(default_mode=mode) registry.register("payments.released", v1) print(f" {mode.value:<9} deploy: {registry.deploy_order('payments.released')}") for label, fields in ( ("add optional field", v1 + [Field("rail", "string", required=False)]), ("add required, no default", v1 + [Field("rail", "string")]), ("add required with default", v1 + [Field("rail", "string", default="rtgs")]), ("remove a required field", v1[:2]), ("change a type", [Field("payment_id", "string"), Field("amount_minor", "string"), Field("currency", "string")]), ): probe = SchemaRegistry(default_mode=mode) probe.register("payments.released", v1) try: probe.register("payments.released", fields) verdict = "OK" except ValueError as exc: verdict = f"REFUSED — {str(exc).split(': ', 1)[1]}" print(f" {label:<26} {verdict}") print(" -> BACKWARD means a NEW reader can read OLD data, so consumers upgrade") print(" first. Getting it backwards produces a release where nothing can read") print(" anything, discovered in production because staging deploys all at once.") print() print("=" * 78) print("11. DATA PRODUCTS HAVE CONTRACTS") print("=" * 78) product_clock = _clock(start=5000) schema = Schema("agent.traces", 1, ( Field("trace_id", "string"), Field("agent_id", "string"), Field("outcome", "string"), Field("cost_micros", "int"))) contract = DataProductContract( "agent.traces.daily", owner="layla.almansouri", schema=schema, freshness_slo_ticks=5, quality_rules=( QualityRule("non_empty", lambda rows: len(rows) > 0, "the product must not be empty"), QualityRule("no_negative_cost", lambda rows: all(r.get("cost_micros", 0) >= 0 for r in rows), "cost is never negative"), ), description="One row per completed agent task.") product = DataProduct(contract, now=product_clock) product.publish([ {"trace_id": "t1", "agent_id": "a1", "outcome": "success", "cost_micros": 4200}, {"trace_id": "t2", "agent_id": "a1", "outcome": "denied", "cost_micros": 900}, ]) print(f" {product.check().report()}") product.publish([{"trace_id": "t3", "agent_id": "a1", "outcome": "success"}]) print(f" {product.check().report()}") product.publish([ {"trace_id": "t4", "agent_id": "a1", "outcome": "success", "cost_micros": -1}]) print(f" {product.check().report()}") for _ in range(10): product_clock() print(f" {product.check().report()}") print(" -> the freshness SLO is what makes it a data PRODUCT rather than a table:") print(" a promise with a number, which can be breached, which means somebody") print(" can be told.") print() print("=" * 78) print("12. RECONCILIATION — THE HABIT WORTH STEALING") print("=" * 78) core_banking = {"PMT-1": 12_500_00, "PMT-2": 6_000_00, "PMT-4": 3_000_00} platform_log = {"PMT-1": 12_500_00, "PMT-2": 6_050_00, "PMT-3": 1_000_00} print(" A = core banking (the system of record), B = the platform's own log") for brk in reconcile(core_banking, platform_log): print(f" {brk}") print() print(" PMT-2 the amounts disagree — somebody rounded, or a partial applied") print(" PMT-3 the platform thinks it made a payment core banking never saw") print(" <- a PHANTOM: publish-then-write, the worse dual-write ordering") print(" PMT-4 core banking made a payment the platform never recorded") print(" <- a LOST event: write-then-publish, section 6's crash exactly") print(" -> a break is a FINDING with an owner and a due date, not an exception to") print(" swallow. Reconciliation is not clever; it runs every day and somebody") print(" is accountable, which is why it catches what monitoring misses.") if __name__ == "__main__": # pragma: no cover main()