Implementing TTL for Key-Value Store

Hello again! Welcome back to our module on "Building Systems Incrementally."

In our last lesson, you built the foundation of an in-memory key-value store, implementing the core GET, SET, and DELETE operations. That simple API is the starting point for countless distributed systems.

Today, we'll add a crucial feature found in many real-world key-value systems, especially those used as caches: Time-To-Live (TTL). The goal is to make entries automatically expire after a specified duration. This is essential for managing memory, ensuring data freshness, and handling transient states like user sessions or temporary locks. Your experience with systems like Rudi, which manage time-sensitive requests and resource allocations, likely involved similar concepts of expiry and cleanup.

By the end of this lesson, you will be able to extend the key-value store to support Time-To-Live (TTL) with a strategy for evicting expired keys.

---

1. TTL in Production Systems: The "Why"

Before we dive into implementation, let's look at how a production system like Redis uses TTL. Redis is a high-performance in-memory key-value store, and its expiration features are one of its most common use cases.

As the video highlighted, TTL is not just for caches. It's a fundamental mechanism for automatic state management. To build on this, let's explore a few more examples.

---

2. Evolving the Data Structure for TTL

Our current store just holds value. To support TTL, we need to associate each value with an expiration timestamp. A clean, type-safe way to do this in Python is with a dataclass.

Let's define a CacheEntry:

from dataclasses import dataclass, field
import time

@dataclass
class CacheEntry:
    """A wrapper for a cached value that includes expiration metadata."""
    value: any
    expires_at: float

A critical design choice here is storing an absolute expiration timestamp (expires_at) rather than the relative TTL duration. This makes checking for expiration extremely efficient: it's a single comparison against the current time.

Another crucial detail is the choice of clock. We should use time.monotonic() instead of time.time(). A monotonic clock always moves forward and cannot be adjusted backward (e.g., by NTP syncs or daylight saving changes). This guarantees that our TTL calculations are robust and predictable.

With this new structure, our set method signature changes:

def set(self, key: str, value: any, ttl_seconds: float = None):
    """
    Sets the value for a given key with an optional TTL.

    Args:
        key: The key to set.
        value: The value to associate with the key.
        ttl_seconds: The time-to-live in seconds. If None, the key never expires.
    """
    if ttl_seconds is not None:
        expires_at = time.monotonic() + ttl_seconds
    else:
        expires_at = float('inf') # Represents never-expiring
    
    self._data[key] = CacheEntry(value=value, expires_at=expires_at)

And our get method must now check for expiration:

def get(self, key: str) -> any:
    entry = self._data.get(key)
    if entry is None:
        return None

    if time.monotonic() > entry.expires_at:



        # The key has expired.
        return None
    
    return entry.value

---

3. Eviction Strategies: Lazy vs. Active Cleanup

Our get method correctly avoids returning expired data. But what about the expired entry still sitting in our _data dictionary? If an expired key is never requested again, it will remain in memory forever. This is a memory leak.

We need an eviction strategy. There are two primary approaches.

Strategy 1: Lazy Eviction (Cleanup on Read)

This is the simplest strategy. When get finds an expired key, it deletes it before returning None.

# Modified get() with lazy eviction
def get(self, key: str) -> any:
    entry = self._data.get(key)
    if entry is None:
        return None

    if time.monotonic() > entry.expires_at:



        # The key has expired, so delete it and return None.
        del self._data[key]
        return None
    
    return entry.value

• Pro: Very simple to implement. No overhead when the cache is idle.
• Con: Expired items that are never accessed again are never removed, leading to a memory leak.

This strategy is often called "lazy expiration" because the cleanup work is deferred until the item is accessed.

Strategy 2: Active Eviction (Background Cleanup)

To solve the memory leak problem, we can introduce a background process that periodically scans the entire cache and removes any expired items. This process is often called a "janitor" or "reaper."

This approach is more robust and memory-efficient but introduces new complexity, especially around concurrency.

For a system like Anthropic's, which demands high reliability, understanding the concurrency implications of an active cleanup strategy is vital. A put operation could race with a cleanup operation on the same key. The AlgoMaster article previews these challenges in its "Concurrency Challenges" and "Sequence Diagram" sections (sections 3, 4, 8, 9), which will be the core focus of our next module on Concurrency.

---

4. Your Task: Implement TTL and Eviction

Now, let's update your KeyValueStore. You will implement TTL support with both lazy and active eviction mechanisms.

Here is the full class structure. Your task is to implement the get and cleanup methods. The set and delete methods are also provided for completeness.

import time
from dataclasses import dataclass

@dataclass
class CacheEntry:
    """A wrapper for a cached value that includes expiration metadata."""
    value: any
    expires_at: float

