What is a Transaction?

At its core, a transaction is a group of read/write operations that should be executed as one atomic unit — either all succeed or all fail.

Think:

• Transferring ₹1,000 from one bank account to another
• Placing an e-commerce order and updating inventory
• Publishing a blog post and tagging followers

Without transactions, we risk inconsistency: partial updates, double-charges, or missing notifications.

Single-Object vs Multi-Object Transactions

Not all systems need full transactional semantics. Redis and DynamoDB offer atomic operations on single keys. But complex systems, like relational databases, often require coordination across multiple rows or tables.

Example: Booking a seat and deducting balance requires transactional integrity across multiple tables. Without multi-object transactions, a system risks overbooking or inconsistent billing.

Implementing Transactions: Write-Ahead Logging

On a single node, transactions are typically backed by:

• Write-Ahead Logs (WALs): Changes are first written to a durable log before applying to the database.
• Undo logs: Roll back in-progress operations on failure.

This mechanism ensures atomicity and durability, but doesn’t help with distributed coordination.

ACID Guarantees

The classic model is ACID:
Atomicity: All or nothing — no partial state

Consistency: Moves the DB from one valid state to another

Isolation: Concurrent txns don’t interfere with each other

Durability: Once committed, data survives crashes

But in real-world distributed systems, ACID is not binary. It’s a sliding scale — especially Isolation.

Concurrency Control: Who Gets to Touch What and When

Concurrency control ensures isolation and consistency under concurrent access.

Pessimistic Concurrency (Locks)

• Assumes conflicts are likely.
• Transactions acquire locks on records/tables.
• Prevents race conditions but causes blocking and potential deadlocks.

Optimistic Concurrency (MVCC)

• Assumes conflicts are rare.
• Transactions proceed without locks.
• Validation occurs at commit time. If conflict is detected, one transaction is aborted.

Isolation Levels

Real systems trade off isolation to boost performance:

Read Uncommitted: Dirty reads — Very high-throughput systems
You read data that was written by an uncommitted transaction.
If that transaction rolls back, you’re left with invalid or inconsistent data.

Read Committed: Non-repeatable reads — Most RDBMS defaults
You read the same row twice in the same transaction, but get different results.
This happens when another transaction modifies or deletes the row between your reads.

Repeatable Read: Phantom reads — Financial systems
You run a query that returns a set of rows, then run it again and get new rows (phantoms).
This is caused by another transaction inserting or deleting rows that match your query condition.

Serializable: No anomalies, full isolation — Strict consistency needs

Serializable isolation ensures that concurrent transactions behave as if executed one after the other — eliminating anomalies like dirty reads, write skew, or phantoms. It’s ideal for correctness, but comes at a cost.

To guarantee serializability, systems need tight coordination across nodes, which introduces latency and complexity due to network delays and partial failures. Two common approaches:

• Two-Phase Locking (2PL):
  Transactions lock all read/write data until commit.
  Ensures serializability
  Causes blocking, deadlocks, and reduced throughput in high-contention scenarios.
• Serializable Snapshot Isolation (SSI):
  Based on MVCC — each transaction works off a consistent snapshot.
  Conflicts are detected at commit time, and transactions are aborted if needed.
  More concurrency-friendly than 2PL
  Still incurs abort overhead and tracking costs.

Distributed Challenges

• Global locks and consistent snapshots are hard to maintain across replicas.
• Coordination protocols often violate availability or increase latency.
• Systems like Spanner and CockroachDB solve this, but with complex clock or consensus designs.

Best Practices:

• Don’t default to serializability — ask what guarantees your use case actually needs.
• If you must use it, plan for coordination, handle contention, and monitor latency tradeoffs.
• Often, weaker models (like Snapshot Isolation) offer a good balance of safety and scalability.

Distributed Transactions and 2PC

In distributed systems, transactions often touch multiple nodes or services. To coordinate these, systems use protocols like:

Two-Phase Commit (2PC)

1. Prepare: Ask all participants if they’re ready to commit
2. Commit: If all say yes, go ahead; else rollback

But 2PC has problems:

• It’s blocking — if the coordinator crashes, participants get stuck
• No built-in rollback timeout — needs app-level recovery logic

Real-World Workarounds

Most high-scale systems avoid distributed transactions in favor of:

• Idempotent writes + retry logic
• Compensating actions (undo logic, e.g., refund after a failed shipment)
• Eventual consistency via messaging/event queues (Kafka, SQS)
• Application-level sagas — a chain of local transactions coordinated by logic

