System requirements

Functional:

1. Basic Operations:
   • Set: Store a key-value pair in the system.
   • Get: Retrieve the value associated with a given key.
   • Delete: Remove a key-value pair from the store.
   • Conditional Update: Update a value only if a certain condition on the existing value is met.
2. Data Replication:
   • The system should replicate data across multiple nodes to enhance fault tolerance and availability.
3. Partitioning:
   • Support for horizontal scaling by partitioning the data across various nodes to ensure load balancing.
4. Persistence:
   • Options to persist data to disk to prevent data loss in case of system crashes or failures.
5. High Availability:
   • Ensure that the system remains operational and accessible, even during failures of some components.
6. Consistency:
   • Provide guarantees around consistency, ensuring that all reads return the most recent write (if needed).
7. Performance:
   • Ability to handle high volumes of read/write operations and provide low-latency responses.
8. Data Expiration:
   • Support for setting expiration times on keys for automated cleanup.
9. Batch Operations:
   • Allow batch writes and reads for better performance with high volume transactions.
10. Monitoring and Management:
    • Provide metrics and logs to monitor system performance and health.

Non-Functional:

1. Performance:
   • The system should provide low latency for operations such as get, set, and delete, ideally in the millisecond range, to ensure swift responses even during high load.
2. Scalability:
   • The system should be able to scale horizontally by adding more nodes without significant reconfiguration or downtime. It should handle increased load efficiently as the amount of data grows or the number of clients increases.
3. Availability:
   • Aim for high availability, with a target uptime of 99.99% or higher, ensuring that the service remains accessible even during failures or maintenance.
4. Durability:
   • Any committed data should be preserved across failures. This can involve strategies like writing to persistent storage before acknowledging successful writes.
5. Fault Tolerance:
   • The system must gracefully handle failures of nodes or network partitions without losing data or compromising service availability.
6. Security:
   • Implement authentication, authorization, and encryption mechanisms to protect data at rest and in transit.
7. Maintainability:
   • The system should be designed for easy upgrades and maintenance. This involves clear logging, monitoring capabilities, and the ability for administrators to manage nodes with minimal effort.
8. Usability:
   • If applicable, provide an intuitive API for developers to interact with the system easily, including clear documentation.
9. Data Consistency:
   • Depending on the choice between eventual consistency or strong consistency, clarify the requirements around data correctness and how conflicts are resolved.
10. Monitoring and Logging:
    • The system should provide metrics and logs for tracking performance, usage, and error rates to facilitate troubleshooting and performance tuning.

Capacity estimation

Estimate the scale of the system you are going to design...

API design

POST /set&key=<key>

GET /get&key=

POST /set/conditional&key=

DELETE /&key=

Database design

Database design is not necessary as the data will be represetned as a hash map in moemry with values residing on disc

High-level design

|System contains of the following elements:

1) Cluster coordinator - a node responsible for coordinating read/write requests among replica nodes

2) Cluster node - a part of the cluster that handles read and write requests, can be elected as a new coordinator

Each cluster node is a self sufficient element of the cluster and can take the role of a coordinator if needed thats why it contains the following parts:

1) Monitoring system - monitors for failures

2) Client API - and api to accept read/writes from client in case this node is elected a coordinator

3) Bloomfilter - a data strctureu to quikcly check the presence of a key on disc

4) Self repair and replication modules - self explanatory

5) Memory log - records data before flushing it to disc, also known as a write ahead log

6) Disc - tape disc to permanenly store data

Request flows

The diagram is on the high level design. This is the request flow in high level.

1. clients use APIs to communicate with the key-value store: get() and put()
2. client sends read/write request to the coordinator node which acts as a proxy between client and server nodes.
3. Nodes are distributed and are spread on a hash ring
4. Nodes are decentralized, they can auto scaling depending on the load.
5. data is replicated onto multiple nodes.
6. No single point failure as each node worked the same responsibility: client APIs, data replica, failure detection, fix failure, etc.

1) Before any requests can come in a cluster elects a coordinator using gossip procotol whre each node votes for a new coordinator

2) When coordinator is elected is starts to accept requesst from a client

3) Depending on the consistency and avaialblity settings when a new requests comes in coordinator waits for a predetermined number of replica acks before signalling that request is complete. We use the following formula to provide consistency guarantee:

R + W > N - strong consistency, where R - number of read acks, W - number of write acks, N - number of nodes in the cluster

R + W <= N - weak consitency

Replica in/out flow:

1) When replica node wants to exit a cluster voluntarily or by failure the coordinator keeps a consistency hashing ring which rebalances data across multiple nodes. In case of key change new node is placed (or removed from) on the ring and some portion of the data has to be rebalanced

Replica health check:

1) To maintanin a view of healthy nodes each node has a list of neighbours that ite periodically checks using a heartbeat protocol

2) The list of neighbours is formed randomly and has a fixed length to not flood the noetwork

3) If the node doesn't respond withing a time frame it is considered offline and it's stauts is propagated to every node using gossip protocl and eventually to a coordinator node which removes it's from a consistency hashing ring

4) If 2 or more nodes agree on a particular node being offline the node is marked as an offline by a coordinator

Conflict resolution for read/writes

1) We use a vector clock approach where every data written to a replica has a version number associated with it (a clock value)