class KeyValueStore:
    """
    An in-memory key-value store with TTL support.
    """
    def __init__(self):
        """Initializes the key-value store."""
        self._data: dict[str, CacheEntry] = {}

    def set(self, key: str, value: any, ttl_seconds: float = None):
        """
        Sets the value for a given key with an optional TTL.

        Args:
            key: The key to set.
            value: The value to associate with the key.
            ttl_seconds: The time-to-live in seconds. If None, the key never expires.
        """
        if ttl_seconds is not None:



            # Use monotonic clock for robustness against system time changes
            expires_at = time.monotonic() + ttl_seconds
        else:



            # Use infinity to represent a key that never expires
            expires_at = float('inf')
        
        self._data[key] = CacheEntry(value=value, expires_at=expires_at)
        print(f"SET key='{key}', value='{value}', ttl={ttl_seconds}s")

    def get(self, key: str) -> any:
        """
        Gets the value for a given key, performing lazy eviction if expired.

        Args:
            key: The key to retrieve.
            
        Returns:
            The value associated with the key, or None if the key is not found or has expired.
        """



        # 1. Retrieve the CacheEntry from self._data.
        # 2. If the entry does not exist, return None.
        # 3. Check if the entry has expired using time.monotonic().
        # 4. If it has expired:
        #    a. Delete the key from self._data.
        #    b. Return None.
        # 5. If it has not expired, return the entry's value.
        pass # Your implementation here

    def delete(self, key: str) -> bool:
        """
        Deletes a key-value pair.
        
        Args:
            key: The key to delete.
            
        Returns:
            True if the key was found and deleted, False otherwise.
        """
        if key in self._data:
            del self._data[key]
            return True
        return False

    def cleanup(self):
        """
        Actively scans and removes all expired keys from the store.
        """

```grasp
{
  "type": "exercise",
  "id": "3c2f9b9e-9624-4ecb-94b7-4c98e7cc4093"
}

    # Note: Iterating and deleting from a dictionary in a single loop
    # requires care. A safe way is to first collect all keys to be
    # deleted, and then loop over that collection to delete them.
    



    # 1. Create a list of keys that have expired.
    #    - Iterate through self._data.items().
    #    - For each key and entry, if the entry is expired, add the key to your list.
    # 2. Iterate through your list of expired keys and delete each one from self._data.
    # 3. Print which keys were cleaned up for verification.
    pass # Your implementation here

def __len__(self) -> int:
    return len(self._data)

Example Usage

if name == 'main':
store = KeyValueStore()

# Set a key with a short TTL
store.set("session_token", "abc123xyz", ttl_seconds=2)



# Set a key that lasts longer
store.set("api_key", "secret-key", ttl_seconds=5)

print(f"\nInitial store size: {len(store)}")




# Get the value before it expires
value = store.get("session_token")
print(f"GET 'session_token': {value}") # Expected: abc123xyz




# Wait for the first key to expire
print("\nWaiting 3 seconds...")
time.sleep(3)




# Try to get the expired key (this should trigger lazy eviction)
value = store.get("session_token")
print(f"GET 'session_token' after 3s: {value}") # Expected: None
print(f"Store size after lazy eviction: {len(store)}") # Expected: 1




# Set another expiring key and one non-expiring key
store.set("temp_data", "data", ttl_seconds=1)
store.set("permanent_data", "important", ttl_seconds=None)

print(f"\nStore size before cleanup: {len(store)}") # Expected: 3




# Wait for 'temp_data' to expire
time.sleep(1.5)




# Run active cleanup
print("\nRunning active cleanup...")
store.cleanup()
print(f"Store size after cleanup: {len(store)}") # Expected: 2 (api_key, permanent_data)

**Bonus Question (For Reflection):**
The NeetCode video [LINK](https://www.youtube.com/watch?v=fu2cD_6E8Hw) on the "Time Based Key-Value Store" problem uses binary search to efficiently find a value at a specific timestamp. When would this be more useful than our simple `expires_at` check? Consider a scenario where you need to support retrieving the value of a key *as it was* at some point in the past (versioning), not just checking for expiry.

---




### Conclusion

Excellent work! You've taken a basic key-value store and layered on a critical production feature. This incremental approach—building a simple core and then adding complexity—is exactly how robust systems are developed and how you can expect to be tested in a system design interview.

**Key Takeaways:**

*   Supporting TTL requires augmenting your data structure to store an **absolute expiration timestamp** alongside the value.
*   Using a **monotonic clock** is essential for creating a robust TTL system that is immune to system time changes.
*   **Lazy eviction** (on `get`) is simple but can lead to memory leaks if expired keys are not accessed.
*   **Active eviction** (via a `cleanup` process) is more memory-efficient but introduces complexity, especially around concurrency. This trade-off between simplicity, performance, and memory usage is at the heart of system design.




### Next Lesson Preview

We've now spent two lessons building and extending a data-centric component. In the next lesson, we will shift focus to a logic-centric one: **a simple task scheduler**. You'll start by implementing a basic First-In-First-Out (FIFO) scheduler using a queue, which will serve as the foundation for adding more advanced features like task priorities and dependencies in subsequent lessons.