Requirements

Functional Requirements:

• Support put/set a value by key, get/retrieve a value by key, and delete/remove a key-value pair.
• Replicate data across multiple nodes for fault tolerance and high availability.
• Partition data across multiple nodes in the cluster to enable horizontal scaling.
• Provide options for persisting data to disk to safeguard against data loss.
• Support conditional writes where a value is updated only if a condition is met, enabling compare-and-swap (CAS) semantics.
• Allow keys to have a time-to-live (TTL) after which they are automatically expired and removed.

Non-Functional Requirements:

• scalability, the system should be able to scale up in case of high spikes, by adding multiple instances
• high availability on both reads and writes, handling at least 200 operations/ms and availability uptime should be at least 99.99%. also there should be no single point of failures.
• handling failures, available nodes should be able to pick up th work load and to give back the control when a non working node comes back up
• handling outages, there should be multiple data centers spreaded which will handle outages
• consistency - we could use eventual consistency which is a form of weak consistency. given enough time the data will eventually become consistent - this way it won't be that heavy on writes and async it will bring up to date all the other replicas
• low latency is a must, even if the responses are eventually consistent, the responses should be fastly received, less than 200ms for an operation

API Design

PUT /key/:{key} body: {value: string} this updates a value by its key

GET /key/{key} retrieves the value for a given key

DELETE /key/{key} deletes a key-value pair

in order to remove duplicates and inconsistencies we could use updates only with certain conditions on the PUT endpoint (the first one in this section) like versioning and vector clocks. only after checking the versions the update will be done

High-Level Design

the client will make a call, either get, set, etc. the call will reach load balancer which routes it to coordinator node. this one will be responsible to route it to the correct node within the has ring. we could you consistent hashing: we have a number of N nodes spread across a circle; whenever we want to add a new operation, we place it in the circle and map it to the closest node to the right. this will ensure that in case of removing one node, we will have to recompute only the ones to its left and not the entire entries.

replication manager will be the one handling scales up and the one actually computing the matching node for an entry using a hash function and applying it to the key. it will be using version vectors to handle concurrent operations and ensuring consistency: for example we will save data[{node,version}] where node is the node the operation will be routed to and version will keep getting incremented on each update of the same data. this will be an array where we keep adding the versions. when the memory is being restrained, we will remove the older entries giving priority to the new ones.

hash ring is the upper method i mentioned where we route the data to a specific node to its right, you can scale up this by simply adding multiple nodes, which are replicas of one main node. these communicate between them using gossip protocol, meaning that a node sends requests to a random set of other nodes and the next node forwards to some other nodes and so on. when multiple nodes start receiving no responses from let's say one specific node, it means that the node might be down, so it checks with a failure detector that afterwards pushes all the writes directed to that node to a message queue to assure consistency when the server is back up. once it's up it start updating it, until then, the node to its right will be taking its place.

for read repairs, we could use merkle trees: each node maintains a Merkle tree for its data partitions. The tree is built by hashing individual key-value pairs at the leaf level, then recursively hashing pairs of hashes up to the root.

during anti entropy (checking consistency between the values of different replicas) nodes exchange only the root level hashes. If roots match, data is consistent. If roots differ, they traverse down the tree level by level — comparing child hashes to identify exactly which data ranges have diverged. This minimizes data transfer since only the differing branches need comparison.

Once inconsistencies are found, read repair kicks in: the node with stale data fetches the correct values from a replica with newer data (determined by version vector timestamps). This happens asynchronously in the background, but can also be triggered during read operations — if a coordinator reads from multiple replicas and detects version mismatches, it repairs the stale replica inline."

ttl manager can be used to evict expired nodes and all their data in order not to consume too much memory useless.

each node will look like this internally: it has in memory stored results (these will be the most read results), and in case of missing results it will check the storage engine. we could use sorted strings tables to get a faster access by the key, by searching them lexicographically

for space reclamation, we use a compaction process. when keys are deleted, they are marked rather than immediately removed. During compaction, multiple sstables are merged together and it removes the keys marked for deletion that have exceeded the grace period

Detailed Component Design

for storage engine I would go with is SSTable. this way all the keys would be sorted lexicographically and whenever an update occurs it will be processed faster. also during deletions they will start as a soft delete (entries will be marked as in need of deletion, also adding a sunset time) and when the time comes, the tables will get compacted and hard delete the marked entries. afterwards the disk space will be reclaimed and used for compaction. Each sstable will have a checksum and in case of corruption, data from a different node will be used. during replication, checksums are also transmitted in order to ensure data integrity on arrival - eg a hash function applied to the data

for replication we could use a replication factor of N (typically it is 3), meaning we store data in N different nodes for fault tolerance. the data gets a hash function applied to it in order to find out where to store it in the hash ring. the main node will be the first one to its right in a clock wise direction. the second store will be the next one and so on. the nodes will be spreaded across multiple availability zones in such manner that if an entire region is down, the others will take its place, ensuring hardware faults are handled.

the coordinator node is the one routing the data to its specific node in the hash ring, after the hash results have been computed. as i've already described in the upper paragraph, the data will be routed to the first to the right node in a clockwise direction

Failure detection:

Nodes communicate via gossip protocol - they send requests to each other and when one node stops responding cu multiple nodes (usually a pre-configured value) then it is markes as possibly down.

after a pre-configured recovery period if the node hasnt come back up, it is marked as failed and a different node (the one to its right) will take over its values until it gets back up.

whilst the recovery period isnt over, all that node's writes are added to a hinted handoff queue and could be saved to a different table as "hints" that they belong to a different node which is down. when the node is back up, it will read these hint as they will be handoffed to it and it will start writing them on its side

for hot keys and frequently used data, we could cache it in memory using LRU strategy - we will discard the least used cached items. only in case the results arent cached, we will fetch it from the sstable