2) We 2 or more read/write requests come from a different replicas at the same time we can check if their versions have a conflict and apply a dedicated conflict resolution startegies like last write wins or resolve manually to resolve the conflict

3) Read conflicts are at teh mercy of the client and our read response always gives back a vector clock value if multiple replicas are involved

Writing data on disc

1) Evey replica contains an in-memory write ahead log that is populated first and if it becomes full the data is lfushed to disc. This is done ti have to batch write requests to disc as writing small chunks of data will influence disc write latency

Reading data from disc

1) The oS kernel caches aggresively any data read from disc in memory and thus it's out first read area when issuing |GET /key

Detailed component design

1. Consistency
   1. To ensure all clients get the same data when reading the same key from different nodes, the data needs to propagate to other nodes when this data is changed or newly created. There are following consistency models:
      1. strong consistency: all read/write operation will be blocked until all replicas agree on this write. This will lower the system performance. It is usually used in the scenario when the high consistency is needed, like bank system.
      2. weak consistency: the client may get the outdated data from other nodes after the write on one node.
      3. eventual consistency: it is a specific weak consistency. Given enough time, the new data will be propagated to other nodes. Based on the requirements, we will use this consistency model.
   2. Consistency Resolution
      1. when two clients write to the same key-value on different nodes, their value will be conflict.
      2. We can use vector clock [ServerNumber, DataVersion] to get the final value by determining the preceding / succeeding relationship of the two data.
         1. downside of the vector clock:
            1. add complicated logic
            2. the vector clock may grow rapidly. This can be solved by removing the outdated [server, version] pair by a predefined threshold.
2. Failure Detection
   1. usually we can't consider a node down from a single source. We need to get it from at least two independent source.
   2. Decentralized Failure Detection (Gossip Protocol):
      1. Each node maintains a node member list which has the memberID and heartbeat counter.
      2. each node periodically increment the heartbeat counter
      3. Each node periodically sends the heartbeat to a set of random other nodes which in turn propagate to other nodes.
      4. If a node heartbeat counter was not incremented for a predefined period, it is considered as offline.
   3. Handling failure:
      1. if a node is down, another nodes will handle the requests temporarily. When the node is up, changes will be pushed back to it to achieve data consistency. This is called "hinted handoff

Every replica node is self sufficient in a sense that it can take the role of a coordinator at any time it contains:

• Disc storage
• Memory cache
• A list of nieghobouurs to check heartbeat
• Client API in case it will become the coordinator
• Self repair module - in case the data on disc becomes corrupted
• Conflict resolution module - to resolve conflict when multiple read/write requests come to the replica
• Monitoring module - that sends monitoring info to an external service
• Mmoery hash map is based on the array data structure that holds a list of key-value pairs. If differnet key hashes to the same bucket we keep a list of key-value paris (collisions) and perform a linera scan when retrieving the keyy. We can optimized this approach by using a b-tree instead of a list

Trade offs/Tech choices

We choose distributed system with a dedicated coordinator module to provide strong constiency in distributed environment and also to make client API easier

We choose consistent hashing to decrease the number of data transfer between replicas when their numbers change

We choose merkle trees to check if any data is corrupted on disc

We choose vector clock as the easiest way to hadle conflicts in a distributed system and limit the number of clock history to prevent memory overflow

We use write-ahead memory log to batch disc writes before flushing them to disc

We use bloom filter to perform a quick check of the key in memory before performing a disc seek

When cache becomes full we LRU strategy to evict nodes and make space

When the node fials to acknowledge read or write request with use a limited exponential backoff to retry the operation

Using the formular R + W > N we can control what system is optimized for, for examepl:

if R = 1, the sytem will be optimized for reads as only 1 read acks will be neaded for read operation to be successfull

if W = 1, the system is optimized for writes in the same manner

if R + W > N, we get strong consitency model

if R + W <=N we get a weak consitency model

To balance the load on the hash ring with duplicate our nodes and call the virtual nodes, meaning that the range of requests will be mapped to the same node instead of a different node every time

To make writes durable we repliate data on W nodes where W is a configurable parameter

To handle coordinator overload we can use a rate limiter installed in every replica or use a multi coordiantor setup behind a dedicated load balancer service.

For a geo replicated data we can position our replicas close to the expected clients location

In case of a data not fitting a single node we can split data into chunks a nd store chunks as a spearate key value pairs

In case of netwrok partiiong the system will serve requests based on the existing nodes and when the ystem will get back up using the gossip protocol the data will be synchronized between the remaining nodes

Potentially to store an asbolutely huge files we can use an external blob storage and store links to it in our database instead of values

Failure scenarios/bottlenecks

In case of a replica crash it's neighbours will get a heartbeat timeout everntually and will propagate it's status to a coordinator (eventually) so that it can remove it from a hash ring

If the replica becomes overloaded it can start replying with 429 too many requests and submit it's status to a coordinator to spin up a new replica to handle the load

In case of a frequent replica entering/leaving the cluster a system can be temporarily overloaded with a coordinatorinon/rebalance messages

In case we setup heartbeat protocol for every replica to check for every neighbour the system can become overloaded with a lot of heartbeat messages

Future improvements

1) user centralized log aggreagtor to monitor the system

2) Use ML with centralized nodes to predict the future load on the system and scale accordingly

3) Use adaptive stragety based on monitored data to predict what replica might crash soon or get overloaded