coreci/atelier
3
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases 28 Wiki Activity

docs(P05): complete examples + cross-links — v0.4

Browse Source 
---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---
This commit is contained in:
Jon Chery
2026-08-05 16:16:55 +00:00
parent c7f46bc0cc
commit bb74dc5e4c
4 changed files with 795 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files
examples/bad/edge-sync-loop.md
+158
View File
@@ -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<T> {
value: T;
timestamp: number; // monotonic logical clock (HLC), not wall time
nodeId: string; // deterministic tiebreak
}
function mergeLWWRegister<T>(
local: LWWRegister<T>,
remote: LWWRegister<T>,
): LWWRegister<T> {
// 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<unknown>, remote: LWWRegister<unknown>) {
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.
examples/bad/messaging-shared-subscription.md
+188
View File
@@ -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.
examples/good/edge-offline-sync.md
+223
View File
@@ -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<T> {
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<unknown>;
queuedAt: number;
}
class OfflineStore {
private queue: QueuedWrite[] = [];
private state: Map<string, LWWRegister<unknown>> = 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<unknown> = {
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<T>(
local: LWWRegister<T>,
remote: LWWRegister<T>,
): LWWRegister<T> {
// 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<ReconcileReport> {
// 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<void> {
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.
examples/good/messaging-idempotent-consumer.md
+226
View File
@@ -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.