Real-Life Examples

1. Ride Sharing: Matching rider and driver, charging payment, and updating availability — often coordinated via retries and compensating actions, not strict transactions.
2. Booking Systems: Airline or hotel bookings often use locks or temporary holds instead of full transactional isolation.
3. Microservices: A checkout service talking to payment, inventory, and shipment services can’t use a shared DB — so it’s forced to coordinate loosely via events.

Key Takeaways

• Transactions are essential for correctness — but expensive across nodes.
• In distributed systems, isolation is a trade-off, not a given.
• Prefer local transactions + application-level orchestration (Sagas, retries).
• Know your consistency needs — sometimes, eventual is “good enough.”

Coming Up Next

In the next post, we’ll dive into The Trouble with Distributed Consensus — the protocols and pain points behind keeping distributed systems in sync.

If replication is about reliability, partitioning is about scalability.

Why Partition at All ?

Partitioning — often called sharding — is essential when your system grows beyond what a single server can handle.

Without partitioning, you hit hard limits on:

• Storage (e.g., disk size, memory)
• Throughput (CPU, IOPS, bandwidth)
• Fault isolation (a bad record takes down everything)

Partitioning distributes both data and load across many machines. Each machine (or node) is responsible for only a subset (partition) of the data.

Partitioning Basics

Partitioning by Key

Every record has a partition key. The system uses this to route the data to a specific node.
Choose your key carefully — it affects: Load balancing, Query efficiency, Join and transaction complexity

Example:
If you’re building a ride-hailing app:

• Partition by city_id → geographic distribution
• Partition by user_id → personalization queries
• Partition by trip_id → useful for trip-centric writes

Common Strategies

a. Hash Partitioning

• Applies a hash function to the key
• Evenly spreads data across nodes
• Used in Cassandra, MongoDB, Kafka partitions

Good for: Write-heavy systems needing uniform load
Limitation: Poor range-query support and tricky rebalancing

b. Range Partitioning

• Assigns ranges of keys to each partition (e.g., A–F, G–L)
• Allows efficient range queries and sorted scans

Used by systems like Google Bigtable, Spanner, HBase

Good for: Analytics, time-series queries
Limitation: Skew if some ranges are hotter (e.g., recent timestamps)

c. Directory-Based Partitioning

• Maintains a key-to-node mapping in a separate service
• Offers total flexibility in routing

Used in older relational sharding architectures or custom systems.

Good for: Complex routing, legacy systems
Limitation: Adds a centralized dependency

Real-World Challenges with Partitioning

Partitioning isn’t just about splitting data — it’s about managing the complexity that follows.

1. Hotspots and Skew

Even with hash partitioning, certain users or keys can dominate traffic.

Real-world example:

