DynamoCore - Distributed Key-Value Store with High Availability and Scalability

A from-scratch implementation of Amazon's Dynamo paper in Pure Java.

Architecture · Concepts · API · Tech Stack · Quick Start · Milestones · Paper Mapping · Project Structure · Results · Improvements · References

What Is DynamoCore?

DynamoCore is a complete, production-grade distributed key-value store built from first principles, directly implementing every concept from the landmark Amazon Dynamo paper (DeCandia et al., SOSP 2007).

The system is designed to answer one question:

How do you build a storage system that is always available for writes - even when servers are crashing, network links are flapping, and entire data centers are going offline?

Amazon's answer was Dynamo. This project implements that answer - every section of the paper, every algorithm, every tradeoff - in Pure Java across 8 progressive milestones.

📌 Core Engineering Documents

Document	Description
System Benchmarks	Implemented & Measured throughput limits, p99 latencies, and performance costs of replication, vector clocks, and anti-entropy, added across the milestones.
Challenges & Fixes	Technical deep-dive into debugging distributed quorum failures, network serialization traps, and gossip state sync issues and how they were systematically resolved.
Paper vs. Reality	7 key design decisions that changed between the 2007 Dynamo paper and the modern DynamoDB service.

CAP Theorem Position

DynamoCore is an AP system (Available + Partition-tolerant).

      C (Consistency)
      |
      |   CP systems                AP systems
      |   Zookeeper, etcd,          DynamoCore, Cassandra,
      |   Google Spanner            CouchDB, Riak
      |
      +--------------------------------- P (Partition Tolerance)
 A
(Availability)

When a network partition occurs in DynamoCore

Writes are accepted on any W healthy nodes (sloppy quorum)
Some nodes may temporarily return stale data
Replicas converge when the partition heals (gossip + anti-entropy)

When to use DynamoCore's Availability model : Shopping carts, sessions, user preferences, product catalogs - any case where rejecting a write is worse than a temporary inconsistency.

When NOT to use this model : Bank balances, inventory counters, anything requiring strict serialisability. Use a CP system (Google Spanner, CockroachDB) for those.

Architecture Implemented

                        +-----------------------------+
                        |          Client             |
                        |  PUT /kv/{key}  GET /kv/{key}|
                        +-------------+---------------+
                                      |
          +--------------------------++--------------------------+
          |                          |                          |
          v                          v                          v
+------------------+     +------------------+     +------------------+
|     Node 1       |     |     Node 2       |     |     Node 3       |
|   :8080          +<--->+   :8081          +<--->+   :8082          |
|                  |     |                  |     |                  |
|  HTTP Server     |     |  HTTP Server     |     |  HTTP Server     |
|  Consist. Hash   |     |  Consist. Hash   |     |  Consist. Hash   |
|  Replication     |     |  Replication     |     |  Replication     |
|  Vector Clocks   |     |  Vector Clocks   |     |  Vector Clocks   |
|  Sloppy Quorum   |     |  Sloppy Quorum   |     |  Sloppy Quorum   |
|  Gossip     <----+-----+---> Gossip  <----+-----+---> Gossip       |
|  Merkle Trees    |     |  Merkle Trees    |     |  Merkle Trees    |
|  RocksDB /       |     |  RocksDB /       |     |  RocksDB /       |
|  InMemory        |     |  InMemory        |     |  InMemory        |
+------------------+     +------------------+     +------------------+
         |                       |                       |
         +-----------------------+-----------------------+
                 Gossip Protocol + Anti-Entropy
               (decentralised, no single point of failure)

Design Principles (from the Dyanamo Paper)

Principle	What It Means	How DynamoCore Implements It
Always Writeable	Writes must never be rejected	Sloppy quorum uses backup nodes when primaries are down
Eventual Consistency	Replicas converge, not instantly consistent	Vector clocks detect divergence; Merkle trees repair it
Decentralisation	No single coordinator or master	Gossip protocol - every node is equal
Incremental Scalability	Add nodes without restarting	Consistent hashing - only K/N keys move
Symmetry	Every node has the same responsibilities	Any node can coordinate any request
Heterogeneity	Different capacity nodes handled gracefully	Virtual nodes - assign more tokens to stronger nodes

