# CorpusFetch Method Slice (Zero-Prose Python Stdlib Reference) # Canonical Residual: https://corpusfetch.com/c/what-message-ordering-invariants-must-survive-when-the-transport-lacks-ordering-guarantees.md # Remnant Warrant: https://corpusfetch.com/c/what-message-ordering-invariants-must-survive-when-the-transport-lacks-ordering-guarantees.warrant.json # Markdown SHA-256: a0418876e2b0b5eef7565821a18644da028883aa98ddee66f4e8eb0f17cdd791 # Code AST SHA-256: 7ffbef9a9328642924fe7726e4b6d3245d2a5f5d74545efd1b7c8bd2d236ef5c # Verified Classes: OrderingVerdict, OrderedMessage, ReceiverState, OrderingPolicy, OrderingManifest, CausalOrderingGuard # Defensive Guards: 0 fail-closed invariants (zero external deps) from dataclasses import dataclass from enum import Enum, auto import hashlib import json class OrderingVerdict(Enum): PROCESS_IMMEDIATELY = auto() BUFFER_WAIT_FOR_ANTECEDENT = auto() ESCALATE_BROKEN_CHAIN = auto() ORDER_INDEPENDENT = auto() @dataclass(frozen=True) class OrderedMessage: message_id: str logical_clock: int causal_dependencies: frozenset[str] order_independent: bool @dataclass(frozen=True) class ReceiverState: processed_message_ids: frozenset[str] buffered_messages: tuple[str, ...] current_epoch: float buffer_start_epoch: dict[str, float] @dataclass(frozen=True) class OrderingPolicy: max_buffer_age_sec: float current_epoch: float @dataclass(frozen=True) class OrderingManifest: verdict: OrderingVerdict directive: str manifest_digest: str class CausalOrderingGuard: @staticmethod def _hash(data: str) -> str: return hashlib.sha256(data.encode("utf-8")).hexdigest() @classmethod def evaluate_message( cls, msg: OrderedMessage, state: ReceiverState, policy: OrderingPolicy ) -> OrderingManifest: # Gate 1: Order independent if msg.order_independent: digest = cls._hash(json.dumps({ "message": msg.message_id, "status": "ORDER_INDEPENDENT" }, sort_keys=True)) return OrderingManifest( verdict=OrderingVerdict.ORDER_INDEPENDENT, directive=f"PROCEED: Message {msg.message_id} is order-independent.", manifest_digest=digest ) # Gate 2: Antecedents present missing = msg.causal_dependencies - state.processed_message_ids if not missing: digest = cls._hash(json.dumps({ "message": msg.message_id, "clock": str(msg.logical_clock), "antecedents": tuple(sorted(msg.causal_dependencies)) }, sort_keys=True)) return OrderingManifest( verdict=OrderingVerdict.PROCESS_IMMEDIATELY, directive=f"PROCEED: All antecedents present for {msg.message_id}.", manifest_digest=digest ) # Gate 3: Buffer with timeout if msg.message_id in state.buffer_start_epoch: age = policy.current_epoch - state.buffer_start_epoch[msg.message_id] if age > policy.max_buffer_age_sec: return OrderingManifest( verdict=OrderingVerdict.ESCALATE_BROKEN_CHAIN, directive=f"HALT: Message {msg.message_id} buffered {age:.0f}s > {policy.max_buffer_age_sec:.0f}s. Chain broken.", manifest_digest=cls._hash(f"BROKEN:{msg.message_id}") ) # Gate 4: Buffer digest = cls._hash(json.dumps({ "message": msg.message_id, "missing": tuple(sorted(missing)) }, sort_keys=True)) return OrderingManifest( verdict=OrderingVerdict.BUFFER_WAIT_FOR_ANTECEDENT, directive=f"BUFFER: {msg.message_id} waiting for {sorted(missing)}.", manifest_digest=digest )