paulpas Distributed Systems Architecture
Implements distributed systems patterns including consensus algorithms, consistency models, data replication strategies, partitioning schemes, clock synchronization mechanisms, and distributed transaction orchestrations to build correct and resilient multi-node architectures.
TL;DR Checklist
- Define the CAP tradeoff for each service before choosing a consistency model
- Use consistent hashing with virtual nodes for any sharding or partitioning scheme
- Implement vector clocks (not physical clocks) for causal event ordering across nodes
- Replace 2PC with Saga orchestration for cross-service transactions in production microservices
- Treat every network call as unreliable — add timeouts, retries with exponential backoff, and circuit breakers
- Assign unique idempotency keys to all mutable operations (commands, message delivery, HTTP requests)
- Design the outbox pattern for reliable event publication from database transactions
- Implement service discovery with heartbeats, health checks, and automatic stale-node removal
- Document consistency guarantees per service in architecture decision records
When to Use
Use this skill when:
- Designing a new multi-service or multi-node system where nodes communicate over a network
- Choosing between consistency models (strong, eventual, causal) for data storage layers
- Implementing leader election or consensus among cluster members (e.g., config management, distributed locks)
- Building a sharded database or partitioned key-value store that must handle node churn gracefully
- Synchronizing events across services and needing causal ordering guarantees without physical clock trust
- Coordinating transactions that span multiple services or databases where ACID is unavailable
- Implementing service discovery, health checking, or dynamic cluster membership management
When NOT to Use
Avoid this skill for:
- Single-node applications with no distributed coordination needs — use standard concurrency patterns instead
- Read-heavy caches that only need TTL-based expiration — Redis or Memcached handles this without complex consensus
- Simple request-response APIs with a single database backend — microservices overhead outweighs benefit
- Real-time systems requiring sub-millisecond deterministic latency — use in-memory locks on a single process
Core Workflow
Map the Failure Model — Document which failures your system must tolerate: node crashes, network partitions (CAP), clock drift, message duplication/reordering. Write each failure scenario as an explicit assumption. Checkpoint: Every inter-service boundary has at least one documented failure mode and a corresponding mitigation pattern.
Choose Consistency per Service — For each data store or service boundary, select the consistency model: strong (linearizable), causal, or eventual. Base this on whether the operation needs to prevent double-spending (strong) versus whether stale reads are tolerable (eventual). Checkpoint: Each service has a documented consistency guarantee in its API contract.
Design Partitioning Strategy — For any horizontally scaled data store, implement consistent hashing with virtual nodes (at least 150–300 vnodes per physical node) using CRC32 or MurmurHash3. Compute hash ring positions, then verify that adding/removing a node redistributes fewer than
total_keys / num_nodeskeys on average. Checkpoint: The rebalance ratio stays below 1/num_nodes when a single node is added or removed.Implement Causal Ordering — Where events from multiple nodes must be ordered causally (not necessarily globally), implement vector clocks with per-node monotonic counters. Provide
happens_before(),concurrent_with(), andmerge()operations. For global ordering requirements, use a timestamp-based total order service. Checkpoint: Every event carries its vector clock; downstream consumers can detect and resolve concurrent updates.Select Transaction Coordination — For cross-service transactions, prefer the Saga pattern (choreography or orchestration) over 2PC. Implement compensating actions for each step that roll back the logical effect. Reserve 2PC only for single-database distributed transaction scenarios where you control all participating resources. Checkpoint: Each saga step has an idempotent execute and a compensating rollback; the orchestrator handles timeouts and retries with exponential backoff.
Build Reliability Primitives — For each inter-service communication path, implement: (a) request-level timeouts (connect ≤ 1s, read ≤ 3s), (b) circuit breaker with half-open state after recovery probe, (c) retry with jittered exponential backoff (base delay 100ms, max 5 retries, max delay 30s). Apply the outbox pattern for event publication. Checkpoint: No single service can hang the entire cluster through synchronous blocking on a downstream call.
When NOT to Use (Anti-Patterns)
- Using distributed consensus for a single-node deployment — Raft/Paxos add 2–3x latency and operational complexity for zero fault tolerance.
- Relying on NTP-correct clocks for ordering critical events — Clock drift of ±500ms is common even with NTP; use vector clocks instead.
- Implementing "eventually consistent" without reconciliation — Without explicit conflict resolution or backfill, divergence accumulates silently until data corruption is detected in production.
Implementation Patterns / Reference Guide
Pattern 1: Consistent Hashing with Virtual Nodes
Consistent hashing maps both keys and nodes onto a circular hash ring. When a node joins or leaves, only keys near that node's position need to move — minimizing data redistribution. Virtual nodes (multiple ring positions per physical node) ensure even key distribution regardless of hash collisions between node identifiers.
This is the foundation of sharding in Dynamo-style systems, Cassandra, and consistent load balancers.
"""Consistent Hashing Ring with Virtual Nodes for distributed partitioning."""
from __future__ import annotations
import hashlib
import bisect
from typing import Iterable
MODULAR_HASH_SPACE = 2**32
def _crc32_hash(key: str) -> int:
"""Compute CRC32-style hash of a string, returning unsigned 32-bit integer."""
return int(hashlib.crc32(key.encode("utf-8")) & 0xFFFFFFFF)
class ConsistentHashRing:
"""Consistent hashing ring with virtual node distribution.
Each physical node is placed at `virtual_nodes` evenly-spaced positions
around the hash ring to ensure uniform key distribution.
Args:
virtual_nodes: Number of virtual nodes per physical node (default 150).
Higher values produce more uniform distribution but use more memory.
hasher: Optional custom hash function taking a string and returning int.
Defaults to CRC32 over the modular hash space.
"""
def __init__(
self,
virtual_nodes: int = 150,
hasher: callable | None = None,
) -> None:
if virtual_nodes < 1:
raise ValueError("virtual_nodes must be >= 1")
self._virtual_nodes = virtual_nodes
self._hasher = hasher or (_crc32_hash)
self._ring: list[int] = [] # sorted hash positions
self._node_map: dict[int, str] = {} # position -> node_id
self._nodes: set[str] = set() # physical node IDs
# ------------------------------------------------------------------
# Public API
# ------------------------------------------------------------------
def add_node(self, node_id: str) -> None:
"""Add a physical node to the ring, distributing it across virtual nodes."""
if node_id in self._nodes:
raise KeyError(f"Node {node_id!r} is already present in the ring")
self._nodes.add(node_id)
for i in range(self._virtual_nodes):
vnode_key = f"{node_id}:vnode_{i}"
pos = self._hasher(vnode_key) % MODULAR_HASH_SPACE
bisect.insort(self._ring, pos)
self._node_map[pos] = node_id
def remove_node(self, node_id: str) -> None:
"""Remove a physical node from the ring and all its virtual nodes."""
if node_id not in self._nodes:
raise KeyError(f"Node {node_id!r} is not in the ring")
# Rebuild the ring without this node (simplest correct approach).
# For production systems with frequent churn, maintain a reverse index.
old_ring = list(self._ring)
old_map = dict(self._node_map)
self._ring.clear()
self._node_map.clear()
for pos in old_ring:
if old_map[pos] != node_id:
bisect.insort(self._ring, pos)
self._node_map[pos] = old_map[pos]
self._nodes.discard(node_id)
def get_node(self, key: str) -> str | None:
"""Find the responsible node for a given key (clockwise from key's hash)."""
if not self._ring:
return None
pos = self._hasher(key) % MODULAR_HASH_SPACE
idx = bisect.bisect_right(self._ring, pos)
# Wrap around the ring (modulo for circularity)
if idx >= len(self._ring):
idx = 0
return self._node_map[self._ring[idx]]
def get_nodes(self, key: str, replicas: int = 3) -> list[str]:
"""Return the first `replicas` distinct nodes responsible for a key."""
if not self._ring:
return []
assigned: list[str] = []
seen: set[str] = set()
pos = self._hasher(key) % MODULAR_HASH_SPACE
idx = bisect.bisect_right(self._ring, pos)
while len(assigned) < replicas and len(seen) < len(self._nodes):
actual_idx = idx % len(self._ring)
node = self._node_map[self._ring[actual_idx]]
if node not in seen:
assigned.append(node)
seen.add(node)
idx += 1
return assigned
def get_distribution_stats(self) -> dict[str, int]:
"""Return key counts per node (for verification of uniformity)."""
stats: dict[str, int] = {n: 0 for n in self._nodes}
if not self._ring:
return stats
# Distribute a large sample of keys and count assignments
num_samples = 10_000
for i in range(num_samples):
node = self.get_node(f"key_{i}")
if node is not None:
stats[node] += 1
return stats
@property
def node_count(self) -> int:
return len(self._nodes)
@property
def vnode_count(self) -> int:
return len(self._ring)
Verification — Uniform Distribution Check:
def verify_distribution_uniformity(ring: ConsistentHashRing, tolerance_pct: float = 5.0) -> bool:
"""Assert that keys are distributed within `tolerance_pct` of perfect uniformity."""
stats = ring.get_distribution_stats()
if not stats:
return True
counts = list(stats.values())
ideal = sum(counts) / len(counts)
for node, count in stats.items():
deviation = abs(count - ideal) / ideal * 100
if deviation > tolerance_pct:
raise AssertionError(
f"Node {node}: distribution deviation {deviation:.1f}% exceeds "
f"tolerance {tolerance_pct}%"
)
return True
Pattern 2: Vector Clocks for Causal Ordering
Physical clocks drift and cannot reliably establish causal order across distributed nodes. Vector clocks provide a logical timestamp scheme where each node maintains a counter for every other node it has communicated with. The happens-before relationship (VC_a < VC_b) can be computed by comparing component-wise maximums.
"""Vector Clocks for causal ordering of events in distributed systems."""
from __future__ import annotations
import copy
from typing import Protocol
class NodeId:
"""Opaque identifier for a node in the distributed system."""
def __init__(self, value: str) -> None:
self._value = value
@property
def value(self) -> str:
return self._value
def __eq__(self, other: object) -> bool:
if not isinstance(other, NodeId):
return NotImplemented
return self._value == other._value
def __hash__(self) -> int:
return hash(self._value)
def __repr__(self) -> str:
return f"NodeId({self._value!r})"
class VectorClock:
"""Causal clock maintaining per-node logical timestamps.
Each component tracks the last known event count from that node.
Supports happens-before, concurrency detection, and merge operations.
Args:
clocks: Dict mapping NodeId to integer counter values.
If None, starts with all-zero counters for every observed node.
"""
__slots__ = ("_clocks",)
def __init__(self, clocks: dict[NodeId, int] | None = None) -> None:
self._clocks: dict[NodeId, int] = dict(clocks) if clocks else {}
# ---------------------------------------------------------------
# Mutation
# ---------------------------------------------------------------
def increment(self, node_id: NodeId) -> None:
"""Advance this node's own clock component by one.
Must be called before processing any local event at the given node.
"""
self._clocks[node_id] = self._clocks.get(node_id, 0) + 1
def merge(self, other: "VectorClock") -> None:
"""Merge with another vector clock by taking component-wise maximum.
Used when a node receives a message carrying the sender's vector clock.
After merging, this node has learned about all events known to the other.
"""
for nid, ts in other._clocks.items():
self._clocks[nid] = max(self._clocks.get(nid, 0), ts)
# ---------------------------------------------------------------
# Comparison operators (causal ordering)
# ---------------------------------------------------------------
def happens_before(self, other: "VectorClock") -> bool:
"""Return True if this clock is causally before `other`.
Definition: VC_a < VC_b iff all components of a <= b AND at least one
component of a < b. In other words, 'other' has learned about strictly
more events than 'this'.
"""
all_keys = set(self._clocks.keys()) | set(other._clocks.keys())
at_least_one_less = False
for k in all_keys:
a_val = self._clocks.get(k, 0)
b_val = other._clocks.get(k, 0)
if a_val > b_val:
return False
if a_val < b_val:
at_least_one_less = True
return at_least_one_less
def is_concurrent(self, other: "VectorClock") -> bool:
"""Return True if this clock and `other` are causally independent.
Two events are concurrent when neither happens-before the other —
meaning they occurred on different branches of execution that haven't
yet been merged. Concurrent events require explicit conflict resolution.
"""
return not self.happens_before(other) and not other.happens_before(self) \
and self._clocks != other._clocks
def is_equal(self, other: "VectorClock") -> bool:
"""Return True if both clocks represent the same causal state."""
all_keys = set(self._clocks.keys()) | set(other._clocks.keys())
return all(
self._clocks.get(k, 0) == other._clocks.get(k, 0)
for k in all_keys
)
# ---------------------------------------------------------------
# Inspectors
# ---------------------------------------------------------------
def total_order(self) -> int:
"""Sum of all components — useful for approximate global ordering.
WARNING: Total order is NOT a substitute for happens-before comparison.
Two events can have the same total but be concurrent, or different totals
with no causal relationship. Use only for tie-breaking in display/logic,
never for correctness-critical ordering.
"""
return sum(self._clocks.values())
def copy(self) -> "VectorClock":
"""Return a deep copy of this vector clock."""
return VectorClock(copies=self._clocks.copy())
def __repr__(self) -> str:
parts = ", ".join(f"{n.value}:{v}" for n, v in sorted(
self._clocks.items(), key=lambda x: x[0].value
))
return f"VC({parts})"
def __eq__(self, other: object) -> bool:
if not isinstance(other, VectorClock):
return NotImplemented
return self.is_equal(other)
Usage — Handling Concurrent Updates (Conflict Resolution):
def resolve_concurrent_conflicts(
local_vc: VectorClock,
remote_vc: VectorClock,
local_value: str,
remote_value: str,
conflict_resolver: callable | None = None,
) -> tuple[str, VectorClock]:
"""Resolve a write-write conflict caused by concurrent updates.
Args:
local_vc: Vector clock of the local node's latest state.
remote_vc: Vector clock from the peer node.
local_value: The value held at the local node.
remote_value: The value received from the peer.
conflict_resolver: Optional callable(local_value, remote_value) -> str.
If None, use last-writer-wins based on NodeId.
Returns:
Tuple of (resolved_value, merged_vector_clock).
"""
merged_vc = local_vc.copy()
merged_vc.merge(remote_vc)
if not local_vc.is_concurrent(remote_vc):
# No conflict — causal order is clear; pick the one that happened after.
if remote_vc.happens_before(local_vc):
return local_value, merged_vc
else:
return remote_value, merged_vc
# True concurrent update — need explicit resolution strategy.
if conflict_resolver is not None:
resolved = conflict_resolver(local_value, remote_value)
else:
# Default: last-writer-wins by lexicographic NodeId comparison
# (In production, use a Lamport-style tie-breaker with unique IDs.)
resolved = max(local_value, remote_value)
return resolved, merged_vc
Pattern 3: Outbox Pattern for Reliable Message Delivery
The outbox pattern guarantees that every business transaction is accompanied by a message publication event. By storing both the database mutation and the corresponding event in the same ACID transaction, you ensure zero message loss even if the publisher crashes after the DB commit but before sending the message. A separate publisher process polls the outbox table and delivers events, retrying on failure.
"""Transactional Outbox Pattern for reliable event publication."""
from __future__ import annotations
import uuid
import time
import logging
from dataclasses import dataclass, field
from datetime import datetime, timezone
from typing import Any
from enum import Enum
logger = logging.getLogger(__name__)
class EnvelopeStatus(str, Enum):
PENDING = "pending"
DELIVERED = "delivered"
FAILED = "failed"
@dataclass(frozen=True)
class OutboxEvent:
"""Immutable record of an event to be published from the outbox table.
Attributes:
id: Unique event ID for deduplication on the consumer side.
aggregate_type: Domain entity type that triggered the event (e.g., 'order').
aggregate_id: Primary key of the affected domain entity.
event_type: Semantic event name (e.g., 'OrderCreated', 'PaymentCompleted').
payload: JSON-serializable event data.
created_at: Timestamp when the event was committed to the outbox.
status: Current delivery status.
version: Monotonic sequence number for ordering within the same aggregate.
"""
id: str
aggregate_type: str
aggregate_id: str
event_type: str
payload: dict[str, Any]
created_at: datetime = field(default_factory=lambda: datetime.now(timezone.utc))
status: EnvelopeStatus = EnvelopeStatus.PENDING
version: int = 0
class TransactionalOutbox:
"""Manages the outbox table — events are committed within DB transactions.
This class simulates a database-backed outbox. In production, replace the
in-memory storage with actual SQL (INSERT into outbox_events table) or
an ORM call within the same explicit transaction as your business data mutation.
The key invariant: every business-state commit has its corresponding outbox
row committed atomically. The publisher reads PENDING rows independently.
"""
def __init__(self) -> None:
self._events: dict[str, OutboxEvent] = {}
# ---------------------------------------------------------------
# Core: Commit with business transaction (atomic with business data)
# ---------------------------------------------------------------
def commit_event(
self,
aggregate_type: str,
aggregate_id: str,
event_type: str,
payload: dict[str, Any],
version: int = 0,
) -> OutboxEvent:
"""Atomically store an event in the outbox alongside a business transaction.
In production, this runs inside the same DB transaction as your UPDATE/INSERT
on the business table. The publisher then reads these rows separately.
Args:
aggregate_type: Domain entity type (e.g., 'order', 'invoice').
aggregate_id: ID of the affected entity.
event_type: Semantic event name for routing/consumption.
payload: JSON-serializable event body.
version: Per-aggregate sequence number to enforce ordering.
Returns:
The committed OutboxEvent with its unique ID and status.
"""
event_id = str(uuid.uuid4())
event = OutboxEvent(
id=event_id,
aggregate_type=aggregate_type,
aggregate_id=aggregate_id,
event_type=event_type,
payload=payload,
version=version,
)
self._events[event_id] = event
return event
# ---------------------------------------------------------------
# Publisher: Read and deliver pending events
# ---------------------------------------------------------------
def fetch_pending(self, batch_size: int = 50) -> list[OutboxEvent]:
"""Retrieve up to `batch_size` PENDING events for publication."""
pending = [
e for e in self._events.values()
if e.status == EnvelopeStatus.PENDING
]
# Sort by creation time, then by version within same aggregate
pending.sort(key=lambda e: (e.aggregate_type, e.aggregate_id, e.created_at))
return pending[:batch_size]
def mark_delivered(self, event_id: str) -> None:
"""Mark an event as successfully delivered to its consumers."""
if event_id not in self._events:
raise KeyError(f"Event {event_id!r} not found in outbox")
self._events[event_id].status = EnvelopeStatus.DELIVERED
def mark_failed(self, event_id: str) -> None:
"""Mark an event as failed (will be retried on next fetch)."""
if event_id not in self._events:
raise KeyError(f"Event {event_id!r} not found in outbox")
self._events[event_id].status = EnvelopeStatus.FAILED
# ---------------------------------------------------------------
# Cleanup (archived delivered events)
# ---------------------------------------------------------------
def purge_delivered(self, older_than: datetime) -> int:
"""Remove delivered events older than `older_than` to reclaim storage.
Returns the number of events purged. Run periodically (e.g., nightly).
Do NOT purge before consumers have had time to process and acknowledge.
"""
before = len([e for e in self._events.values() if e.status == EnvelopeStatus.DELIVERED])
to_delete = [
eid for eid, e in self._events.items()
if e.status == EnvelopeStatus.DELIVERED and e.created_at < older_than
]
for eid in to_delete:
del self._events[eid]
after = len([e for e in self._events.values() if e.status == EnvelopeStatus.DELIVERED])
return before - after
Pattern 4: Leader Election (Raft Conceptual Implementation)
Leader election provides a single point of decision for log ordering. In Raft, nodes cycle through FOLLOWER, CANDIDATE, and LEADER states. A candidate requests votes from peers; if it receives a majority (N/2 + 1), it becomes leader and begins sending heartbeat messages to maintain authority.
"""Conceptual Leader Election — Raft-style with majority quorum."""
from __future__ import annotations
import enum
import time
import threading
from dataclasses import dataclass, field
from typing import Optional
class NodeState(str, enum.Enum):
FOLLOWER = "follower"
CANDIDATE = "candidate"
LEADER = "leader"
@dataclass
class ElectionTerm:
"""Raft term (monotonically increasing election round)."""
number: int
leader_id: Optional[str] = None
def __post_init__(self) -> None:
if self.number < 0:
raise ValueError("term number must be non-negative")
class Node:
"""Represents a Raft-style node in the cluster.
In production, each Node instance runs on a separate machine/process.
This example simulates message passing in-memory for demonstration.
Args:
node_id: Unique identifier for this node.
peer_ids: IDs of all other nodes in the cluster (including self).
"""
def __init__(self, node_id: str, peer_ids: list[str]) -> None:
self._node_id = node_id
self._peer_ids = [pid for pid in peer_ids if pid != node_id]
self._state = NodeState.FOLLOWER
self._term = ElectionTerm(number=0)
self._voted_for: Optional[str] = None
self._last_heartbeat = time.time()
self._election_timeout_ms = 150 + hash(node_id) % 150 # randomized timeout
self._lock = threading.Lock()
@property
def node_id(self) -> str:
return self._node_id
@property
def state(self) -> NodeState:
return self._state
@property
def term(self) -> ElectionTerm:
return self._term
def start_election(self, cluster_nodes: list["Node"]) -> bool:
"""Begin leader election: increment term, transition to CANDIDATE, vote for self.
Args:
cluster_nodes: All nodes in the cluster (for message simulation).
Returns:
True if this node won the election and became LEADER.
"""
with self._lock:
old_term = self._term.number
self._term = ElectionTerm(number=old_term + 1, leader_id=None)
self._state = NodeState.CANDIDATE
self._voted_for = self._node_id
votes_received = 1 # vote for self
total_nodes = len(cluster_nodes)
majority = (total_nodes // 2) + 1
for peer in cluster_nodes:
if peer.node_id == self._node_id:
continue
result = peer.receive_vote_request(self._term)
if result:
votes_received += 1
if votes_received >= majority:
with self._lock:
self._state = NodeState.LEADER
self._term.leader_id = self._node_id
return True
else:
with self._lock:
self._state = NodeState.FOLLOWER
return False
def receive_vote_request(self, incoming_term: ElectionTerm) -> bool:
"""Handle a vote request from a CANDIDATE.
Votes are granted if the candidate's term is >= current term AND
this node hasn't voted for another candidate in this term.
"""
with self._lock:
if incoming_term.number > self._term.number:
self._term = ElectionTerm(number=incoming_term.number, leader_id=None)
self._state = NodeState.FOLLOWER
self._voted_for = None
if incoming_term.number >= self._term.number and (
self._voted_for is None or self._voted_for == self._node_id
):
# For simplicity, grant vote if term is higher and not already voted.
# In production Raft, also check log up-to-date-ness.
if self._voted_for is None:
self._voted_for = self._node_id # Would be candidate's ID in real impl
return True
return False
def receive_heartbeat(self) -> bool:
"""Receive a heartbeat from the current LEADER.
Resets election timer and validates term consistency.
"""
with self._lock:
now = time.time()
if now - self._last_heartbeat > self._election_timeout_ms / 1000:
# Timer expired — should start new election in a background goroutine.
return False
self._last_heartbeat = now
return True
def get_current_leader(self) -> Optional[str]:
"""Return the current known leader ID, or None if no leader exists."""
with self._lock:
if self._state == NodeState.LEADER:
return self._node_id
if self._term.leader_id:
return self._term.leader_id
return None
def elect_leader(cluster: list[Node]) -> str | None:
"""Run leader election across a simulated cluster.
Tries each node as candidate in order; the first to win becomes LEADER.
Returns the winning leader's ID or None if no leader could be elected.
"""
for node in cluster:
if node.start_election(cluster):
return node.node_id
return None
Pattern 5: Saga Orchestration with Compensation
The Saga pattern breaks a distributed transaction into a sequence of local transactions, each with a compensating action that reverses its effect. An orchestrator coordinates the saga steps and executes compensations when a step fails. This avoids the synchronous blocking nature of 2PC while still providing ACID-like guarantees at the business level.
"""Saga Orchestration — multi-step distributed transaction with compensation."""
from __future__ import annotations
import logging
from dataclasses import dataclass, field
from enum import Enum
from typing import Callable, Any
logger = logging.getLogger(__name__)
class SagaStatus(str, Enum):
RUNNING = "running"
COMPLETED = "completed"
COMPENSATING = "compensating"
FAILED = "failed"
@dataclass
class SagaStep:
"""A single step in a saga with its forward and compensating actions.
Args:
name: Human-readable step identifier (e.g., 'reserve_inventory').
execute: Function that performs the business action. Returns result dict or raises on failure.
compensate: Function that reverses the effect of `execute`. Receives the original result.
Should be idempotent — safe to call multiple times.
"""
name: str
execute: Callable[..., dict[str, Any]]
compensate: Callable[[dict[str, Any]], None]
class SagaOrchestrator:
"""Orchestrates a saga by executing steps sequentially and compensating on failure.
Args:
saga_id: Unique identifier for this saga instance (used for tracing/logging).
"""
def __init__(self, saga_id: str) -> None:
self._saga_id = saga_id
self._status = SagaStatus.RUNNING
self._completed_steps: list[tuple[str, dict[str, Any]]] = [] # (name, result)
self._error: Exception | None = None
@property
def status(self) -> SagaStatus:
return self._status
@property
def saga_id(self) -> str:
return self._saga_id
def execute_saga(self, steps: list[SagaStep]) -> bool:
"""Execute all saga steps in order. On any failure, compensate backwards.
Args:
steps: List of SagaStep definitions to execute sequentially.
Returns:
True if the saga completed successfully (all steps executed).
False if compensation was needed or final status is FAILED.
"""
try:
for step in steps:
logger.info(
"[Saga %s] Executing step: %s",
self._saga_id, step.name,
)
result = step.execute()
self._completed_steps.append((step.name, result))
except Exception as exc:
self._error = exc
self._status = SagaStatus.COMPENSATING
logger.error(
"[Saga %s] Step failed: %s — starting compensation. Error: %s",
self._saga_id, step.name, exc,
)
self._compensate_failed(steps)
self._status = SagaStatus.FAILED
return False
self._status = SagaStatus.COMPLETED
return True
def _compensate_failed(self, steps: list[SagaStep]) -> None:
"""Compensate completed steps in reverse order.
Compensates only the steps that finished execution before the failure point.
Each compensation action is wrapped in try/except to prevent one failed
compensation from aborting the entire rollback sequence.
"""
for step_name, result in reversed(self._completed_steps):
# Find the corresponding step definition
step_def = next((s for s in steps if s.name == step_name), None)
if step_def is None:
logger.warning(
"[Saga %s] No step definition found for compensation of %s",
self._saga_id, step_name,
)
continue
try:
logger.info(
"[Saga %s] Compensating step: %s",
self._saga_id, step_name,
)
step_def.compensate(result)
except Exception as comp_exc:
# Compensation itself failed — log and continue.
# In production, this needs manual intervention / dead-letter queue.
logger.error(
"[Saga %s] COMPENSATION FAILED for step %s: %s",
self._saga_id, step_name, comp_exc,
)
def get_result(self) -> dict[str, Any]:
"""Return summary of saga execution for monitoring/tracing."""
return {
"saga_id": self._saga_id,
"status": self._status.value,
"steps_completed": len(self._completed_steps),
"error": str(self._error) if self._error else None,
}
Example — Ordering Saga:
def example_order_saga() -> bool:
"""Demonstrate a complete order-fulfillment saga."""
# --- Business state (simulated) ---
inventory: dict[str, int] = {"widget": 100}
payments: list[dict[str, Any]] = []
shipments: list[dict[str, Any]] = []
def reserve_inventory(qty: int) -> dict[str, Any]:
if inventory["widget"] < qty:
raise ValueError(f"Insufficient inventory: need {qty}, have {inventory['widget']}")
inventory["widget"] -= qty
return {"item": "widget", "quantity_reserved": qty}
def cancel_inventory_reservation(reservation: dict[str, Any]) -> None:
item = reservation["item"]
qty = reservation["quantity_reserved"]
inventory[item] = inventory.get(item, 0) + qty
def process_payment(amount: float) -> dict[str, Any]:
payment_id = f"pay_{len(payments) + 1}"
payments.append({"id": payment_id, "amount": amount})
return {"payment_id": payment_id}
def cancel_payment(payment: dict[str, Any]) -> None:
# Refund logic — in production, call payment gateway API.
pass
def create_shipment(order_data: dict[str, Any]) -> dict[str, Any]:
shipment_id = f"ship_{len(shipments) + 1}"
shipments.append({"id": shipment_id, "order": order_data})
return {"shipment_id": shipment_id}
def cancel_shipment(shipment: dict[str, Any]) -> None:
# Mark shipment as cancelled in logistics system.
pass
# --- Build saga steps ---
steps = [
SagaStep(name="reserve_inventory", execute=lambda: reserve_inventory(1), compensate=cancel_inventory_reservation),
SagaStep(name="process_payment", execute=lambda: process_payment(29.99), compensate=cancel_payment),
SagaStep(name="create_shipment", execute=lambda: create_shipment({"item": "widget"}), compensate=cancel_shipment),
]
# --- Execute ---
orchestrator = SagaOrchestrator(saga_id="order_saga_001")
result = orchestrator.execute_saga(steps)
print(orchestrator.get_result())
return result
Pattern 6: Service Discovery with Heartbeats
Service discovery enables dynamic service registration, health checking, and automatic deregistration of failed nodes. This pattern is essential for load balancers, client-side routing, and cloud-native architectures where instances scale up and down frequently.
"""Simple service registry with heartbeats and automatic stale-node removal."""
from __future__ import annotations
import time
import threading
from dataclasses import dataclass, field
from typing import Optional
@dataclass
class ServiceInstance:
"""Represents a running instance of a service in the registry.
Attributes:
instance_id: Unique identifier for this instance (e.g., UUID).
service_name: Logical service name (e.g., 'payment-service').
host: IP address or hostname.
port: Listening port number.
metadata: Arbitrary key-value metadata (version, zone, capabilities).
registered_at: Timestamp when the instance was first registered.
last_heartbeat: Timestamp of the most recent heartbeat received.
health_status: Current health state.
"""
instance_id: str
service_name: str
host: str
port: int
metadata: dict[str, str] = field(default_factory=dict)
registered_at: float = field(default_factory=time.time)
last_heartbeat: float = 0.0
health_status: str = "healthy"
class ServiceRegistry:
"""In-memory service registry with heartbeat-based stale-node removal.
Args:
stale_timeout_seconds: Seconds without a heartbeat before marking an instance stale.
Default 30s — adjust based on your deployment's expected latency.
"""
def __init__(self, stale_timeout_seconds: float = 30.0) -> None:
self._stale_timeout = stale_timeout_seconds
self._instances: dict[str, ServiceInstance] = {}
self._service_index: dict[str, list[str]] = {} # service_name -> [instance_ids]
self._lock = threading.Lock()
def register(self, instance: ServiceInstance) -> None:
"""Register a new service instance. Overwrites existing registration by instance_id."""
with self._lock:
old_service = self._instances.get(instance.instance_id)
if old_service:
# Remove from old service index if service name changed
if old_service.service_name != instance.service_name:
self._unindex_instance(old_service)
self._instances[instance.instance_id] = instance
self._service_index.setdefault(instance.service_name, []).append(instance.instance_id)
def deregister(self, instance_id: str) -> Optional[ServiceInstance]:
"""Remove a service instance. Returns the removed instance or None."""
with self._lock:
instance = self._instances.pop(instance_id, None)
if instance is not None:
self._unindex_instance(inst
…(truncated)