Dynamo Concepts Implemented

1. Consistent Hashing with Virtual Nodes

Section 4.2 of the Dynamo paper

The hash space (0 to 2^64) is arranged as a ring. Each physical node is assigned 150 virtual nodes (tokens) at random positions. A key maps to the first node clockwise from murmur3_128(key).

Why virtual nodes?

Uniform load distribution (law of large numbers with 150 tokens)
When a node fails, its load spreads across all remaining nodes - not just two neighbours
High-capacity nodes can be assigned proportionally more tokens

Key implementation: ring/ConsistentHashRing.java

Uses ConcurrentSkipListMap<Long, String> - O(log N) clockwise lookup via ceilingKey()
Adding/removing a node moves only K/N keys on average

2. Replication with Preference Lists

Section 4.3 of the Dynamo paper

Every key is replicated across N=3 nodes (configurable). The coordinator fires all N-1 replication requests simultaneously using CompletableFuture and waits for W acknowledgments.

PUT cart:user99
       |
       v
  Coordinator (Node 1)
  +-- Write locally ---------------------- ACK 1
  +-- Async replicate to Node 2 --------> ACK 2   <- W=2 reached -> SUCCESS
  +-- Async replicate to Node 3 --------> ACK 3   (still in flight, ignored)

Latency = slowest of first W responses, not sum of all N. This is why the paper reports sub-300ms p99 latencies despite touching 3 nodes.

3. Data Versioning with Vector Clocks

Section 4.4 of the Dynamo paper

Every version of every object carries a vector clock - a map of {nodeId -> counter}. This tracks causality between versions without relying on physical timestamps (which cannot be trusted across machines due to clock drift).

Client writes initial value  ->  D1: {Sx:1}
Same client updates          ->  D2: {Sx:2}       (D1 happened-before D2)

Network partition. Two clients update D2 simultaneously:
  Client A via node Sy       ->  D3: {Sx:2, Sy:1}
  Client B via node Sz       ->  D4: {Sx:2, Sz:1}

D3 || D4  (CONCURRENT - neither happened-before the other)
-> CONFLICT detected. Both versions returned to application.

Client merges D3 + D4, writes D5:
  D5: {Sx:3, Sy:1, Sz:1}  <- subsumes both D3 and D4. Conflict resolved.

Three relationships between any two clocks:

Relationship	Condition	Action
`VC1 -> VC2` (before)	All counters in VC1 <= VC2	VC1 is stale - discard it
`VC2 -> VC1` (after)	All counters in VC2 <= VC1	VC2 is stale - discard it
`VC1		VC2` (concurrent)

Key implementation: model/VectorClock.java, versioning/Reconciler.java

4. Sloppy Quorum + Hinted Handoff

Section 4.6 of the Dynamo paper

A strict quorum requires W acks from the exact designated W replicas. A sloppy quorum accepts W acks from any W healthy nodes - even backup nodes.

Quorum Consistency Equation:

$$\text{Replication Factor } (N) = 3, \quad \text{Write Quorum } (W) = 2, \quad \text{Read Quorum } (R) = 2$$

$$R + W > N \quad (2 + 2 > 3)$$

Because $R + W > N$, the read quorum set and write quorum set are guaranteed to overlap by at least one node, ensuring strong consistency guarantees on reads under normal network conditions.

Sloppy Quorum Availability: In the event of physical node outages, write requests are accepted by any healthy successor node in the hash ring up to $N + \text{backups}$, satisfying the $W=2$ requirement and preserving the always-writeable availability property.

Ring: [A, B, C, D, E, F]   Key K's preference list: [A, B, C]   N=3, W=2

Normal:
  Write -> A (ACK 1), B (ACK 2), C (ACK 3)  -> W=2 reached -> SUCCESS

