From 79b67492111a1ce81ffbec864f91668607d5891d Mon Sep 17 00:00:00 2001 From: Jon Chery Date: Wed, 5 Aug 2026 16:16:55 +0000 Subject: [PATCH] =?UTF-8?q?docs(P05):=20complete=20examples=20+=20cross-li?= =?UTF-8?q?nks=20=E2=80=94=20v0.4?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit ---ci--- project: atelier phase: 5 milestone: v0.4 status: complete phase_role: execution phase_tag: v0.3.5 requirements: covered: [ATELIER-112, ATELIER-113, ATELIER-114] partial: [] audit: cross_links_verified: 57 cross_links_broken: 0 edge_messaging_bidirectional: resolved ---/ci--- --- examples/bad/edge-sync-loop.md | 158 ++++++++++++ examples/bad/messaging-shared-subscription.md | 188 +++++++++++++++ examples/good/edge-offline-sync.md | 223 +++++++++++++++++ .../good/messaging-idempotent-consumer.md | 226 ++++++++++++++++++ 4 files changed, 795 insertions(+) create mode 100644 examples/bad/edge-sync-loop.md create mode 100644 examples/bad/messaging-shared-subscription.md create mode 100644 examples/good/edge-offline-sync.md create mode 100644 examples/good/messaging-idempotent-consumer.md diff --git a/examples/bad/edge-sync-loop.md b/examples/bad/edge-sync-loop.md new file mode 100644 index 0000000..e896a51 --- /dev/null +++ b/examples/bad/edge-sync-loop.md @@ -0,0 +1,158 @@ +# Bad Example: Edge Sync Loop (Livelock) + +> A two-node sync that oscillates divergent state forever. This is +> the pre-specified `edge-sync-loop` chaos anti-pattern (ATELIER-110). +> **Single-breach per D-068:** the principle breached is **Edge P4 +> (Sync Conflicts are Bounded, Not Infinite)**. The prose explains +> why this is a livelock, not eventual consistency. + +## The Code + +```typescript +// Two edge nodes (a tablet and a dispatcher workstation) sync a +// work-order status. The merge uses wall-clock timestamps with no +// vector clock and no deterministic tiebreak. A clock skew flips +// the winner on every merge pass; the state oscillates forever. + +interface WorkOrderState { + orderId: string; + status: string; + wallClock: number; // P4 VIOLATION: wall time, not monotonic + nodeId: string; +} + +// P4 VIOLATION: the merge picks the later wall-clock write as the +// winner. Wall time skews across nodes; a skew of even a few +// milliseconds flips the winner. With no vector clock, concurrent +// writes are not detected; with no deterministic tiebreak, equal +// timestamps are resolved by whichever node's clock is ahead. +function mergeBad(local: WorkOrderState, remote: WorkOrderState): WorkOrderState { + // No vector clock. No monotonic logical clock. No deterministic + // tiebreak by nodeId. This is wall-clock-only LWW on durable + // state — the row in the CRDT-vs-LWW decision matrix that is a + // P4 violation for anything but ephemeral state. + if (local.wallClock >= remote.wallClock) return local; + return remote; +} + +// The sync loop: each node merges the other's state, writes the +// result, and the next sync pass flips it back. The loop never +// terminates. +async function syncLoop(node: "tablet" | "dispatcher", peer: WorkOrderState) { + let local = store.get(peer.orderId); + while (true) { + const merged = mergeBad(local, peer); + store.set(merged.orderId, merged); + await pushToPeer(merged); // peer receives, merges, pushes back + peer = await pullFromPeer(); // peer's clock is now ahead — flips winner + local = store.get(peer.orderId); // local re-merges; flips again + // The loop runs forever. Convergence is never reached. This is + // a livelock: the system is making progress (each pass writes) + // but the state never converges. + } +} +``` + +The tablet sets `status: "completed"` at wall time `10:00:00.500`. +The dispatcher's clock is 50ms ahead; it sets `status: "reassigned"` +at wall time `10:00:00.550`. The first merge: dispatcher wins +(`10:00:00.550 > 10:00:00.500`). The tablet receives `reassigned`, +but its clock drifts ahead by 100ms during the next sync pass; it +writes `completed` at `10:00:00.650`. The second merge: tablet +wins. The dispatcher's clock drifts ahead again; it writes +`reassigned` at `10:00:00.750`. The third merge: dispatcher wins. +The state flips between `completed` and `reassigned` on every sync +pass. The loop runs forever. + +## Why It Violates + +### Sync Conflicts are Bounded, Not Infinite (Edge P4, C1, C5) + +- **The breach:** the merge has no convergence bound. The CRDT-vs- + LWW decision matrix in `domains/edge/sync.md` is explicit: + wall-clock-only LWW on durable state carries an **unbounded + failure mode** — clock skew = oscillation — and is a P4 + violation for anything but ephemeral state. This code is the + matrix's failure column made real. +- **Why it is a livelock, not eventual consistency:** eventual + consistency guarantees that, in the absence of new writes, all + replicas eventually converge. This system never converges even + with no new writes: the clock skew alone drives the oscillation. + Each sync pass writes (so the system is "busy"), but the state + never settles — the definition of a livelock. The P4 contract is + that the merge terminates and converges; this merge does neither. +- The missing pieces, per the decision matrix: + - **No monotonic logical clock** (a hybrid logical clock or + vector clock) — wall time skews, and the skew is unbounded. + - **No vector clock** — concurrent writes are not detected, so + the conflict is invisible; the merge silently flips instead of + surfacing. + - **No deterministic tiebreak** (e.g., `nodeId`) on equal + timestamps — equal wall times are resolved by whichever node's + clock is ahead, which is not a stable property. +- The P4 bound requires that the merge function be associative, + commutative, and idempotent, terminating in one pass. A + wall-clock merge with clock skew satisfies none of these: it is + not associative (order of merges flips the winner), not + idempotent (a re-merge after a clock drift flips the result), + and not terminating (the loop runs forever). + +## The Fix + +```typescript +// Fix: replace wall-clock LWW with a CRDT register (state-based +// CvRDT) using a monotonic logical clock and a deterministic +// tiebreak by nodeId. The merge is now a total order — no +// oscillation (P4 bounded). See the CRDT-vs-LWW decision matrix: +// for a register on an unreliable network, a state-based CRDT is +// the correct row. + +interface LWWRegister { + value: T; + timestamp: number; // monotonic logical clock (HLC), not wall time + nodeId: string; // deterministic tiebreak +} + +function mergeLWWRegister( + local: LWWRegister, + remote: LWWRegister, +): LWWRegister { + // Associative, commutative, idempotent (P4 bound, P5 idempotent). + // (timestamp, nodeId) is a total order — convergence in one pass. + if (local.timestamp > remote.timestamp) return local; + if (local.timestamp < remote.timestamp) return remote; + return local.nodeId > remote.nodeId ? local : remote; +} + +// The sync is now one pass: merge, write, done. No loop, no +// oscillation. A re-merge of the same two replicas yields the same +// result (P5 idempotent), so a retried sync is safe. +async function syncOnce(local: LWWRegister, remote: LWWRegister) { + const converged = mergeLWWRegister(local, remote); + store.set(converged); + await pushToPeer(converged); + // Done. No while(true). The merge terminates. +} +``` + +The fix selects the CRDT row of the decision matrix (a register on +an unreliable network): a monotonic logical clock (HLC) eliminates +clock skew; a deterministic tiebreak by `nodeId` eliminates the +equal-timestamp flip. The merge is now a total order that converges +in one pass — the P4 bound. See `examples/good/edge-offline-sync.md` +for the full good-example version of this pattern. + +## Cross-Domain Links + +- `domains/edge/sync.md` — the CRDT-vs-LWW decision matrix; the + wall-clock-only LWW row is the failure mode this example + instantiates; the CRDT row is the fix. +- `domains/edge/first-principles.md` — P4 (Sync Conflicts are + Bounded, Not Infinite) is the principle breached. +- `review/anti-patterns.md` — the `edge-sync-loop` chaos + anti-pattern (edge P4, C1, C5 — infinite oscillation is a + correctness failure, not eventual consistency; the sync is a + livelock). +- `examples/good/edge-offline-sync.md` — the good-example version + of this pattern: a CRDT register merge that converges in one + pass. \ No newline at end of file diff --git a/examples/bad/messaging-shared-subscription.md b/examples/bad/messaging-shared-subscription.md new file mode 100644 index 0000000..93940ff --- /dev/null +++ b/examples/bad/messaging-shared-subscription.md @@ -0,0 +1,188 @@ +# Bad Example: Messaging Shared Subscription + +> Two consumers share one subscription; the broker dispatches each +> message to an arbitrary consumer. Per-consumer ordering breaks; +> per-consumer dedup is wrong. This is the pre-specified +> `messaging-shared-subscription` chaos anti-pattern (ATELIER-110). +> **Single-breach per D-068:** the primary principle breached is +> **Messaging P2 (Ordering is a Property, Not an Assumption)**. P3 +> (Consumers are Idempotent) is noted as the compounding +> consequence — the example remains single-breach in its named +> violation. + +## The Code + +```python +# Two email workers share ONE subscription on the "users" topic. +# The broker round-robins: worker A gets msg 1, msg 3; worker B +# gets msg 2, msg 4. Per-consumer order is broken (P2 breach). +# Each worker has its OWN dedup store — a redelivery to the other +# worker re-processes (P3 compounding). + +import json + +def shared_subscription_bad(broker, send_email, worker_id: str): + # P2 VIOLATION: both workers call subscribe with the SAME + # subscription name. The broker dispatches each message to an + # arbitrary worker in the shared group. Worker A sees msg 3 + # before worker B sees msg 1; per-worker order is broken. + sub = broker.subscribe(topic="users", subscription="welcome-shared") + + # P3 compounding: per-worker dedup. A redelivered message may + # land on the OTHER worker, which has not seen it, so the + # per-worker dedup store does not catch it — the message is + # processed twice across the two workers. + local_dedup = DedupStore(backend=redis_for(worker_id)) + + for message in sub.receive(): + payload = json.loads(message["body"]) + if local_dedup.seen(payload["idempotencyKey"]): + sub.ack(message); continue + send_email(payload["email"], "Welcome!") + local_dedup.mark(payload["idempotencyKey"]) + sub.ack(message) +``` + +```python +# The two workers are launched with the same subscription name. +# Worker A and Worker B both call subscribe("users", "welcome-shared"). +# The broker sees one shared subscription; it round-robins. + +def start_workers(broker): + # Worker A + spawn(shared_subscription_bad, broker, send_email, worker_id="A") + # Worker B — same subscription name, same topic + spawn(shared_subscription_bad, broker, send_email, worker_id="B") + # The broker dispatches: A gets msg 1, B gets msg 2, A gets msg 3, + # B gets msg 4. If worker A is slow, B may get msg 2 and msg 4 + # before A acks msg 1. Per-worker order is broken (P2). +``` + +The scenario: a user signs up, then immediately updates their +email preference (two messages in topic order: `user.signed-up.v1`, +`user.preference-updated.v1`). The broker dispatches `signed-up` to +worker A and `preference-updated` to worker B. Worker B sends the +preference email before worker A sends the welcome email — the +user sees the preference confirmation before the welcome. Then the +broker redelivers `signed-up` (worker A's ack was slow); this time +it dispatches to worker B. Worker B's local dedup store has never +seen `signed-up` (it was processed by A), so B sends the welcome +email again. The user receives two welcome emails. Per-consumer +ordering broke (P2); per-consumer dedup did not catch the +cross-worker redelivery (P3 compounding). + +## Why It Violates + +### Ordering is a Property, Not an Assumption (Messaging P2, C1, C2) + +- **The breach (primary):** the two workers share one subscription, + and the broker dispatches each message to an arbitrary worker in + the shared group. Per-consumer ordering is broken: worker A sees + `preference-updated` (msg 2) before it sees `signed-up` (msg 1) + if the broker round-robins them to different workers. The + workers' code assumes topic order, but the shared subscription + provides no per-consumer order guarantee — the broker's dispatch + is arbitrary. P2 requires that the ordering property be explicit + and documented; here it is assumed (topic order) but not provided + (arbitrary dispatch). The assumption is wrong. +- The P2 contract is that "it's FIFO" is a claim backed by the + broker's partitioning contract, not an assumption the consumer + makes. A shared subscription's contract is "no per-consumer + order"; the workers' code assumes the opposite. See + `domains/messaging/pubsub.md` (Shared vs Independent + Subscriptions) and `domains/messaging/first-principles.md` P2. + +### Consumers are Idempotent — compounding consequence (Messaging P3, C1) + +- **The compounding consequence (not the named breach):** each + worker has its OWN dedup store. A redelivered message that lands + on the *other* worker is not in that worker's dedup store, so it + is processed again. The dedup is per-worker, but the subscription + is shared — the dedup must be shared across workers to be + correct under a shared subscription. The per-worker dedup store + is wrong for a shared subscription; a shared dedup store (a + shared Redis, a shared DB) is required. +- Per D-068, the example remains single-breach in its named + violation: P2 is the primary breach (the shared subscription + breaks ordering); P3 is the compounding consequence (the + per-worker dedup is wrong *because* the subscription is shared). + If the subscription were independent, per-worker dedup would be + correct. The shared subscription is the root cause; P3 is the + downstream effect. + +## The Fix + +```python +# Fix 1 (default): independent subscriptions. Each consumer gets +# its own durable cursor; per-consumer order holds (P2); per- +# consumer dedup is correct (P3). This is the default per +# domains/messaging/pubsub.md. + +def independent_subscriptions_good(broker, send_email, worker_id: str): + # Each worker has its OWN subscription name. The broker + # delivers every message to every subscription in topic order. + sub = broker.subscribe( + topic="users", + subscription=f"welcome-{worker_id}", # per-consumer + ) + # Per-consumer dedup is now correct: a redelivery to THIS + # worker is caught by THIS worker's dedup store. + dedup = DedupStore(backend=redis_for(worker_id)) + for message in sub.receive(): + payload = json.loads(message["body"]) + if dedup.seen(payload["idempotencyKey"]): + sub.ack(message); continue + send_email(payload["email"], "Welcome!") + dedup.mark(payload["idempotencyKey"]) + sub.ack(message) +``` + +```python +# Fix 2 (if a shared subscription is genuinely required): the +# consumers must be stateless, the processing order-independent, +# AND the dedup store must be SHARED across workers. Document the +# ordering property as "none across consumers" (P2 — the property +# is explicit, not assumed) and use a shared dedup backend (P3). + +def shared_subscription_stateless(broker, send_email): + # P2: document the ordering property. A shared subscription + # provides NO per-consumer order; processing must be order- + # independent. Do not assume topic order. + sub = broker.subscribe(topic="users", subscription="welcome-shared") + # P3: SHARED dedup. A redelivery to any worker is caught by the + # shared store. + shared_dedup = DedupStore(backend=shared_redis) + for message in sub.receive(): + payload = json.loads(message["body"]) + if shared_dedup.seen(payload["idempotencyKey"]): + sub.ack(message); continue + send_email(payload["email"], "Welcome!") # order-independent + shared_dedup.mark(payload["idempotencyKey"]) + sub.ack(message) +``` + +The default is Fix 1 (independent subscriptions): per-consumer +ordering holds, per-consumer dedup is correct, and the code is +simpler. Fix 2 is the narrow opt-in for genuinely stateless, +order-independent processing — and even then, the dedup must be +shared. See `domains/messaging/pubsub.md` for the full +shared-vs-independent discussion. + +## Cross-Domain Links + +- `domains/messaging/pubsub.md` — the shared-vs-independent + subscriptions section; the `messaging-shared-subscription` + anti-pattern lives here (pre-specified in P4 ATELIER-110). +- `domains/messaging/first-principles.md` — P2 (Ordering is a + Property, Not an Assumption) is the primary breach; P3 + (Consumers are Idempotent) is the compounding consequence. +- `domains/messaging/delivery-semantics.md` — the idempotency-key + dedup store; the per-subscription dedup key scoping + (`(subscription, idempotencyKey)`) that prevents one + subscription's dedup from masking another's redelivery. +- `review/anti-patterns.md` — the `messaging-shared-subscription` + chaos anti-pattern (messaging P2, P3, C1 — shared subscription + breaks ordering and dedup). +- `examples/good/messaging-idempotent-consumer.md` — the good- + example version of the consumer pattern: an independent consumer + with a dedup store and a DLQ routing rule. \ No newline at end of file diff --git a/examples/good/edge-offline-sync.md b/examples/good/edge-offline-sync.md new file mode 100644 index 0000000..ee1b049 --- /dev/null +++ b/examples/good/edge-offline-sync.md @@ -0,0 +1,223 @@ +# Good Example: Edge Offline-First + Sync Reconcile + +> A field-service tablet app that operates through a 90-minute network +> partition, queues writes locally, and reconciles on reconnect using +> the CRDT-vs-LWW decision matrix from `domains/edge/sync.md`. Each +> aspect cites the edge principle it satisfies. + +## The Code + +```typescript +// The local store: an offline-first write queue + a CRDT register +// per record. The tablet keeps working through the partition; the +// queue drains on reconnect; the merge converges (P4 bounded). + +interface LWWRegister { + value: T; + timestamp: number; // monotonic logical clock (HLC), NOT wall time + nodeId: string; // tablet id — deterministic tiebreak (P4) +} + +interface QueuedWrite { + idempotencyKey: string; // P5: retried syncs are safe + collection: string; + recordId: string; + register: LWWRegister; + queuedAt: number; +} + +class OfflineStore { + private queue: QueuedWrite[] = []; + private state: Map> = new Map(); + private telemetryBuffer: TelemetryRecord[] = []; // P10 local-first + + // P2: write succeeds offline. The queue is the durable record of + // intent; sync is deferred, not blocked. + write(collection: string, recordId: string, value: unknown, key: string): void { + const register: LWWRegister = { + value, + timestamp: this.hlc.now(), // monotonic; no wall-clock skew + nodeId: this.nodeId, + }; + this.state.set(`${collection}:${recordId}`, register); + this.queue.push({ + idempotencyKey: key, // P5: idempotent sync + collection, recordId, register, + queuedAt: this.hlc.now(), + }); + // P10: buffer the write event locally; forwarded on reconnect. + this.telemetryBuffer.push({ event: "local.write", at: Date.now(), key }); + } + + // P2: read from local state offline. The UI never blocks on the + // network. + read(collection: string, recordId: string): unknown { + return this.state.get(`${collection}:${recordId}`)?.value; + } +} +``` + +```typescript +// The CRDT register merge — a state-based CvRDT (convergent). Per the +// CRDT-vs-LWW decision matrix in domains/edge/sync.md, a state-based +// CRDT is the choice when the data model fits a register and the +// network is unreliable (full-state merge tolerates dropped ops). + +function mergeLWWRegister( + local: LWWRegister, + remote: LWWRegister, +): LWWRegister { + // Associative, commutative, idempotent (P4 bound, P5 idempotent). + // (timestamp, nodeId) is a total order — no oscillation. + if (local.timestamp > remote.timestamp) return local; + if (local.timestamp < remote.timestamp) return remote; + return local.nodeId > remote.nodeId ? local : remote; // deterministic tie +} +``` + +```typescript +// The reconnect reconcile: drain the offline queue, pull remote +// state, three-way merge (CRDT registers), push converged state. +// Idempotent keys make a retried reconcile safe (P5). + +async function reconcile(store: OfflineStore, server: SyncServer): Promise { + // P5: the reconcile is idempotent. The idempotency key on each + // queued write means a retry (network flapped mid-reconcile) does + // not double-apply. + const queued = store.drainQueue(); + let pushed = 0, merged = 0, conflicts = 0; + + // 1. Push local writes. The server dedups by idempotencyKey (P5). + for (const w of queued) { + await server.applyWrite(w.idempotencyKey, w.collection, w.recordId, w.register); + pushed++; + } + + // 2. Pull remote state for every record we touched + every record + // the server changed since our last sync cursor. Merge via CRDT. + const remoteRecords = await server.fetchChanged(store.syncCursor()); + for (const [key, remoteReg] of remoteRecords) { + const localReg = store.localRegister(key); + if (localReg) { + // P4: CRDT merge converges. The merge is a pure function of + // two inputs; the total order (timestamp, nodeId) guarantees + // no oscillation. + const converged = mergeLWWRegister(localReg, remoteReg); + if (converged !== localReg) { conflicts++; } + store.setLocal(key, converged); + merged++; + } else { + store.setLocal(key, remoteReg); // remote-only record + merged++; + } + } + + // 3. P10: flush the buffered telemetry. The partition did not blind + // the operator — the events survived on-node. + store.flushTelemetry(); + + return { pushed, merged, conflicts, converged: true }; +} +``` + +```typescript +// P10: local-first telemetry. Events are buffered on-node and +// forwarded on reconnect. A fire-and-forget pipeline loses data when +// the link drops; a local-first buffer survives. + +interface TelemetryRecord { event: string; at: number; key: string; } + +class TelemetryBuffer { + private buffer: TelemetryRecord[] = []; + + push(rec: TelemetryRecord): void { this.buffer.push(rec); } + + // Called from reconcile() on reconnect. The buffer is the P10 + // guarantee: the operator sees the partition-window activity, + // not a gap. + async flush(sink: TelemetrySink): Promise { + for (const rec of this.buffer) { await sink.emit(rec); } + this.buffer = []; + } + + depth(): number { return this.buffer.length; } +} +``` + +## The Scenario + +A field-service tablet is dispatched to a basement site with no +cellular coverage. The technician updates the work-order status +(`started`, `parts-ordered`, `completed`) five times over 90 +minutes. Each write lands in the local store immediately — the UI +never blocks on the network. The writes are queued with an +idempotency key (the work-order id + a monotonic sequence). + +When the tablet reconnects, the reconcile drains the queue: the +server dedups by idempotency key (a retry mid-reconcile does not +double-apply). The server also returns a remote update — the +dispatcher re-assigned the work order to a different technician at +minute 45, then reverted at minute 60. The CRDT merge converges: +the register with the later logical timestamp wins; the tiebreak by +node id is deterministic. The merge terminates in one pass (P4 +bounded); it does not oscillate between the dispatcher's revert and +the technician's status updates. The buffered telemetry flushes, +and the operator sees the full partition-window activity — no gap. + +## Principles Demonstrated + +### Offline is a First-Class State (Edge P2, C1, C5) +- The tablet writes and reads through the partition. The UI never + blocks on the network; the offline write queue is the durable + record of intent. Partition is the norm, not the exception; + reconciliation happens on reconnect. An app that crashes on + disconnect has no offline state; this app engineers it. +- See `domains/edge/offline-first.md` (offline write-queue) and + `domains/edge/first-principles.md` P2. + +### Sync Conflicts are Bounded, Not Infinite (Edge P4, C1, C5) +- The CRDT register merge is associative, commutative, and + idempotent. The total order `(timestamp, nodeId)` guarantees + convergence in one pass — no oscillation, no infinite loop. This + is the P4 bound: the merge terminates. The CRDT-vs-LWW decision + matrix in `domains/edge/sync.md` selected a state-based CRDT + because the data model fits a register and the network is + unreliable (full-state merge tolerates dropped ops). +- See `domains/edge/sync.md` (CRDT-vs-LWW decision matrix, merge + semantics) and `domains/edge/first-principles.md` P4. + +### Edge Operations are Idempotent (Edge P5, C1) +- Every queued write carries an idempotency key; the server dedups + by key. A reconcile retried mid-flap does not double-apply. The + merge function is idempotent (`merge(a, a) === a`) — a retried + merge of the same two replicas yields the same result. Sync, the + retried-by-nature operation, is safe. +- See `domains/edge/sync.md` (Merge Semantics, P5) and + `domains/edge/first-principles.md` P5. + +### Edge Observability Survives Partition (Edge P10, C7, C5) +- Telemetry is buffered on-node (`telemetryBuffer`) and flushed on + reconnect. The partition did not blind the operator — the + partition-window activity is forwarded, not lost. A + fire-and-forget pipeline would have a 90-minute gap; the + local-first buffer survives. +- See `domains/edge/offline-first.md` (local-first logging) and + `domains/edge/first-principles.md` P10. + +## Cross-Domain Links + +- `domains/edge/offline-first.md` — the offline write-queue pattern + this app instantiates; the partition-as-norm discipline. +- `domains/edge/sync.md` — the CRDT-vs-LWW decision matrix exercised + here (state-based CvRDT chosen for register + unreliable network); + the merge-semantics properties (associative, commutative, + idempotent). +- `domains/edge/first-principles.md` — P2, P4, P5, P10 are the + principles demonstrated. +- `domains/concurrency/patterns` — the in-process immutability + analog (`concurrency/P1 Immutability by Default`) that makes the + merge a pure function of two inputs. +- `domains/observability/metrics` — the generic SLI/SLO discipline + the local-first telemetry buffer builds on. +- `review/anti-patterns.md` — the `edge-sync-loop` chaos + anti-pattern is the inverse of this example's bounded CRDT merge. \ No newline at end of file diff --git a/examples/good/messaging-idempotent-consumer.md b/examples/good/messaging-idempotent-consumer.md new file mode 100644 index 0000000..443e5aa --- /dev/null +++ b/examples/good/messaging-idempotent-consumer.md @@ -0,0 +1,226 @@ +# Good Example: Idempotent Consumer with Dedup + DLQ + +> An orders-consumer that achieves exactly-once-via-idempotency: +> at-least-once delivery plus a TTL-bounded dedup store and a DLQ +> routing rule. Each aspect cites the messaging principle it +> satisfies. Exercises the idempotency + DLQ guidance in +> `domains/messaging/delivery-semantics.md`. + +## The Code + +```python +# The idempotency-key dedup store (P3). TTL-bounded (P6): a dedup +# store with no TTL is a memory leak. The TTL exceeds the broker's +# max-redelivery window; beyond it, the key is expired (the broker +# has given up). + +import time, json + +DEDUP_TTL_SECONDS = 24 * 3600 # > broker max-redelivery window + +class DedupStore: + """P3 (idempotent), P6 (TTL-bounded). seen() before process; + mark() after process; the order gives at-least-once + dedup.""" + + def __init__(self, backend): + # backend is Redis or a shared DB. MUST be shared across + # consumer instances (see messaging/pubsub.md on shared vs + # independent subscriptions). + self.backend = backend + + def seen(self, key: str) -> bool: + ts = self.backend.get(key) + if ts is None: + return False + if time.time() - ts > DEDUP_TTL_SECONDS: + self.backend.delete(key) # P6: expired; not a redelivery + return False + return True + + def mark(self, key: str): + self.backend.set(key, time.time(), ttl=DEDUP_TTL_SECONDS) +``` + +```python +# The idempotent consumer. Order: dedup BEFORE process, mark AFTER +# process, ack AFTER mark. A crash before mark re-processes (the +# dedup store lacks the key); a crash before ack redelivers and the +# dedup store makes the redelivery a no-op (P3). For a +# non-idempotent process (a payment that must not double-charge), +# process+mark are one DB transaction — exactly-once via +# idempotency (P4). + +def consume_orders(broker, dedup: DedupStore, process_order): + for message in broker.receive(): + payload = json.loads(message["body"]) + + # P3: dedup BEFORE process. A redelivered message is a + # no-op, not a double-apply. + if dedup.seen(payload["idempotencyKey"]): + broker.ack(message) # already processed; skip + continue + + try: + # P4: the declared semantic is at-least-once + idempotent + # dedup = exactly-once-via-idempotency. For a payment, + # process_order + dedup.mark run in one DB transaction + # so the mark commits iff the process commits. + process_order(payload) + dedup.mark(payload["idempotencyKey"]) + broker.ack(message) + + except TransientError as exc: + # P6: bounded retry with backoff. Nack for redelivery; + # the broker redelivers after exponential backoff. + broker.nack(message, delay=backoff(payload.get("attempt", 0))) + + except (ValueError, SchemaError) as exc: + # P5: poison message — unparseable. Route immediately, + # do NOT retry (no retry will fix a bad schema). + route_to_dlq(broker, message, exc, kind="poison") + broker.ack(message) + + except PermanentError as exc: + # P5: permanent failure (e.g., not-found dependency). + # Retry will not fix it — DLQ now. + route_to_dlq(broker, message, exc, kind="dlq") + broker.ack(message) +``` + +```python +# The DLQ routing rule (P5 dead-letter handling, P10 DLQ depth +# alert). Distinguishes poison (unparseable; never retried) from +# DLQ (exhausted retry budget on a transient). Both carry audit +# metadata; both emit a depth metric. + +DLQ = "orders-dlq" +POISON = "orders-poison" +MAX_RETRY_TTL_SECONDS = 30 * 60 # 30 min retry window + +def route_to_dlq(broker, message, reason, kind: str): + target = POISON if kind == "poison" else DLQ + broker.send(target, body=json.dumps({ + "original": message["body"], + "reason": str(reason), + "kind": kind, # poison vs dlq + "deadLetteredAt": now_iso(), + "redeliveryCount": message.get("attempt", 0), + })) + # P10: emit a metric so DLQ depth alerts fire. A DLQ that grows + # with no alert is a silent correctness defect (P5/P10). + metrics.increment(f"{kind}.depth", tags={"queue": "orders"}) + + +def consume_with_retry_budget(broker, dedup, process_order): + # Combines TTL-with-backoff for transient failures (P6) with + # poison-queue + DLQ + alert (P5/P10). + for message in broker.receive(): + payload = json.loads(message["body"]) + if dedup.seen(payload["idempotencyKey"]): + broker.ack(message); continue + + first_attempt_ts = payload.get("firstAttemptTs", time.time()) + attempt = payload.get("attempt", 0) + + try: + process_order(payload) + dedup.mark(payload["idempotencyKey"]) + broker.ack(message) + + except TransientError as exc: + # P6: if the retry window is exhausted, route to DLQ; + # otherwise redeliver with exponential backoff. + if time.time() - first_attempt_ts > MAX_RETRY_TTL_SECONDS: + route_to_dlq(broker, message, exc, kind="dlq") # P5 + broker.ack(message) + else: + broker.nack(message, delay=backoff(attempt)) +``` + +## The Scenario + +An orders queue delivers `order.created` events to the consumer at +**at-least-once** (the declared semantic, P4). The broker redelivers +on consumer crash or ack-timeout. Three things happen: + +1. **Normal delivery** — the consumer dedups by `idempotencyKey`, + processes, marks, acks. +2. **Redelivery after a crash before ack** — the consumer crashed + after `mark` but before `ack`. The broker redelivers; `seen()` + returns true; the consumer acks without re-processing (P3). +3. **Poison message** — a malformed JSON body. The consumer routes + it to the poison queue immediately (no retry will fix a parse + error), acks the origin, and emits a `poison.depth` metric. The + operator is paged on poison-queue growth (P10). + +A transient downstream failure (the payments API is briefly 503) +retries with exponential backoff for 30 minutes (P6); if it exceeds +the budget, the message routes to the DLQ with `reason`, +`redeliveryCount`, and `deadLetteredAt` — auditable, drainable, +observable (P5). The DLQ depth metric alerts the operator; the DLQ +entry's audit metadata lets the operator replay after the bug is +fixed (P5 reversibility). + +## Principles Demonstrated + +### Consumers are Idempotent (Messaging P3, C1) +- The consumer dedups by idempotency key before processing. A + redelivered message is a no-op, not a double-apply. The + `process → mark → ack` order gives at-least-once + idempotent + dedup; for a non-idempotent process, `process + mark` are one DB + transaction (exactly-once via idempotency, P4). +- See `domains/messaging/delivery-semantics.md` (idempotency-key + dedup store) and `domains/messaging/first-principles.md` P3. + +### Delivery Semantics are Explicit (Messaging P4, C1, C2) +- The channel is declared **at-least-once + idempotent consumer** — + the engineering practice that collapses to exactly-once under + correct dedup (P3). The semantic is not emergent; it is the + declared choice per channel. The tradeoff (dedup-store cost, + transactional-process complexity) is conscious and documented. +- See `domains/messaging/queues.md` (the three-semantics comparison + table) and `domains/messaging/delivery-semantics.md` (exactly- + once via idempotency). + +### Dead-Letter Handling is Defined (Messaging P5, C1, C5) +- Poison messages (unparseable) route immediately to the poison + queue — no retry will fix them. Transient failures retry with + backoff until the TTL, then route to the DLQ. Both carry audit + metadata (`reason`, `redeliveryCount`, `deadLetteredAt`); both are + drainable and observable. The DLQ is the reversibility mechanism + — a dead-lettered message can be reprocessed after the bug is + fixed. +- See `domains/messaging/delivery-semantics.md` (dead-letter + strategy comparison table, DLQ routing rule) and + `domains/messaging/first-principles.md` P5. + +### Messaging is Observable (Messaging P10, C7, C1) +- DLQ depth and poison-queue depth are emitted as metrics and wired + to alerts. A DLQ that grows silently is a correctness defect; + the alert makes it visible. Silent backlog is a bug, not a + feature — the operator is paged on growth, not on a customer + report. +- See `domains/messaging/delivery-semantics.md` (DLQ depth as an + alert) and `domains/messaging/first-principles.md` P10. + +## Cross-Domain Links + +- `domains/messaging/delivery-semantics.md` — the idempotency-key + dedup-store pattern and the dead-letter strategy comparison table + exercised here (TTL-with-backoff + poison-queue + DLQ + alert). +- `domains/messaging/queues.md` — the three-semantics comparison + table; the ack/nack/redelivery model this consumer uses. +- `domains/messaging/first-principles.md` — P3, P4, P5, P10 are the + principles demonstrated. +- `domains/concurrency/patterns` — the in-process retry/backoff + analog (Pattern 6, Timeout on Every Block); messaging owns the + broker-backed instance where redelivery comes across a network. +- `domains/errors/patterns` — errors as data: a DLQ entry is the + async-messaging instance of an error log (observable, auditable, + drainable). +- `domains/observability/metrics` — the generic SLI/SLO discipline + the DLQ-depth alert builds on. +- `review/anti-patterns.md` — the `messaging-unbounded-retry` and + `messaging-shared-subscription` chaos anti-patterns are the + inverse of this example's bounded retry + independent-consumer + discipline. \ No newline at end of file