• Partitioning temperature sensor data by timestamp causes all real-time writes to hit the same partition. This creates a write hotspot every minute, overwhelming the “current time” shard.
• E-commerce flash sales may overload a partition if orders are partitioned by product ID.
• Partitioning trending content by hashtag causes all posts with a viral tag (e.g., #WorldCup) to land on the same shard.

Solutions:

• Add randomness to keys (e.g., userId#salt)
• Time-bucketed partitioning (e.g., 2025-Q1#userId)

2. Rebalancing Data

What happens when you add a new node?

In naive partitioning:

• You might need to re-shard everything (huge data movement)

Systems like:

• Cassandra and Dynamo use consistent hashing with virtual nodes to minimize rebalancing effort
• Bigtable and Spanner split range-based partitions dynamically under load

Design tip: Use virtual shards and dynamic mapping from the start.

3. Secondary Indexing Across Partitions

Partitioning optimizes lookups by the primary key. But what if you need to query by other fields?

Example:

• Partitioned by user_id, but need to search by email
• Partitioned by trip_id, but need trips in a city

Options:

• Scatter-Gather: Query all partitions and merge results (expensive)
• Global Index: A separate index partitioned differently
• Denormalization: Duplicate data for query efficiency

Engineering Trade-off: Better performance vs. more complexity and consistency handling.

4. Cross-Partition Joins and Transactions

When related records are in different partitions, joins become distributed queries, and transactions need coordination.

Real-world implications:

• Joins across user → order → payment might span multiple nodes
• Cross-shard transactions require 2PC, which adds latency and potential for locks/failures

Mitigation:

• Co-locate related data in the same shard (e.g., customer + orders)
• Denormalize (store redundant data to reduce joins)
• Avoid distributed transactions where possible

Best Practices

1. Choose a partition key that aligns with your access patterns

You don’t want to query by email if you partition by user_id.

2. Design with future growth in mind

• Use virtual shards
• Expect rebalancing and automate it

3. Monitor partition health continuously

• Skew and lag show up long after launch

4. Denormalize or duplicate wisely

• Normalize for correctness, denormalize for scale

5. Accept that distributed joins and transactions are expensive

• Try to stay within a shard boundary for critical flows

As systems grow, your partitioning decisions age. What works at 100K users may break at 10M.
That’s why partitioning should be revisited with scale, not just designed once and forgotten.

In the next post, we’ll move from how data is split to how it’s kept correct: distributed transactions — and why correctness, in a partitioned world, is harder than it looks.

In the world of distributed systems, replication is a cornerstone of reliability, availability, and performance. But with these benefits come a host of engineering trade-offs. This post distills key insights from the Replication chapter of Designing Data-Intensive Applications (DDIA), arguably one of the best books written on modern data systems.

This post offers a high-level overview of replication. In the upcoming posts of this series, we’ll dive deeper into specific replication strategies, challenges, and real-world applications.
We’ll begin by covering key concepts at a broad level before exploring each topic in greater detail.

Why Replicate Data?

Replication means storing copies of the same data on multiple machines. The main goals are:

• Fault Tolerance: If one node fails, another can take over.
• High Availability: Minimize downtime and serve read requests faster.
• Performance: Serve users closer to the data (geo-replication).

Replication Models

1. Single-Leader Replication (a.k.a. Primary-Replica)

• One leader handles all writes.
• Followers replicate data from the leader.
• Pros: Simple, strong consistency (if reads go to leader).
• Cons: Downtime during leader failure; can’t scale writes.

2. Multi-Leader Replication

• Multiple nodes can accept writes.
• Common in multi-region setups or offline-first apps (e.g., CRDTs or mobile sync).

Challenges:

• Write conflicts
• Conflict resolution strategies (timestamps, last-write-wins, custom merge logic)

3. Leaderless Replication

• No designated leader. Any node can accept writes.
• Uses quorum-based consistency (e.g., read/write majority).
• Found in systems like Amazon Dynamo, Cassandra.

Trade-offs:

• Complex conflict resolution
• Higher consistency management burden on client or database

The Consistency Trade-off

“In a distributed system, consistency is not a yes/no question — it’s a spectrum.” — Martin Kleppmann

Replication forces us to choose between:

• Strong consistency (all replicas always agree)
• Eventual consistency (replicas may temporarily diverge)
• Tunable consistency (you choose R/W quorum levels, e.g., in Cassandra)

This naturally ties into the CAP Theorem.

Pitfalls and Edge Cases

Some challenges that arise in replicated systems:

• Replication lag: Followers behind the leader can cause stale reads.
• Read Your Own Writes (RYOW) issues
• Write Conflicts in multi-leader and leaderless systems
• Split-brain scenarios in failovers
• Anti-entropy mechanisms to reconcile differences between nodes

Best Practices and Design Tips

• Know your consistency requirements before choosing a replication strategy.
  Strong consistency : Bank transactions, stock trading, password updates.
  Eventual consistency : News feed updates, analytics counters, product likes.
• Use single-leader if you need strong consistency and your write load is low to moderate.
  Thought: How would you design if consistency is of paramount importance and write load is heavy?
• Use multi-leader for disconnected systems (e.g., mobile apps).
  Best suited for offline-first or geo-distributed systems where users or regions need to write locally and sync later, collaboartive apps, POS systems in low connectivity environments.
• Use leaderless if availability is paramount and eventual consistency is acceptable.
  Leaderless systems are ideal for high-volume, high-availability workloads, but require discipline in modeling data and tuning consistency levels.
• Be explicit about conflict resolution logic — don’t leave it to “last-write-wins” unless you’re sure it’s safe.
  CRDTs for collaborative documents (e.g., real-time editors), Mergeable shopping carts or wishlist updates, Custom logic like “merge tags” instead of replacing them.
• Monitor replication lag and handle stale reads carefully.
  Profile update not reflecting immediately, Message sent but not appearing in chat UI. Stronger guarantees (read-your-writes) may require reading from the leader, increasing latency.

Replication is where performance, availability, and consistency collide. Knowing how to navigate that collision is what separates a reliable system from a fragile one.