Node A is DOWN:
  Write -> B (ACK 1), C (ACK 2)  -> W=2 reached -> SUCCESS
  D accepts a hinted write on behalf of A.
  Hint stores: { key=K, value=V, intendedNode=A }

A recovers -> D delivers hint to A -> D deletes hint

The shopping cart always gets the item added. Writes are never rejected.

Key implementation: quorum/SloppyQuorum.java, quorum/HintedHandoffStore.java, quorum/HintDeliveryTask.java

5. Gossip Protocol + Failure Detection

Section 4.8 of the Dynamo paper

Every second, each node picks a random alive peer and exchanges membership tables. Information about any event propagates to all N nodes in O(log N) rounds.

Node A's view:               After gossip with B:
  A: heartbeat=10              A: heartbeat=10
  B: heartbeat=4   --gossip--> B: heartbeat=9   (B's counter went up -> B is alive)
  C: heartbeat=7               C: heartbeat=7
                               D: heartbeat=3   (new node discovered via B)

Failure state machine:

ALIVE --(silent > 5s)--> SUSPECT --(silent > 10s)--> DEAD
  ^                                                     |
  +-------------------(heartbeat received)--------------+

When a node is declared DEAD, it is removed from the consistent hash ring. When it recovers, it is added back automatically.

Key implementation: gossip/GossipProtocol.java, gossip/FailureDetector.java

!["Gossip Protocol and Merkel Anti-Entropy](./static/Gossip Protocol and Merkel Anti-Entropy.jpeg)

6. Merkle Tree Anti-Entropy

Section 4.7 of the Dynamo paper

After failures and partitions, replicas can diverge. Anti-entropy repairs this silently in the background every 60 seconds.

The problem: Two nodes may have 1,000,000 keys. Only 10 differ. How do you find the 10 without transferring all 1,000,000?

Merkle tree solution:

         Root Hash
        /          \
   H(left)       H(right)         <- Compare hashes level by level
   /     \       /     \          <- Skip equal subtrees entirely
 H(k1) H(k2) H(k3) H(k4)        <- O(D x log N) comparisons
                                     where D = number of differences

Sync protocol:

Node A sends root_hash_A to Node B - O(1) check
If equal -> in sync, done. If different -> continue.
Exchange child hashes level by level, skip equal subtrees
Identify exactly which keys differ
Transfer only the differing key-value pairs

For 1,000,000 keys with 10 differences: ~40 hash comparisons, 10 key transfers. Without Merkle trees: 1,000,000 hash comparisons.

Key implementation: antientropy/MerkleTree.java, antientropy/AntiEntropyService.java

API Reference

`PUT /kv/{key}` - Store a value

curl -X PUT http://localhost:8080/kv/cart:user99 \
     -H "Content-Type: application/json" \
     -d '{"value": "[book, laptop]"}'

{
  "key": "cart:user99",
  "status": "stored",
  "acksReceived": 2,
  "nodes": ["node1", "node2"]
}

`GET /kv/{key}` - Retrieve a value

No conflict (normal case):

{
  "key": "cart:user99",
  "hasConflict": false,
  "context": "eyJjbG9jayI6eyJub2RlMSI6MX19",
  "value": "[book, laptop]",
  "clock": {"node1": 1}
}

Conflict detected (concurrent writes from two clients):

{
  "key": "cart:user99",
  "hasConflict": true,
  "context": "eyJjbG9jayI6eyJub2RlMSI6Miwibm9kZTIiOjF9fQ==",
  "versions": [
    {"value": "[book, laptop]", "clock": {"node1": 2}},
    {"value": "[book, phone]",  "clock": {"node1": 1, "node2": 1}}
  ],
  "message": "Conflict detected. Merge versions and PUT with the context token."
}

`PUT /kv/{key}` with context - Resolve a conflict

The client merges conflicting versions and writes back with the context token from the previous GET. The context tells the coordinator which version(s) this write supersedes.

curl -X PUT http://localhost:8080/kv/cart:user99 \
     -H "Content-Type: application/json" \
     -d '{
       "value": "[book, laptop, phone]",
       "context": "eyJjbG9jayI6eyJub2RlMSI6Miwibm9kZTIiOjF9fQ=="
     }'

`DELETE /kv/{key}` - Logical delete (tombstone)

curl -X DELETE http://localhost:8080/kv/user:alice

Deletes write a tombstone, not a physical removal. This prevents deleted keys from "resurrecting" when a stale replica replicates an old version.

Operational Endpoints

Endpoint	Method	Description
`/admin/health`	GET	Node status, memory, uptime, key count
`/admin/members`	GET	Gossip membership: ALIVE / SUSPECT / DEAD
`/admin/antientropy`	GET	Merkle tree stats and root hash
`/admin/keys`	GET	All keys in local storage
`/ring/info`	GET	Ring nodes, token distribution
`/ring/lookup/{key}`	GET	Preference list for a key
`/internal/antientropy/root`	GET	Merkle root hash (for peer comparison)
`/internal/antientropy/trigger`	POST	Manually trigger anti-entropy round

Tech Stack

Component	Technology	Why This Choice
Language	Java 21	Modern concurrency, switch expressions, records
HTTP Server	JDK `com.sun.net.httpserver`	Zero external dependency - shows HTTP understanding
JSON	Jackson 2.17	Industry standard, handles nested objects (vector clocks)
Hashing	Guava MurmurHash3 (128-bit)	Fast, non-cryptographic, excellent distribution
Storage	`ConcurrentSkipListMap` / RocksDB	In-memory for dev; RocksDB for production
Async I/O	Java `CompletableFuture` + `HttpClient`	Native async without framework overhead
Build	Maven 3.8+	Standard, reproducible
Container	Docker + Docker Compose	One-command 3-node cluster

Quick Start

Prerequisites

java -version    # Java 17 or 21
mvn -version     # Maven 3.8+

Build

git clone https://github.com/NayakSubhransu/DynamoCore.git
cd DynamoCore
mvn clean package -DskipTests    //Build Command

Run a 3-node cluster

# Three terminals:
java -jar target/dynamo-kv-1.0.0.jar config/node1.properties
java -jar target/dynamo-kv-1.0.0.jar config/node2.properties
java -jar target/dynamo-kv-1.0.0.jar config/node3.properties

# Or with the script:
./scripts/start-cluster.sh

# Or with Docker:
docker-compose up -d

Verify

curl http://localhost:8080/admin/health | jq .

curl -X PUT http://localhost:8080/kv/user:alice \
     -H "Content-Type: application/json" \
     -d '{"value": "alice@example.com"}'

# Read from a DIFFERENT node (proves replication)
curl http://localhost:8082/kv/user:alice | jq .value
# -> "alice@example.com"

Full demo

./scripts/demo.sh

Configuration

# config/node1.properties

node.id=node1
node.host=localhost
node.port=8080

# The core tradeoff knobs (from the paper)
quorum.n=3          # total replicas per key
quorum.r=2          # nodes that must respond to a read  (R+W > N = consistent)
quorum.w=2          # nodes that must ack a write

storage.engine=memory            # "memory" (dev) or "rocksdb" (production)
storage.data.dir=data/node1

peers=localhost:8081,localhost:8082

ring.tokens.per.node=150         # virtual nodes per physical node
vectorclock.max.entries=10       # max vector clock size before truncation

gossip.interval.ms=1000          # gossip every 1 second
gossip.suspect.threshold.ms=5000 # 5s silence -> SUSPECT
gossip.dead.threshold.ms=10000   # 10s silence -> DEAD

antientropy.interval.ms=60000    # background sync every 60 seconds

NRW Tradeoff Configurations

Config (N,R,W)	Use Case	Tradeoff
N=3, R=2, W=2	Default - balanced	R+W=4 > 3: consistent. Tolerates 1 failure.
N=3, R=1, W=3	Read-heavy, rarely updated	Fast reads. Writes hit all 3 nodes.
N=3, R=3, W=1	Write-heavy, eventual reads ok	Writes always succeed. Reads check all.
N=3, R=1, W=1	Maximum availability	No consistency guarantee (R+W=2 < N=3)

Implementation Milestones

DynamoCore was built layer-by-layer. Every milestone is independently runnable and demoable.

Milestone	Feature	Paper Section	Key Classes
M1	Single-node KV Store + HTTP API	-	`api/KVHandler`, `storage/InMemoryStorage`
M2	Consistent Hash Ring + Virtual Nodes	§4.2	`ring/ConsistentHashRing`, `ring/RingManager`
M3	N-Replica Replication + Quorum	§4.3, §4.5	`replication/ReplicationManager`, `replication/NodeClient`
M4	Vector Clocks + Conflict Detection	§4.4	`model/VectorClock`, `versioning/Reconciler`
M5	Sloppy Quorum + Hinted Handoff	§4.6	`quorum/SloppyQuorum`, `quorum/HintedHandoffStore`
M6	Gossip Protocol + Failure Detection	§4.8	`gossip/GossipProtocol`, `gossip/FailureDetector`
M7	Merkle Tree Anti-Entropy	§4.7	`antientropy/MerkleTree`, `antientropy/AntiEntropyService`
M8	Docker + Demo + Polish	-	`Dockerfile`, `docker-compose.yml`, `scripts/`

Dynamo Paper To Code Mapping

Paper Section	Concept	Implementation	Key Method
§4.2	Consistent hashing	`ring/ConsistentHashRing.java`	`getPreferenceList()`
§4.2	Virtual nodes	`ring/ConsistentHashRing.java`	`addNode()` - 150 tokens
§4.3	Replication	`replication/ReplicationManager.java`	`coordinateWrite()`
§4.4	Vector clocks	`model/VectorClock.java`	`increment()`, `happenedBefore()`
§4.4	Conflict detection	`versioning/Reconciler.java`	`reconcile()`
§4.4	Clock truncation	`model/VectorClock.java`	`truncateIfNeeded()`
§4.5	Read coordinator	`replication/ReplicationManager.java`	`coordinateRead()`
§4.5	Read repair	`replication/ReplicationManager.java`	`performReadRepair()`
§4.6	Sloppy quorum	`quorum/SloppyQuorum.java`	`sloppyWrite()`
§4.6	Hinted handoff	`quorum/HintedHandoffStore.java`	`store()`, `remove()`
§4.6	Hint delivery	`quorum/HintDeliveryTask.java`	`runDeliveryRound()`
§4.7	Merkle tree	`antientropy/MerkleTree.java`	`findDifferences()`
§4.7	Anti-entropy	`antientropy/AntiEntropyService.java`	`runAntiEntropyRound()`
§4.8	Gossip protocol	`gossip/GossipProtocol.java`	`gossipRound()`
§4.8	Failure detection	`gossip/FailureDetector.java`	`detect()`
§4.8	Membership	`gossip/MembershipTable.java`	`mergeGossip()`

Full detailed mapping: docs/dynamo_paper-to-code-mapping.md

Project Structure

Running Tests

mvn test

Test Class	What It Proves
`VectorClockTest`	Figure 3 of the paper reproduced: D1->D2->D3/D4 conflict->D5 merge
`ConsistentHashRingTest`	Adding a node moves only ~1/N keys. Preference list has N distinct nodes.
`ReconcilerTest`	Syntactic reconciliation removes dominated versions. Concurrent versions both survive.
`ReplicationTest`	Write to one node is visible on all replicas. Tombstone propagates correctly.
`SloppyQuorumTest`	Hints stored, retrieved, deduplicated, removed after delivery.
`GossipTest`	Gossip merge takes maximum counter. New node auto-discovered. ALIVE/SUSPECT/DEAD.
`MerkleTreeTest`	1000-key tree with 1 difference - finds it without comparing all 1000.
`EndToEndTest`	Complete shopping cart: write, concurrent conflict, merge, tombstone, Merkle.

Performance Characteristics

From the original Dynamo paper's production measurements (similar N=3, R=2, W=2 configuration):

The Complete code has been successfully executed, and the complete performance benchmark results are stored at docs/benchmarks.md. Below is a summary table demonstrating the cumulative cost and throughput impact of each distributed systems feature added across the milestones:

Milestone	Feature Added	Write p99	Read p99	Throughput	Overhead vs M1
M1	Single node	2ms	3ms	955 writes/s	baseline
M2	+ Hash ring	7ms (lookup)	3ms	455 ring lookups/s	~0% (Ring lookup overhead ~0ms)
M3	+ Replication N=3 W=2	12ms	10ms	233 writes/s	~75% throughput drop (~3.1x slower)
M4	+ Vector clocks	9ms	7ms	Not measured separately	Negligible (~0ms overhead)
M5	+ Sloppy quorum	33ms	Not measured	109 writes/s	~88% throughput drop
M6	+ Gossip (1s rounds)	30ms	Not measured	166 writes/s	~83% throughput drop
M7	+ Anti-entropy (60s)	Not measured	Not measured	Background service (O(1) root hash check 6ms p99)	Negligible in steady state

Known Limitations & Future Improvements

1. Partitioning Strategy (Strategy 1 -> Strategy 3)

Current: Random tokens (Strategy 1) - simple, the original Dynamo approach. Improvement: Equal-sized partitions with Q/S tokens (Strategy 3).

Benefits: faster node bootstrapping, stable Merkle tree boundaries, three orders of magnitude less metadata per node. The Dynamo paper (Section 6.2) shows Strategy 3 achieves the best load balance efficiency.

2. Dotted Version Vectors

Current: Standard vector clocks with truncation at 10 entries. Problem: Truncation can make two versions appear concurrent when one actually descended from the other, causing unnecessary client reconciliation. Improvement: Dotted Version Vectors (DVV) - eliminates false conflicts from truncation entirely. Used by Riak 2.0+ (Preguiça et al., 2012).

3. Multi-Datacenter Awareness

Current: All nodes treated as a flat pool. Improvement: Tag nodes with datacenter IDs. Construct preference lists that span DCs, ensuring full datacenter outages do not cause data loss. This is how production Dynamo handles Amazon's regional infrastructure.

4. Per-Request Tunable NRW

Current: N, R, W are global config values applied to every request. Improvement: Allow per-request NRW via HTTP headers (X-Dynamo-R: 1 for a fast read). Amazon services use this to tune their own SLAs without redeploying Dynamo.

References

DeCandia, G. et al. (2007). Dynamo: Amazon's Highly Available Key-value Store. SOSP '07, 205-220. PDF
Lamport, L. (1978). Time, clocks, and the ordering of events in a distributed system. CACM 21(7), 558-565. (Basis for vector clocks)
Merkle, R. (1988). A digital signature based on a conventional encryption function. CRYPTO '88, 369-378. (Basis for Merkle tree anti-entropy)
Preguiça, N. et al. (2012). A Brief History of Consistent Hashing and Dotted Version Vectors. (Future improvement: replaces vector clocks)
Elhemali, M. et al. (2022). Amazon DynamoDB: A Scalable, Predictably Performant, and Fully Managed NoSQL Database Service. USENIX ATC '22. Paper (Details the modern, fully managed evolution of the original Dynamo architecture)

Built with deep respect for the original Amazon Dynamo engineering team.

Connect with me on LinkedIn : Subhransu Priyaranjan Nayak.

(back to top)

Name		Name	Last commit message	Last commit date
Core Algos		Core Algos
References		References
benchmark		benchmark
config		config
docs		docs
scripts		scripts
src		src
static		static
.gitignore		.gitignore
Dockerfile		Dockerfile
Makefile		Makefile
README.md		README.md
dependency-reduced-pom.xml		dependency-reduced-pom.xml
docker-compose.yml		docker-compose.yml
pom.xml		pom.xml

Latest commit

History

Folders and files

Repository files navigation