Distributed Systems : A Master Guide for AI Infrastructure
Distributed systems are what you get when one machine is no longer enough.
Maybe one GPU cannot train your model.
Maybe one server cannot handle all inference requests.
Maybe your vector database contains terabytes of embeddings.
Maybe thousands of robots are continuously sending telemetry.
Maybe an AI application has dozens of services:
User
↓
API Gateway
↓
Authentication
↓
Agent Service
↓
LLM Inference Cluster
↓
Vector Database
↓
Tool Services
↓
Message Queue
↓
Background Workers
↓
Object Storage
Once these components live on different machines, a strange thing happens:
Your program stops behaving like one program.
Machines disagree about time.
Messages arrive late.
Messages arrive twice.
Machines crash halfway through operations.
Network connections disappear.
Two servers can both believe the other one is dead.
Databases may temporarily disagree.
A request may succeed even though the client believes it failed.
A leader may crash exactly while coordinating an important operation.
Distributed systems engineering is largely the study of how to build correct and useful systems despite those realities.
For AI engineers, this knowledge becomes increasingly important because modern AI infrastructure is inherently distributed:
distributed model training
multi-GPU inference
model-serving clusters
vector databases
distributed feature stores
event pipelines
data lakes
agent orchestration
GPU schedulers
distributed robotics fleets
telemetry platforms
recommendation systems
search infrastructure
cloud-native applications
The purpose of this chapter is not to memorize algorithms such as Paxos.
The goal is to develop a mental model that allows you to reason about systems when machines, networks, clocks, and processes fail.
1. What Is a Distributed System?
A distributed system is a collection of independent computers cooperating to provide some larger service.
For example:
┌─────────────┐
│ Load Balancer│
└──────┬──────┘
│
┌─────────────┼─────────────┐
↓ ↓ ↓
Server A Server B Server C
│ │ │
└─────────────┼─────────────┘
↓
Distributed DB
To the user this might appear to be a single service.
Internally, however, there may be hundreds or thousands of machines.
That creates the fundamental problem of distributed computing:
There is no perfectly reliable shared memory, shared clock, or reliable communication channel connecting all machines.
Everything must happen through messages.
And messages can:
arrive
arrive late
arrive twice
arrive out of order
be lost
Meanwhile machines can:
slow down
pause
restart
crash
lose network connectivity
recover with stale state
Distributed systems are therefore systems built under uncertainty.
2. The Three Hard Problems
Most distributed-system problems come from three realities.
2.1 Machines fail independently
Suppose you have:
Service A → Service B → Database
Service A may be running.
Service B may have crashed.
The database may be fine.
Or B may be fine but the network between A and B may be broken.
Failure is no longer binary.
2.2 Networks are unreliable
A normal function call is approximately:
result = function()
You know whether the call returned.
A network call behaves differently:
result = remote_service()
What if no response comes back?
Possible explanations include:
1. Request never reached the server.
2. Request reached the server but execution failed.
3. Request succeeded but response was lost.
4. Server is simply slow.
5. Network is temporarily partitioned.
The caller cannot always tell which happened.
This ambiguity is one of the most important concepts in distributed systems.
2.3 There is no perfect global clock
Every machine has its own clock.
Machine A: 10:00:00.100
Machine B: 10:00:00.073
Machine C: 10:00:00.141
They are close.
But not identical.
That sounds harmless until correctness depends on deciding which event happened first.
3. Physical Clocks
Computers maintain physical clocks using hardware oscillators.
Ideally:
1 second on machine A
=
1 second on machine B
In reality, oscillators drift.
Suppose:
Server A clock: 12:00:00.500
Server B clock: 12:00:00.450
A receives an event at:
12:00:00.480
B receives another event at:
12:00:00.470
According to timestamps, B's event happened first.
But because the clocks differ, that conclusion may be wrong.
This is called clock skew.
Over time clocks also drift at different rates:
Clock A ─────────────────────>
Clock B ────────────────────>
Clock C ──────────────────────>
This is clock drift.
4. Clock Synchronization
Machines periodically synchronize their clocks.
Common mechanisms include:
NTP — Network Time Protocol
PTP — Precision Time Protocol
GPS-based clock synchronization
NTP may ask another machine:
"What time is it?"
and adjust the local clock.
But network latency makes perfect synchronization impossible.
Suppose:
A → request → B
takes 20 ms.
And:
B → response → A
takes 80 ms.
A cannot perfectly determine what B's clock was when the response arrived.
Therefore distributed systems generally assume:
Physical clocks are approximate, not absolute truth.
For logging and monitoring, wall-clock timestamps are extremely useful.
For proving event causality, they are often insufficient.
That leads to logical clocks.
5. Logical Clocks
Logical clocks answer a different question.
Instead of asking:
What exact physical time did this happen?
they ask:
What happened before what?
This relationship is called happened-before.
If event A causes event B:
A → B
then:
A happened-before B
Written mathematically:
A → B
6. Lamport Logical Clocks
Leslie Lamport introduced a simple logical-clock system.
Every process maintains a counter.
Example:
Process A
A1 A2 A3
1 → 2 → 3
Every local event increments the counter.
When sending a message:
A counter = 5
A ───── message(timestamp=5) ─────> B
When B receives it:
B = max(B_local, received_timestamp) + 1
Suppose:
B local clock = 3
message timestamp = 5
Then:
B = max(3,5)+1
= 6
Now the logical ordering preserves causality.
If:
A → B
then:
L(A) < L(B)
But the reverse does not necessarily hold.
If:
L(A) < L(B)
you cannot conclude that A caused B.
Two independent events can still receive ordered numbers.
7. Vector Clocks
Vector clocks preserve more causality information.
Suppose we have three nodes:
A
B
C
Each maintains a vector:
[A, B, C]
Initially:
[0,0,0]
When A performs an operation:
[1,0,0]
B independently performs one:
[0,1,0]
Neither vector dominates the other.
Therefore the events are concurrent.
This is useful in systems where different replicas can accept writes independently.
Vector clocks help distinguish:
A caused B
from:
A and B occurred independently
They appear conceptually in systems such as distributed key-value stores and conflict-resolution mechanisms.
8. Ordering Is More Important Than Time
A powerful distributed-systems insight is:
Often we do not need to know when something happened. We need to know what order operations must appear to have happened in.
Consider:
deposit $100
withdraw $50
Correct order matters.
Exact nanosecond timestamps usually do not.
Many distributed algorithms therefore focus on establishing a shared ordering of events.
That brings us to consensus.
9. Consensus
Consensus means several machines agreeing on something.
For example:
Which machine is the leader?
What is log entry #1024?
Was transaction X committed?
What configuration is active?
Suppose five nodes maintain a replicated log:
Node A
Node B
Node C
Node D
Node E
They need to agree that:
Entry 41 = SET model_version = v17
Consensus algorithms allow them to reach that agreement despite some failures.
This sounds simple.
It is not.
Consensus is one of the deepest problems in distributed computing.
10. Majority / Quorum
Many consensus systems rely on a majority.
For five nodes:
majority = 3
If three agree:
A ✓
B ✓
C ✓
D ?
E ?
the operation can often be considered committed.
Why majority?
Because two independent majorities must overlap.
For a five-node cluster:
Majority group 1:
A B C
Majority group 2:
C D E
At least one member overlaps.
That overlap helps prevent contradictory decisions.
11. Raft
Raft was designed to make consensus easier to understand than earlier algorithms such as Paxos.
A Raft cluster usually contains nodes in three possible states:
Follower
Candidate
Leader
Normally:
Leader
/ | \
↓ ↓ ↓
Follower Follower Follower
The leader handles client writes and replicates them to followers.
12. Raft Terms
Raft divides time into logical periods called terms.
Example:
Term 1
Term 2
Term 3
Term 4
Each term can have at most one elected leader.
Conceptually:
Term 1: Node A leader
Term 2: Node C leader
Term 3: Node C leader
Term 4: Node B leader
Terms help nodes recognize stale information.
If a node receives a message from an older term, it can reject it.
13. Raft Leader Election
Followers expect periodic heartbeat messages from the leader.
Leader ─heartbeat→ Followers
If a follower receives no heartbeat for some randomized timeout:
Follower → Candidate
The candidate:
increments its term
votes for itself
asks other nodes for votes
If it gets a majority:
Candidate → Leader
Randomized election timeouts reduce the chance that all nodes become candidates simultaneously.
14. Raft Log Replication
Suppose the client sends:
SET model = v4
The leader appends it:
Leader log:
1 A
2 B
3 SET model=v4
Then replicates it:
Leader
↓
Follower A
↓
Follower B
Once a majority has stored the entry, the leader can consider it committed.
Eventually followers execute the committed operation.
Raft therefore creates a shared replicated history.
15. Paxos
Paxos solves essentially the same fundamental consensus problem.
It is historically foundational and theoretically extremely important.
Its roles are often described as:
Proposers
Acceptors
Learners
A simplified Paxos idea is:
Phase 1: Prepare / Promise
Phase 2: Accept / Accepted
Nodes attach increasing proposal numbers to proposals.
Acceptors promise not to accept older proposals after seeing a newer one.
A majority quorum determines the chosen value.
The actual theory is subtle.
For working infrastructure knowledge, remember:
Paxos is a family of quorum-based consensus protocols used to ensure that distributed nodes agree on a value despite failures.
Raft emphasizes understandability.
Paxos emphasizes the underlying consensus theory.
You do not normally implement either yourself unless building infrastructure such as a database, coordination system, or distributed control plane.
16. Consensus Is Not Replication
These two ideas are related but different.
Replication asks:
How do I keep multiple copies of data?
Consensus asks:
How do multiple nodes agree on which state or sequence of operations is authoritative?
You can replicate data without strongly coordinating every write.
For example:
Region US → replica
Region EU → replica
Region Asia → replica
Depending on the system, replication may be:
synchronous
asynchronous
leader-based
leaderless
17. Replication
Why replicate data?
Three major reasons:
Availability
Durability
Performance
If one database disappears:
Replica A ✗
Replica B ✓
Replica C ✓
the system can continue.
Replication can also place data near users:
USA users → USA replica
Europe users → Europe replica
Asia users → Asia replica
reducing network latency.
18. Leader/Follower Replication
One common architecture is:
Writes
↓
Leader
/ | \
↓ ↓ ↓
Follower Follower Follower
Clients normally send writes to the leader.
The leader replicates changes to followers.
Reads may come from either:
leader only
or:
followers too
depending on the consistency requirements.
19. Synchronous vs Asynchronous Replication
Synchronous
Leader waits for replicas:
Client
↓
Leader
↓
Follower
↓
Acknowledgment
↓
Client success
Advantage:
stronger durability
Disadvantage:
higher latency
Asynchronous
Leader responds immediately:
Client
↓
Leader
↓
SUCCESS
Then later:
Leader → Followers
Advantage:
lower latency
Disadvantage:
A leader can crash before replication completes.
Some acknowledged writes may therefore disappear during failover.
Distributed systems constantly trade among:
latency
consistency
availability
durability
20. Replication Lag
Followers are often slightly behind the leader.
Suppose:
Leader:
x = 10
Follower:
x = 7
The follower has not yet received the latest write.
A user might perform:
WRITE x=10
READ x
and surprisingly receive:
x=7
if the read hits a stale replica.
This is why databases offer different consistency guarantees.
21. Network Partitions
Imagine:
A ───── B ───── C
Now the network link breaks:
A ───── B X C
All machines are still running.
But communication is impossible between some of them.
This is a network partition.
Perhaps:
Partition 1:
A B
Partition 2:
C D E
Both sides may believe the other side failed.
Neither side can know for certain.
This creates difficult decisions.
Should both partitions continue accepting writes?
Should one side stop?
This is the context behind CAP.
22. CAP Theorem
CAP describes behavior during network partitions.
The letters mean:
C — Consistency
A — Availability
P — Partition tolerance
Partition tolerance means the system continues operating despite communication failures between nodes.
For practical distributed systems, network partitions are unavoidable.
Therefore the meaningful trade-off during a partition is often:
Consistency vs Availability
23. CAP Consistency
The C in CAP roughly means clients observe the system as though there were one current value.
Suppose:
WRITE x = 5
A consistent system should not allow another client to immediately read some older value from another authoritative replica.
24. CAP Availability
Availability means:
Every request to a nonfailed node eventually receives a response.
It does not mean the response is necessarily the newest data.
25. CP Systems
During a partition, a CP system prioritizes consistency.
Suppose:
A B | C D E
Only the side with sufficient quorum may continue writes.
The minority partition may reject requests:
503 unavailable
Rather than risk creating conflicting states.
Conceptually:
Better to temporarily stop
than return contradictory authoritative data.
26. AP Systems
An AP-oriented system may allow both sides to continue.
A B → accept writes
C D E → accept writes
When connectivity returns, conflicts must be reconciled.
Conceptually:
Better to remain available
and resolve inconsistencies later.
Neither choice is universally correct.
A banking ledger may prefer stronger consistency.
Social-media reaction counts may tolerate temporary disagreement.
27. CAP Is Frequently Misunderstood
CAP does not mean:
Pick any two out of three forever.
The actual trade-off becomes important specifically when a network partition exists.
Outside partitions, systems can often provide both good availability and strong consistency.
Also, real database design involves additional trade-offs such as:
latency
durability
throughput
consistency
availability
operational complexity
A useful extension is the PACELC perspective:
If Partition:
Availability vs Consistency
Else:
Latency vs Consistency
28. Consistency Models
"Consistent" is not a binary property.
Distributed systems expose different consistency models.
From stronger to weaker, you may encounter ideas such as:
Linearizability
Sequential consistency
Causal consistency
Read-your-writes consistency
Monotonic reads
Eventual consistency
These guarantees determine what clients are allowed to observe.
29. Linearizability
Linearizability gives the illusion that there is one copy of the data and each operation happens atomically at some instant between invocation and response.
Example:
Client A:
WRITE x=5
← success
Client B:
READ x
After A's write completed, B should not see an older value.
Linearizability is extremely intuitive.
It is also expensive across distant regions because coordination requires network communication.
30. Sequential Consistency
Sequential consistency requires all processes to observe operations in some single common order.
But that order does not necessarily have to respect real wall-clock completion order as strictly as linearizability.
This makes it weaker than linearizability.
31. Causal Consistency
If one event caused another, everyone should observe them in that order.
Suppose:
Alice:
"Deploy model v2"
Bob sees Alice's message and replies:
"v2 looks good"
Clients should not see Bob's reply before Alice's original message.
Independent events, however, may be observed in different orders.
32. Eventual Consistency
Eventual consistency says:
If writes stop, all replicas will eventually converge to the same value.
Suppose:
Replica A: x=5
Replica B: x=4
Replica C: x=3
Later replication runs:
A: x=5
B: x=5
C: x=5
During the convergence period, clients may observe different values.
Many internet-scale systems exploit this because it allows high availability and low latency.
33. Read-Your-Writes Consistency
Suppose you update your profile picture.
Immediately afterward, you expect to see the new picture.
A system providing read-your-writes guarantees ensures:
your subsequent reads
will not go backward behind your own writes
Other users might temporarily see the old value.
You will not.
This weaker guarantee is often enough for good user experience.
34. Distributed Queues
Direct synchronous communication looks like:
Service A → Service B
If B is unavailable, A may fail.
Queues add a buffer:
Producer
↓
Queue
↓
Consumer
Now A does not necessarily need B to be alive immediately.
Example:
User uploads video
↓
API writes job
↓
Queue
↓
GPU worker
↓
Video embedding generated
This is extremely common in AI infrastructure.
35. Why Queues Matter
Queues provide several useful properties.
Decoupling
Producer and consumer operate independently.
Buffering
Traffic spikes can accumulate in the queue.
Normal:
100 jobs/sec
Spike:
5,000 jobs/sec
Workers process the backlog gradually.
Retryability
Failed jobs can be attempted again.
Work distribution
Many workers can consume tasks:
Queue
/ | \
↓ ↓ ↓
Worker1 Worker2 Worker3
This is useful for:
model inference
embedding generation
image processing
training jobs
agent tasks
document indexing
36. Message Delivery Semantics
Queues often discuss:
at-most-once
at-least-once
exactly-once
At-most-once
Message is delivered zero or one time.
Possible loss.
No duplication.
At-least-once
Message is delivered one or more times.
No intentional loss.
Duplicates are possible.
Exactly-once
Each logical operation appears to happen exactly once.
This is difficult in genuinely distributed environments and normally requires carefully scoped transactional or deduplication mechanisms.
A useful engineering assumption is:
Design consumers so duplicate execution is safe.
That brings us to idempotency.
37. Idempotency
An operation is idempotent if repeating it has the same final effect as performing it once.
For example:
SET status = "completed"
can safely run repeatedly.
But:
balance += $100
is not idempotent.
Execute it twice:
+100
+100
and you accidentally add $200.
38. Idempotency Keys
Suppose a client makes a payment request:
POST /payment
The server processes it.
But the response is lost.
The client retries.
Without protection:
Payment #1 → charged
Payment #2 → charged again
Instead attach:
Idempotency-Key: abc123
Server records:
abc123 → already processed
Retrying the request returns the earlier result rather than executing the payment again.
The same pattern matters for:
job submission
agent actions
GPU scheduling
model deployment
database mutations
webhooks
39. Retries
Network failures are normal.
Therefore services retry requests.
A naive retry loop:
while failed:
retry()
is dangerous.
If thousands of clients retry simultaneously after an outage:
server recovers
↓
10,000 clients retry
↓
server overloaded again
This is a retry storm.
40. Exponential Backoff
Instead retry progressively slower:
1 second
2 seconds
4 seconds
8 seconds
16 seconds
Conceptually:
delay ≈ base × 2^attempt
Usually add randomness called jitter.
Instead of every client retrying at exactly:
8.000 seconds
clients retry around:
7.1 sec
8.4 sec
6.9 sec
9.0 sec
This prevents synchronized traffic waves.
41. Stream Processing
Queues commonly contain discrete jobs.
Streams represent continuously arriving events.
Example:
Robot telemetry
Robot telemetry
Robot telemetry
Robot telemetry
...
or:
click
click
purchase
search
click
A stream-processing pipeline may look like:
Events
↓
Kafka
↓
Stream Processor
↓
Aggregations
↓
Database / Dashboard / Model
Technologies in this space include concepts represented by systems such as:
Kafka
Flink
Spark Streaming
Pulsar
The important part is the architecture, not the brand.
42. Event Logs
Many streaming systems model data as an append-only log:
Offset 100: event A
Offset 101: event B
Offset 102: event C
Offset 103: event D
Consumers track their position:
consumer offset = 102
If the consumer crashes, it can restart from its previous offset.
This architecture enables replay.
You can process old events again using new software.
For ML systems this is powerful.
Suppose you deploy a better feature extractor.
You can replay historical events:
old events
↓
new feature extractor
↓
new feature dataset
43. Partitions in Streaming Systems
The word partition has another meaning in systems like Kafka.
A topic may be divided:
Topic: robot_telemetry
Partition 0
Partition 1
Partition 2
Partition 3
Each partition can be processed independently.
This increases parallelism.
Partition 0 → Worker A
Partition 1 → Worker B
Partition 2 → Worker C
Partition 3 → Worker D
Ordering is usually guaranteed within a partition, not globally across all partitions.
Choosing the partition key therefore matters.
For example:
partition_key = robot_id
ensures all messages for one robot remain ordered.
44. Distributed Storage
Eventually one machine cannot store all your data.
Suppose you have:
500 TB
You distribute it across machines:
Node A → data 1
Node B → data 2
Node C → data 3
Node D → data 4
This is sharding or partitioning.
Distributed storage systems must solve:
where data lives
how data is replicated
how nodes fail
how data moves
how clients find data
how consistency works
45. Sharding
Suppose users have numeric IDs.
You might use:
shard = user_id % 4
Then:
User 100 → shard 0
User 101 → shard 1
User 102 → shard 2
User 103 → shard 3
Simple modulo sharding has a major problem.
If you increase from four nodes to five:
user_id % 5
a huge fraction of keys move.
That can be expensive.
46. Consistent Hashing
Consistent hashing reduces data movement when nodes join or leave.
Conceptually imagine keys arranged on a ring:
Node A
|
--------+--------
/ \
Node D Node B
\ /
--------+--------
|
Node C
Keys are also hashed onto the ring.
Each key belongs to a nearby node.
Adding one server changes ownership mainly for neighboring key ranges rather than reshuffling everything.
This technique appears throughout distributed caches and storage systems.
47. Distributed File/Object Storage
Large AI workloads rarely store all training data inside a relational database.
Instead they often use distributed/object storage.
Conceptually:
training/
shard-00001.parquet
shard-00002.parquet
shard-00003.parquet
models/
model-v17/
config.json
weights-00001
weights-00002
Systems provide:
durability
replication
massive capacity
parallel access
AI pipelines frequently depend on object storage for:
datasets
checkpoints
model artifacts
logs
evaluation outputs
embeddings
48. Distributed Compute
Storage is only half the problem.
We often need thousands of machines to process the data.
Suppose you want to embed:
1 billion documents
One machine might take months.
Instead:
Documents
↓
Scheduler
↓
┌──────┬──────┬──────┬──────┐
↓ ↓ ↓ ↓
GPU1 GPU2 GPU3 GPU4
Each worker processes part of the workload.
This is distributed compute.
49. Data Parallelism
In ML training, the same model may run on multiple GPUs.
GPU 0 → batch A
GPU 1 → batch B
GPU 2 → batch C
GPU 3 → batch D
Each GPU computes gradients.
Then gradients are synchronized:
gradients
↓
AllReduce
↓
updated model
This is data parallel training.
The bottleneck may become communication rather than computation.
50. Model Parallelism
What if the model itself cannot fit on one GPU?
Split the model:
GPU 0:
Layers 1–20
GPU 1:
Layers 21–40
GPU 2:
Layers 41–60
GPU 3:
Layers 61–80
This is model/pipeline parallelism.
Large-model infrastructure often combines several forms:
data parallelism
tensor parallelism
pipeline parallelism
expert parallelism
Once training reaches this scale, distributed-systems concepts become unavoidable.
51. Distributed Inference
Inference systems also distribute work.
Suppose millions of LLM requests arrive:
Users
↓
Load Balancer
↓
Scheduler
↓
GPU workers
The scheduler may consider:
GPU memory
model loaded
request priority
prompt length
batching opportunity
KV-cache availability
hardware health
Large-model inference may itself span multiple GPUs.
Now serving one request can require coordinated communication among multiple machines.
52. Failure Detection
Suppose node A stops receiving responses from node B.
Is B dead?
You cannot know with absolute certainty.
Maybe:
B crashed
or:
B is overloaded
or:
network is delayed
or:
A's network interface is broken
Distributed failure detection is therefore based on suspicion.
53. Heartbeats
A common technique is:
Node B → heartbeat → Coordinator
every few seconds.
If several heartbeats disappear:
Coordinator:
"B may have failed."
But timeout choice creates a trade-off.
Very short timeout:
fast detection
more false positives
Long timeout:
fewer false positives
slower recovery
54. Failure Detectors Are Imperfect
A fundamental distributed-systems rule is:
Silence is not proof of death.
If node A cannot communicate with node B, it only knows:
I cannot currently communicate with B.
It does not know:
B is definitely dead.
This is why leader elections, leases, quorum systems, fencing tokens, and consensus exist.
55. Split Brain
Suppose:
Leader A
Follower B
Follower C
A network partition occurs.
A believes it is still leader.
Meanwhile B and C elect B leader.
Now:
A thinks: I am leader.
B thinks: I am leader.
This is a split-brain situation.
If both accept writes:
A receives x=5
B receives x=10
state diverges.
Consensus and quorum rules exist partly to prevent this.
A minority partition normally must not continue pretending to be authoritative.
56. Leases
Sometimes leadership is granted for a limited period:
lease valid until T
After the lease expires, the node must renew it before continuing.
Leases are useful for:
leadership
locks
resource ownership
cache authority
But clock assumptions matter when implementing them.
57. Fencing Tokens
Suppose worker A receives lock:
token = 41
Worker A becomes slow.
Its lock expires.
Worker B receives:
token = 42
Then A wakes up and attempts to write.
The storage system sees:
A token = 41
current token = 42
and rejects A's stale operation.
This is a fencing token.
It solves an important problem:
Lock expiration alone does not prevent the old holder from continuing to act.
58. Partial Failure
One of the defining properties of distributed systems is partial failure.
In a normal program:
program running
or
program crashed
In distributed systems:
Service A healthy
Service B overloaded
Service C unreachable
Database healthy
Cache stale
Queue healthy
GPU worker dead
network between A/C broken
The overall application can be partially functioning.
Good infrastructure therefore expects components to fail independently.
59. Timeouts
Every network call should generally have a timeout.
Without one:
Service A
waits forever
for Service B
Eventually A's worker pool fills.
Then A fails too.
This can propagate:
B slow
↓
A threads exhausted
↓
Gateway requests pile up
↓
whole system fails
Timeouts limit how long resources remain trapped.
60. Cascading Failures
Distributed failures often spread.
Suppose database latency increases.
Database slow
↓
API requests slow
↓
connections accumulate
↓
thread pools fill
↓
clients retry
↓
traffic increases
↓
database becomes slower
A small problem becomes an outage.
Resilient systems use mechanisms such as:
timeouts
backpressure
rate limiting
load shedding
circuit breakers
retry budgets
queues
bulkheads
61. Circuit Breakers
If a downstream service is clearly failing, repeatedly calling it can make things worse.
A circuit breaker behaves roughly like:
CLOSED
requests allowed
too many failures
↓
OPEN
requests rejected immediately
after cooldown
↓
HALF-OPEN
test a few requests
If those succeed:
CLOSED
Otherwise:
OPEN
This protects the failing service and prevents upstream resource exhaustion.
62. Backpressure
Suppose:
Producer = 1000 jobs/sec
Consumer = 100 jobs/sec
The queue grows:
1000
2000
3000
...
Eventually memory or storage is exhausted.
Backpressure means slowing producers when downstream consumers cannot keep up.
This is essential in:
streaming
GPU inference
robot telemetry
data ingestion
distributed pipelines
63. Load Shedding
Sometimes the system simply cannot serve all requests.
Instead of allowing everything to fail, intentionally reject low-priority work.
Example:
GPU utilization = 100%
priority:
1. interactive inference
2. enterprise jobs
3. batch embedding
4. background analytics
The system might pause category 4.
This keeps critical paths alive.
64. Distributed Transactions
Suppose a business operation touches two services:
Bank service:
subtract $100
Inventory service:
reserve item
What if the first succeeds and the second fails?
Now the system is inconsistent.
Traditional databases solve similar problems using transactions.
Across independent services, transactions become much more difficult.
65. Two-Phase Commit
A classic distributed transaction protocol is 2PC.
Coordinator asks:
PREPARE?
Participants respond:
YES
YES
YES
Then coordinator says:
COMMIT
Conceptually:
Phase 1:
Can everyone commit?
Phase 2:
Everyone commit.
The major weakness is coordination/blocking if the coordinator fails at an unfortunate time.
Therefore many service architectures prefer application-level patterns rather than broad distributed ACID transactions.
66. Saga Pattern
A saga breaks a large operation into smaller transactions.
Example:
1. reserve inventory
2. charge payment
3. schedule delivery
If step 3 fails, perform compensating operations:
refund payment
release inventory
This is not magical rollback.
It is explicit business logic.
AI orchestration systems may use similar ideas when agents invoke multiple side-effecting tools.
67. Queues + Idempotency + Retry = Core Pattern
One of the most reusable infrastructure patterns is:
Producer
↓
Durable Queue
↓
Idempotent Worker
↓
Database
If worker crashes:
job retries
If duplicate delivery occurs:
idempotency prevents duplicate effect
If traffic spikes:
queue absorbs load
This simple combination powers enormous amounts of production infrastructure.
68. Distributed Tracing
Imagine one user request crosses:
API Gateway
↓
Authentication
↓
Agent Service
↓
Vector Search
↓
LLM Server
↓
Tool Service
↓
Database
The user reports:
"The request took 11 seconds."
Where was the delay?
Normal logs on six machines are difficult to correlate.
Distributed tracing solves this.
69. Trace IDs and Span IDs
Assign the entire request a trace ID:
trace_id = abc123
Every service propagates it.
Individual operations become spans:
Trace abc123
├─ API Gateway 80 ms
├─ Agent Service 250 ms
├─ Vector DB 75 ms
├─ LLM Inference 9400 ms
└─ Tool Call 700 ms
Now the bottleneck is obvious.
70. Trace Context Propagation
Service A sends:
trace_id=abc123
span_id=1
to Service B.
B creates another span:
trace_id=abc123
span_id=2
parent_span=1
The resulting graph may look like:
Request
|
+── API
|
+── Agent
|
+── Vector Search
|
+── LLM
|
+── Tool Call
This gives you the causal structure of a distributed request.
71. Logs, Metrics, and Traces
The three major observability signals are:
Metrics
Logs
Traces
Metrics answer:
Is something wrong?
Example:
p99 latency = 8.2 seconds
error rate = 7%
GPU utilization = 99%
Logs answer:
What happened?
Example:
timeout calling inference_worker_7
Traces answer:
Where did the request spend its time?
Together they provide observability.
72. Percentiles Matter
Average latency can hide bad behavior.
Suppose:
99 users: 100 ms
1 user: 10 seconds
Average does not tell the full story.
Infrastructure commonly monitors:
p50
p95
p99
p99.9
If:
p50 = 100 ms
p95 = 400 ms
p99 = 7 sec
you have a tail-latency problem.
Distributed systems often amplify tail latency because one request may depend on many remote operations.
73. Fan-Out and Tail Latency
Suppose a search request queries 100 shards:
Search
|
┌───────────┼───────────┐
↓ ↓ ↓
Shard 1 Shard 2 ... Shard 100
The overall request may wait for nearly the slowest shard.
Even if individual shard latency is usually good, querying many machines increases the chance that at least one is slow.
This is why large distributed systems care enormously about tail latency.
74. Caching in Distributed Systems
Caches reduce repeated expensive work.
Example:
User asks same embedding
↓
Cache?
/ \
hit miss
↓ ↓
response GPU model
Caches can exist:
in process
on the same machine
distributed across machines
at the CDN
near databases
near model servers
But caches create consistency problems.
75. Cache Invalidation
Suppose:
Database:
model_version = v5
Cache:
model_version = v4
The cache is stale.
You need policies such as:
TTL
explicit invalidation
write-through
write-back
cache-aside
versioned keys
The famous saying:
There are only two hard things in computer science: cache invalidation and naming things.
Distributed caching makes the first one especially visible.
76. Distributed Locks
Sometimes multiple machines must coordinate access to one resource.
Example:
only one worker should perform model migration
A distributed lock might ensure:
Worker A → owns lock
Worker B → waits
Worker C → waits
But distributed locks are harder than local mutexes because:
lock holder may crash
network may partition
lease may expire
clock may drift
old worker may wake up later
Correct implementations often involve:
leases
consensus
fencing tokens
Do not casually implement distributed locks using one database row without understanding failure modes.
77. Leader-Based Systems
Leadership appears everywhere.
A leader may coordinate:
replication
job scheduling
metadata
cluster membership
distributed locks
database writes
Followers make the system resilient.
Typical lifecycle:
Leader healthy
↓
Leader crashes
↓
Failure suspected
↓
Election
↓
New leader selected
↓
Service resumes
Consensus ensures old and new leaders do not independently control the same authoritative state.
78. Leaderless Systems
Not every system requires one leader.
Some distributed databases allow clients to write to several replicas directly.
For example:
N = 3 replicas
Write quorum W = 2
Read quorum R = 2
If:
R + W > N
the read and write sets overlap.
This helps clients discover the newest data.
Leaderless architectures often require techniques such as:
versioning
conflict resolution
read repair
anti-entropy
quorums
79. Gossip Protocols
How can thousands of machines distribute cluster information?
A central coordinator is not always ideal.
Gossip protocols work somewhat like rumors.
A tells B
A tells C
B tells D
C tells E
...
Eventually information spreads through the cluster.
Gossip is useful for:
membership
failure detection
metadata dissemination
replica synchronization
It scales well because each node communicates with only a few peers.
80. Anti-Entropy
Replicas can gradually diverge.
Anti-entropy mechanisms periodically compare replicas and repair differences.
Conceptually:
Replica A
↕ compare
Replica B
Instead of transferring every key, systems may compare compact summaries such as tree structures or hashes.
This enables efficient convergence.
81. Control Plane vs Data Plane
A useful infrastructure distinction:
Control plane
Decides what should happen.
Examples:
Which GPU runs this model?
Who is leader?
Where should replicas live?
What configuration is active?
Data plane
Actually handles the workload.
Examples:
Serve inference
Store bytes
Forward network packets
Execute kernels
A Kubernetes-like mental model:
Control Plane
↓
schedules work
Worker Nodes
↓
execute work
Many distributed architectures become easier to understand once you separate these responsibilities.
82. Stateless vs Stateful Services
Stateless services are easy to replicate.
Example:
API Server A
API Server B
API Server C
If none stores important local state, any request can go to any instance.
State lives elsewhere:
database
cache
object store
queue
This makes autoscaling and recovery much easier.
Stateful systems are harder because machine identity and local data matter.
Examples:
databases
Kafka brokers
distributed filesystems
model servers with huge loaded weights
GPU workers with KV caches
83. Horizontal Scaling
Vertical scaling means:
bigger machine
For example:
32 GB → 128 GB RAM
1 GPU → 8 GPUs
Horizontal scaling means:
more machines
For example:
1 inference server → 100 inference servers
Distributed systems primarily exist to make horizontal scaling possible.
But horizontal scaling introduces coordination, replication, routing, and failure problems.
84. Sharding vs Replication
These are often confused.
Sharding
Different nodes store different data.
A → users 1–1M
B → users 1M–2M
C → users 2M–3M
Purpose:
capacity / throughput
Replication
Different nodes store copies of the same data.
A → shard 1
B → shard 1 copy
C → shard 1 copy
Purpose:
availability / durability / read scaling
Production databases often combine both.
Shard 1:
A1 A2 A3
Shard 2:
B1 B2 B3
Shard 3:
C1 C2 C3
85. AI Example: Distributed Vector Database
Suppose you have:
2 billion embeddings
They cannot fit efficiently on one server.
You partition them:
Shard A → vectors 0–500M
Shard B → vectors 500M–1B
Shard C → vectors 1B–1.5B
Shard D → vectors 1.5B–2B
Each shard may itself have replicas:
Shard A:
A1 leader
A2 follower
Shard B:
B1 leader
B2 follower
A query may fan out:
Embedding query
↓
Coordinator
┌────┼────┬────┐
↓ ↓ ↓ ↓
A B C D
\ | | /
\ | | /
merge
↓
top-k results
Now nearly every concept in this chapter appears:
sharding
replication
fan-out
failover
consistency
timeouts
tail latency
tracing
load balancing
86. AI Example: LLM Inference Cluster
Imagine a production LLM platform.
Users
↓
API Gateway
↓
Request Router
↓
Inference Scheduler
↓
GPU Cluster
The scheduler needs to know:
which GPUs are healthy
which model versions are loaded
available GPU memory
queue depth
request priority
batching opportunities
Workers send heartbeats.
GPU Worker → heartbeat → Scheduler
Requests may queue.
Request → queue → GPU
Failures require retries.
But generation requests can have side effects if tools are involved, so idempotency may matter.
Tracing tells you whether latency came from:
queueing
tokenization
prefill
decoding
network
tool calls
storage
Distributed systems are not a separate concern from AI serving.
They are AI serving infrastructure.
87. AI Example: Distributed Training
Consider training a large transformer:
Dataset
↓
Data Loader
↓
Thousands of GPUs
↓
Gradient synchronization
↓
Checkpoint storage
Failure possibilities include:
GPU failure
machine failure
network failure
storage failure
collective communication timeout
straggler
scheduler failure
Training systems therefore need:
checkpointing
fault detection
distributed communication
scheduling
replicated metadata
object storage
restart logic
At large scale, reliability becomes a statistical problem.
If one server fails once every three years, that sounds reliable.
But with:
10,000 servers
some machine may fail almost every few hours.
Large systems must be designed assuming failure is routine.
88. AI Example: Robot Fleet
Now imagine 100,000 autonomous robots.
Each robot sends:
position
battery
sensor state
diagnostics
mission status
camera events
Cloud architecture might look like:
Robots
↓
Ingestion Gateway
↓
Stream Platform
↓
Telemetry Processor
↓
Fleet State Database
↓
Monitoring / Planning / ML
Commands travel back:
Cloud
↓
Command Queue
↓
Robot
Now distributed systems interact with physical reality.
A duplicated database operation may be annoying.
A duplicated robot command can be dangerous.
Therefore concepts such as:
sequence numbers
idempotency
command IDs
leases
heartbeats
timeouts
reconciliation
become safety-critical.
89. Desired State vs Actual State
A powerful architecture used in infrastructure and robotics is reconciliation.
Suppose desired state says:
Model v5 should run on 10 GPUs.
Actual state says:
Model v5 running on 8 GPUs.
A controller computes:
difference = desired - actual
and starts two workers.
This loop repeats:
observe
compare
act
repeat
This is the reconciliation loop.
Kubernetes controllers are built around this idea.
So are many robotics control and orchestration systems.
It works well because distributed systems are constantly drifting away from intended state because of failures.
Instead of assuming commands always succeed, reconciliation keeps trying to make reality match intent.
90. Exactly-Once Thinking Can Be Dangerous
Engineers often wish for:
Exactly once execution.
Across arbitrary distributed boundaries, that is much harder than it sounds.
Suppose:
Client → Server
The server performs an operation.
Then crashes before sending the acknowledgment.
From the client's perspective:
Did it happen?
Impossible to know solely from the missing response.
Therefore robust distributed architectures often prefer:
at-least-once delivery
+
idempotent processing
+
deduplication
rather than pretending duplicate execution cannot happen.
91. The Fundamental Ambiguous Failure
Memorize this scenario.
Client sends:
Charge card $50
No response returns.
What happened?
Possibility 1:
request was lost
Possibility 2:
server crashed before processing
Possibility 3:
server charged card and crashed
Possibility 4:
server charged card and response was lost
The client cannot distinguish these cases.
This tiny example contains a huge portion of distributed-systems difficulty.
The solution is not simply "retry."
It is:
unique operation ID
+
durable state
+
idempotent handling
+
careful retry policy
92. At-Least-Once Mindset
When designing distributed components, a strong default mental model is:
Any message may be delivered more than once.
Therefore ask:
What happens if this executes twice?
Examples:
send_email()
charge_customer()
deploy_model()
move_robot()
create_job()
update_inventory()
If repetition causes harm, add deduplication or make the operation idempotent.
93. Assume Messages Can Be Reordered
Consider:
Message 1:
status = RUNNING
Message 2:
status = COMPLETED
The network might deliver:
COMPLETED
RUNNING
If the receiver blindly applies them, the final state becomes wrong.
Solutions include:
sequence numbers
versions
logical clocks
timestamps with appropriate guarantees
monotonic state transitions
For example:
event 41 → RUNNING
event 42 → COMPLETED
Receiver ignores event 41 after having processed 42.
94. Assume Messages Can Be Delayed
A message might arrive minutes after it was useful.
Imagine:
Robot command:
MOVE FORWARD
sent at mission sequence 100.
Robot has already advanced to sequence 120.
A delayed sequence-100 command should not suddenly execute.
Therefore distributed physical systems often attach:
command_id
sequence number
mission version
expiration time
95. Version Everything Important
Versions are enormously useful in distributed systems.
Examples:
model_version = 17
configuration_version = 402
schema_version = 3
robot_plan_version = 921
resource_version = 8482
Then stale operations can be rejected.
Example:
update if current_version == 17
If another process already changed it to 18:
reject update
This is optimistic concurrency control.
96. Compare-and-Swap
Compare-and-swap conceptually performs:
if current_value == expected:
current_value = new_value
else:
fail
Example:
expected leader = A
new leader = B
Only update if A is still the leader.
Database systems expose related primitives through conditional updates, transactions, versions, or revision numbers.
These are foundational for coordination.
97. Hotspots
Sharding alone does not guarantee even load.
Suppose requests are partitioned by:
celebrity_user_id
One celebrity receives millions of requests.
One shard becomes:
100% CPU
while others are:
10%
12%
8%
This is a hot shard or hotspot.
Distributed architectures therefore care about:
key distribution
load balancing
partition splitting
consistent hashing
replication
request routing
98. Stragglers
A straggler is a worker that is much slower than others.
Suppose 100 distributed workers finish:
Worker 1: 10 sec
Worker 2: 10 sec
...
Worker 99: 11 sec
Worker 100: 80 sec
If the entire job waits for all workers:
total = 80 sec
One worker determines completion time.
Stragglers matter greatly in:
MapReduce
distributed training
vector search
batch processing
Possible mitigation includes:
speculative execution
work stealing
timeouts
adaptive scheduling
redundant requests
99. Coordination Is Expensive
One of the best high-level rules in distributed systems is:
Coordination costs latency and availability.
If every write must ask five remote machines:
"Do you agree?"
your system pays network round trips.
If one required participant is unavailable, progress may stop.
Therefore high-scale systems try to coordinate only when required.
Strong correctness needs coordination.
Some operations can tolerate weaker guarantees.
Architecture is often about deciding where coordination is actually worth paying for.
100. Local Decisions Scale Better
Compare:
Globally coordinated
Node A
Node B
Node C
Node D
Node E
Everyone must agree.
vs.
Partitioned responsibility
A controls partition 1
B controls partition 2
C controls partition 3
Independent local operations scale better.
This insight explains many architectures involving:
sharding
partitioned logs
actors
ownership
single-writer designs
101. Single-Writer Principle
For a particular piece of mutable state, having one authoritative writer simplifies ordering dramatically.
Instead of:
A writes
B writes
C writes
use:
A/B/C requests
↓
Owner
↓
state
The owner serializes operations.
Leader-based databases and actor systems exploit this idea.
The trade-off is that ownership itself must survive failures.
102. Distributed Systems Are State Machines
A very important theoretical idea:
Many reliable distributed systems are replicated deterministic state machines.
Suppose every server begins with:
x = 0
and all servers execute the same ordered commands:
1. x += 5
2. x *= 2
3. x -= 3
Then every replica reaches:
x = 7
Therefore the difficult part becomes agreeing on the order of commands.
Consensus algorithms such as Raft effectively provide that ordered log.
Consensus
↓
ordered commands
↓
deterministic state machine
↓
replicated state
This mental model connects consensus, replication, databases, and distributed coordination.
103. The Replicated Log Is a Powerful Abstraction
Imagine:
1 CREATE user 42
2 SET model=v4
3 START job 81
4 COMPLETE job 81
If every replica processes this same log in the same order, their states converge.
A replicated log becomes the source of truth.
This idea appears in:
consensus systems
streaming platforms
database replication
event sourcing
control planes
Learn to recognize it.
104. Event Sourcing
Instead of storing only current state:
balance = 125
store events:
+100
+50
-25
Current state is derived by replaying events.
Advantages can include:
auditability
replay
historical reconstruction
new projections
But event-sourced systems introduce complexity around:
schema evolution
event ordering
large histories
replay cost
The key idea remains valuable:
State can be viewed as the result of an ordered event history.
105. Distributed Systems and AI Agents
Agent systems increasingly behave like distributed workflows.
An agent may:
read email
query database
call LLM
invoke browser
run code
write document
send message
These actions may involve independent services.
Failures can happen between any two steps.
Therefore reliable agent systems need concepts familiar from distributed workflows:
persistent state
tool-call IDs
idempotency
timeouts
retries
workflow checkpoints
compensation
tracing
Suppose an agent:
1. generates invoice
2. emails invoice
3. marks invoice sent
It crashes after step 2.
After restart:
Should it email again?
This is a distributed-system problem, not merely an LLM problem.
106. Distributed Systems and Robotics
A robot itself is often a distributed system.
It may contain:
camera node
localization node
planner
motor controller
perception model
navigation system
safety controller
cloud connection
Messages flow between components.
ROS 2 itself exposes many distributed concepts:
publish/subscribe
QoS
message delivery
node discovery
timestamps
distributed communication
failure
When multiple robots communicate with cloud infrastructure, the problem expands further.
Distributed-systems knowledge therefore directly strengthens robotics engineering.
107. Questions to Ask When Designing Any Distributed System
Whenever you see two services talking over a network, ask:
What if the request never arrives?
What if the request executes but the response never arrives?
What if the request executes twice?
What if messages arrive out of order?
What if one node becomes extremely slow?
What if the network partitions?
What if the leader crashes?
What if two nodes both believe they are leader?
What if a replica is stale?
What if the retry makes the outage worse?
What state survives restart?
How will I know where a request failed?
Those questions will often reveal architecture problems before production does.
108. Working Mental Model
For day-to-day engineering, reduce distributed systems to the following mental picture.
Machines fail.
Networks fail.
Messages may be:
lost
delayed
duplicated
reordered
Clocks disagree.
Remote failures are ambiguous.
Therefore:
replicate important state
establish ownership
use consensus where authority matters
use versions
use timeouts
retry carefully
make operations idempotent
design for duplicate delivery
monitor heartbeats
expect partial failure
propagate trace IDs
reconcile desired and actual state
That is the practical core.
109. Topic Map
When you later need to revisit a specific area, think of the subject as this map:
DISTRIBUTED SYSTEMS
│
├── TIME
│ ├── physical clocks
│ ├── clock drift
│ ├── synchronization
│ ├── Lamport clocks
│ └── vector clocks
│
├── AGREEMENT
│ ├── consensus
│ ├── quorum
│ ├── Raft
│ ├── Paxos
│ └── replicated state machines
│
├── DATA
│ ├── replication
│ ├── leader/follower
│ ├── leaderless replication
│ ├── sharding
│ ├── consistent hashing
│ └── distributed storage
│
├── CONSISTENCY
│ ├── partitions
│ ├── CAP
│ ├── linearizability
│ ├── causal consistency
│ └── eventual consistency
│
├── MESSAGING
│ ├── queues
│ ├── streams
│ ├── partitions
│ ├── consumer offsets
│ └── delivery semantics
│
├── COMPUTE
│ ├── distributed jobs
│ ├── schedulers
│ ├── data parallelism
│ ├── model parallelism
│ └── distributed inference
│
├── FAILURE
│ ├── heartbeats
│ ├── failure detection
│ ├── leader election
│ ├── split brain
│ ├── retries
│ ├── exponential backoff
│ ├── idempotency
│ ├── circuit breakers
│ └── backpressure
│
└── OBSERVABILITY
├── logs
├── metrics
├── trace IDs
├── spans
└── distributed tracing
110. What an AI Infrastructure Engineer Should Actually Know
You do not need to become a distributed-systems researcher before building useful AI systems.
For practical AI infrastructure work, you should be very comfortable with:
network failure is ambiguous
timeouts are mandatory
retries can duplicate work
idempotency prevents retry corruption
replicas may become stale
strong consistency costs coordination
network partitions can split clusters
leaders need elections and quorum
Raft gives you replicated authoritative state
queues decouple producers and workers
streams represent ordered event histories
partitioning enables horizontal scale
replication provides resilience
distributed storage separates data across machines
distributed compute separates computation across machines
heartbeats detect suspected failures
backpressure prevents overload
circuit breakers prevent cascading failures
traces reconstruct requests across services
If you understand why each of those statements is true, you already possess the foundation necessary to reason about a large amount of modern infrastructure.
111. The Deeper Lesson
The deepest lesson of distributed systems is not Raft.
It is not Paxos.
It is not CAP.
It is this:
You can never assume that another machine shares your exact view of reality.
Your process may believe:
request failed
while the server believes:
request succeeded
Your node may believe:
leader is dead
while the leader is perfectly healthy but unreachable.
One replica may believe:
x = 5
while another still sees:
x = 4
One worker may believe:
I own this job.
while another has already taken ownership.
A robust distributed system does not eliminate uncertainty.
It designs protocols that remain safe despite it.
That is why distributed systems rely on:
quorums
versions
logs
epochs
terms
idempotency keys
leases
fencing tokens
heartbeats
timeouts
reconciliation
Each mechanism is a way of imposing enough structure on uncertainty to make the system dependable.
112. Final Mental Model for AI Infrastructure
Imagine you are building an autonomous AI platform:
┌───────────────┐
│ Clients │
└───────┬───────┘
↓
┌───────────────┐
│ API Gateway │
└───────┬───────┘
↓
┌───────────────────────┐
│ Agent / Orchestrator │
└───────────┬───────────┘
│
┌──────────────────┼───────────────────┐
↓ ↓ ↓
Vector Store Message Queue LLM Router
│ │ │
↓ ↓ ↓
Storage Shards Worker Fleet GPU Cluster
│ │ │
└──────────────────┼───────────────────┘
↓
Distributed Storage
Behind this diagram are all the ideas we studied.
The vector database uses:
sharding
replication
consistency
The worker fleet uses:
queues
heartbeats
retries
idempotency
The GPU cluster uses:
distributed scheduling
failure detection
load balancing
distributed compute
The control plane may use:
leader election
consensus
Raft
The entire system requires:
timeouts
backpressure
circuit breakers
metrics
logs
distributed tracing
And if parts of it span regions:
clock uncertainty
network partitions
replication lag
CAP trade-offs
Distributed systems therefore are not merely another CS subject sitting beside AI.
They are the engineering substrate that allows AI to operate beyond a single computer.
When an AI model moves from:
model.py
on your laptop to:
millions of requests
thousands of GPUs
petabytes of data
many regions
robot fleets
autonomous agents
you are no longer solving only a machine-learning problem.
You are solving a distributed-systems problem.
And the most useful question to carry with you is:
What happens when this message, machine, replica, clock, or assumption fails?
If your design has a convincing answer to that question, you are beginning to think like a distributed-systems engineer.