Requirements

Functional Requirements:

1. The system should be able to support set, get, delete and conditional updates
2. The system should be able to efficiently handle high volumes of read/write operations.
3. The system should be able to evict items either with a TTL or a LRU eviction policy based on user configuration.

Non-Functional Requirements:

1. The system should be available - but with eventual consistency. If it is down, other applications upstream and downstream lose out on sub millisecond latency reads potentially.
2. The system should be fault tolerant with a graceful recovery from any failures.
3. The cache should have a replication factor of 3 at least.
4. The system should be an in memory cache with data loss at restart being acceptable. Storing it in a durable store with SSTable, WAL snapshots potentially increases read latency.
5. Data is partitioned across nodes using consistent hashing so that no single node holds all keys,

Capacity Estimation

1. The number of reads in the key value store would likely be of the order of 10:1 - 10 times as many reads as writes.
2. Each of the nodes will be around 16 GB RAM of which roughly 4 GB is taken up by system overhead. Now we will have 12 GB of RAM and based on the type of data stored in each key value pair, we could have roughly 8-16 bytes of data in each row. That translates to at least 750 million rows within a single node. If we expect our volume of data to be 1 million transactions/rows per second, this is likely to be filled up by 750 seconds - 12 minutes. So each node will likely last at least 10-15 minutes before reaching memory issues. If the number of writes are 100000 per second, then each node will likely last longer - 7500 seconds - roughly 2 hours.
3. Assuming million transactions/rows of writes per second with 10 x read ratio we can see at least 10 million reads per second. With a read capacity of 1 ms per query for a node, within a second we can process at least 1000 reads. This does not seem to make sense. So let us go back and expect 10000 writes per second. This translates 100000 reads per second enabling 1 ms per query for a node.

API Design

We will use REST API as our interface. This gives us flexibility to define our TTL semantics.

1. GET - /getValue?key=""
   1. response: {value="",expires_at="",version=""}
2. POST - /putValue
   1. request: {key="",value="",ttl="",if_version:""} -> if ttl is set as the configuration for the expiry and eviction.
   2. response: {value="",putStatus = boolean}
3. DELETE -/deleteKey
   1. request:{key=""}
   2. response:{status: boolean}
4. For admins - POST - /setEvictionConfiguration:
   1. request:{config = "LRU/TTL"}
   2. response:{status: boolean}

High-Level Design

1. To avoid any network hop latencies, in the cache, the api would reside within each node along with data. For example each node would have the data and the doubly linked list. The doubly linked list will have the node containing the pointer to the next node and the hashmap as a data structure in the data.
2. We use consistent hashing for our replication. The hash is computed for the key and within our cluster based on the hash the appropriate node is selected within which the data gets updated/inserted/deleted.
3. Within each node of our cache we use something like a cleaner service or a janitor service to remove any keys which are past the expiry timestamp.
4. Any writes to the cache needs to be acknowledged only by the primary node, with async replication post write for the replicas. If we need a quorum to respond that potentially increases the distributed cache latency. This is a tradeoff that can be decided based on the system this is used for.

Database Design

1. hash map with key and value attributes.
2. A separate doubly linked list - optional in memory structure for every store to keep track of which one was accessed recently. For example, the head doubly linked list would be the recently used item linking to the next node and so on. This would be doubly linked list to enable us traverse in any direction - likely to be optimal for deletion. For example the least recently used would be towards the tail. So whenever capacity is exceeded, the tail is disconnected and moved to the head. This is an O(1) operation.
3. We will use consistent hashing - which uses a ring with the nodes residing on the ring. The replicas would be the immediate following nodes after the node where the key hits. This gives us a RF=3. If we assign it to two other random nodes on the ring, if any node fails or is taken off, the transfer is through a longer path and needs to be kept in a separate hash table or lookup table. Of course with this approach we could run in to hot nodes, which causes a few nodes to always receive traffic. We avoid this by adding virtual nodes.

Detailed Component Design

1. within each node of the cache cluster, the janitor service runs every minute - again this should be able to be set as configurable so that we do not overload a given node. The janitor would step over each node in the doubly linked list and if any nodes have key value past their expiry time, the cleaner service removes that node from the linked list - effectively removing the data from the cache.
2. When data is being updated for an existing key, the version of the existing value is checked and incremented and returned back to the client. This enables potential rewrites for the same key, or even any overwrites.
3. The cache is fault tolerant because, if one of the primary nodes fail, the replicas take over steps up through our consistent hashing mechanism. During this transition period, if a particular primary node is not available, we fetch the data from the replica. If it is a write, alternately, we can set up the communication between the primary and the replicas as a gossip protocol so that either of the replicas translate the write to the new primary.
4. For keys which are hot - we can replicate the hot keys in other nodes or potentially salt the hot keys.
5. Since we are using versioning, we could potentially run in to wrong updates if the requests time out for response. So we can use a client generated request id or something - unique for a given write - which will be used only once for a given write.