feat: Implement TTL and expiration support for Redis commands

Author: iamd3vil

README.md

@@ -40,6 +40,7 @@ Blobasaur is a high-performance, sharded blob storage server written in Rust. It
   - [Storage Compression](#storage-compression-1)
 - [Advanced Topics](#advanced-topics)
   - [Database Schema](#database-schema)
+  - [TTL and Key Expiration](#ttl-and-key-expiration)
   - [Race Condition Handling](#race-condition-handling)
   - [Redis Cluster Compatibility](#redis-cluster-compatibility)
 - [Development](#development)
@@ -57,6 +58,7 @@ Blobasaur is a high-performance, sharded blob storage server written in Rust. It
 - **🗜️ Storage Compression**: Configurable compression with multiple algorithms (Gzip, Zstd, Lz4, Brotli)
 - **📊 Namespacing**: Hash-based namespacing using `HGET`/`HSET` commands for logical data organization
 - **🏷️ Metadata Tracking**: Automatic tracking of creation time, update time, expiration, and version numbers
+- **⏰ TTL Support**: Redis-compatible key expiration with `SET EX/PX`, `TTL`, and `EXPIRE` commands plus automatic background cleanup
 - **🌐 Redis Cluster Support**: Full cluster protocol support with automatic node discovery and client redirection
 - **⚙️ Highly Configurable**: Flexible configuration for shards, compression, batching, and performance tuning
 
@@ -159,6 +161,11 @@ just build-linux
    ```bash
    redis-cli -p 6379 SET mykey "Hello, World!"
    redis-cli -p 6379 GET mykey
+   
+   # Test TTL functionality
+   redis-cli -p 6379 SET session:123 "user_data" EX 60  # Expires in 60 seconds
+   redis-cli -p 6379 TTL session:123                    # Check remaining time
+   redis-cli -p 6379 EXPIRE mykey 300                   # Set 5 minute expiration
    ```
 
 ## CLI Usage
@@ -260,12 +267,19 @@ level = 1
 
 Blobasaur implements core Redis commands for blob operations:
 
-- **`SET key value`**: Store or replace a blob
+- **`SET key value [EX seconds] [PX milliseconds]`**: Store or replace a blob with optional TTL
   ```bash
+  # Basic SET
   redis-cli SET mykey "Hello, World!"
+  
+  # SET with TTL in seconds
+  redis-cli SET mykey "Hello, World!" EX 60
+  
+  # SET with TTL in milliseconds  
+  redis-cli SET mykey "Hello, World!" PX 30000
   ```
 
-- **`GET key`**: Retrieve a blob
+- **`GET key`**: Retrieve a blob (excludes expired keys)
   ```bash
   redis-cli GET mykey
   ```
@@ -275,11 +289,28 @@ Blobasaur implements core Redis commands for blob operations:
   redis-cli DEL mykey
   ```
 
-- **`EXISTS key`**: Check if a blob exists
+- **`EXISTS key`**: Check if a blob exists (excludes expired keys)
   ```bash
   redis-cli EXISTS mykey
   ```
 
+- **`TTL key`**: Get the remaining time to live for a key
+  ```bash
+  redis-cli TTL mykey
+  # Returns:
+  # -1 if key exists but has no expiration
+  # -2 if key does not exist or has expired
+  # positive number: remaining TTL in seconds
+  ```
+
+- **`EXPIRE key seconds`**: Set expiration time for an existing key
+  ```bash
+  redis-cli EXPIRE mykey 120  # Expire in 2 minutes
+  # Returns:
+  # 1 if expiration was set successfully
+  # 0 if key does not exist
+  ```
+
 ### Namespaced Commands
 
 Use namespaces to organize data into logical groups:
@@ -461,6 +492,32 @@ CREATE TABLE blobs (
 - Same schema as default table
 - Isolated from other namespaces
 
+### TTL and Key Expiration
+
+**Features:**
+- **Redis-Compatible TTL**: Full support for `SET EX/PX`, `TTL`, and `EXPIRE` commands
+- **Automatic Expiry**: All read operations (`GET`, `EXISTS`, `TTL`) automatically filter expired keys
+- **Background Cleanup**: Per-shard cleanup tasks run every 60 seconds to remove expired keys
+- **Efficient Storage**: Uses indexed `expires_at` timestamps for fast expiry queries
+
+**Implementation Details:**
+- Expiration timestamps stored as Unix epoch seconds in `expires_at` column
+- Database indexes on `expires_at` for efficient cleanup queries  
+- Background cleanup processes both main `blobs` table and namespaced tables
+- Race-free expiry checking: keys are considered expired at query time
+
+**Usage Examples:**
+```bash
+# Set with 60 second expiration
+redis-cli SET session:abc123 "user_data" EX 60
+
+# Check remaining TTL  
+redis-cli TTL session:abc123
+
+# Add expiration to existing key
+redis-cli EXPIRE permanent_key 3600
+```
+
 ### Race Condition Handling
 
 **The Problem:** Async writes could cause GET requests to miss recently SET data.

src/main.rs

@@ -160,6 +160,17 @@ async fn run_server(config_path: &str) -> Result<()> {
         ));
     }
 
+    // Spawn cleanup tasks for each shard
+    let cleanup_interval_secs = 60; // Clean up expired keys every 60 seconds
+    for i in 0..cfg.num_shards {
+        let pool = shared_state.db_pools[i].clone();
+        tokio::spawn(shard_manager::shard_cleanup_task(
+            i,
+            pool,
+            cleanup_interval_secs,
+        ));
+    }
+
     // Start HTTP metrics server if enabled
     if let Some(handle) = prometheus_handle {
         let metrics_addr = cfg

src/redis/integration_tests.rs

@@ -81,7 +81,8 @@ mod integration_tests {
             parsed_command,
             RedisCommand::Set {
                 key: "mykey".to_string(),
-                value: Bytes::from_static(b"hello world")
+                value: Bytes::from_static(b"hello world"),
+                ttl_seconds: None
             }
         );
     }
@@ -220,9 +221,10 @@ mod integration_tests {
 
             match parsed_command {
                 RedisCommand::Get { key } => assert_eq!(key, "key1"),
-                RedisCommand::Set { key, value } => {
+                RedisCommand::Set { key, value, ttl_seconds } => {
                     assert_eq!(key, "key2");
                     assert_eq!(value, Bytes::from_static(b"value"));
+                    assert_eq!(ttl_seconds, None);
                 }
                 RedisCommand::Del { key } => assert_eq!(key, "key3"),
                 _ => panic!("Unexpected command"),
@@ -241,9 +243,10 @@ mod integration_tests {
         let parsed_command = parse_command(parsed_resp).unwrap();
 
         match parsed_command {
-            RedisCommand::Set { key, value } => {
+            RedisCommand::Set { key, value, ttl_seconds } => {
                 assert_eq!(key, "binary");
                 assert_eq!(value, Bytes::copy_from_slice(binary_data));
+                assert_eq!(ttl_seconds, None);
             }
             _ => panic!("Expected SET command"),
         }
@@ -267,9 +270,10 @@ mod integration_tests {
         let parsed_command = parse_command(parsed_resp).unwrap();
 
         match parsed_command {
-            RedisCommand::Set { key, value } => {
+            RedisCommand::Set { key, value, ttl_seconds } => {
                 assert_eq!(key, large_key);
                 assert_eq!(value, Bytes::from(large_value.into_bytes()));
+                assert_eq!(ttl_seconds, None);
             }
             _ => panic!("Expected SET command"),
         }
@@ -283,9 +287,10 @@ mod integration_tests {
         let parsed_command = parse_command(parsed_resp).unwrap();
 
         match parsed_command {
-            RedisCommand::Set { key, value } => {
+            RedisCommand::Set { key, value, ttl_seconds } => {
                 assert_eq!(key, "empty");
                 assert_eq!(value, Bytes::new());
+                assert_eq!(ttl_seconds, None);
             }
             _ => panic!("Expected SET command"),
         }
@@ -415,6 +420,105 @@ mod integration_tests {
         assert!(matches!(result, Err(ParseError::Invalid(_))));
     }
 
+    #[tokio::test]
+    async fn test_set_with_ttl_integration() {
+        // Test SET with EX option
+        let set_ex_command = b"*5\r\n$3\r\nSET\r\n$8\r\nttl_test\r\n$5\r\nvalue\r\n$2\r\nEX\r\n$2\r\n60\r\n";
+        let (parsed_resp, _) = parse_resp_with_remaining(set_ex_command).unwrap();
+        let parsed_command = parse_command(parsed_resp).unwrap();
+        match parsed_command {
+            RedisCommand::Set { key, value, ttl_seconds } => {
+                assert_eq!(key, "ttl_test");
+                assert_eq!(value, Bytes::from_static(b"value"));
+                assert_eq!(ttl_seconds, Some(60));
+            }
+            _ => panic!("Expected SET command with TTL"),
+        }
+
+        // Test SET with PX option
+        let set_px_command = b"*5\r\n$3\r\nSET\r\n$8\r\nttl_test\r\n$5\r\nvalue\r\n$2\r\nPX\r\n$5\r\n60000\r\n";
+        let (parsed_resp, _) = parse_resp_with_remaining(set_px_command).unwrap();
+        let parsed_command = parse_command(parsed_resp).unwrap();
+        match parsed_command {
+            RedisCommand::Set { key, value, ttl_seconds } => {
+                assert_eq!(key, "ttl_test");
+                assert_eq!(value, Bytes::from_static(b"value"));
+                assert_eq!(ttl_seconds, Some(60)); // 60000ms = 60s
+            }
+            _ => panic!("Expected SET command with TTL"),
+        }
+    }
+
+    #[tokio::test]
+    async fn test_ttl_command_integration() {
+        let ttl_command = b"*2\r\n$3\r\nTTL\r\n$7\r\nmykey42\r\n";
+        let mock_response = b":-1\r\n"; // Key exists but no expiration
+
+        let port = create_mock_server(vec![mock_response.to_vec()]).await;
+        let addr = format!("127.0.0.1:{}", port);
+
+        let response = send_command_and_receive(&addr, ttl_command).await;
+        assert_eq!(response, mock_response);
+
+        // Parse the command to verify our parser works
+        let (parsed_resp, _) = parse_resp_with_remaining(ttl_command).unwrap();
+        let parsed_command = parse_command(parsed_resp).unwrap();
+        assert_eq!(
+            parsed_command,
+            RedisCommand::Ttl {
+                key: "mykey42".to_string()
+            }
+        );
+    }
+
+    #[tokio::test]
+    async fn test_expire_command_integration() {
+        let expire_command = b"*3\r\n$6\r\nEXPIRE\r\n$7\r\nmykey42\r\n$3\r\n120\r\n";
+        let mock_response = b":1\r\n"; // Key exists and expiration was set
+
+        let port = create_mock_server(vec![mock_response.to_vec()]).await;
+        let addr = format!("127.0.0.1:{}", port);
+
+        let response = send_command_and_receive(&addr, expire_command).await;
+        assert_eq!(response, mock_response);
+
+        // Parse the command to verify our parser works
+        let (parsed_resp, _) = parse_resp_with_remaining(expire_command).unwrap();
+        let parsed_command = parse_command(parsed_resp).unwrap();
+        assert_eq!(
+            parsed_command,
+            RedisCommand::Expire {
+                key: "mykey42".to_string(),
+                seconds: 120
+            }
+        );
+    }
+
+    #[tokio::test]
+    async fn test_ttl_edge_cases() {
+        // Test TTL for non-existent key
+        let ttl_command = b"*2\r\n$3\r\nTTL\r\n$11\r\nnonexistent\r\n";
+        let mock_response = b":-2\r\n"; // Key doesn't exist
+
+        let port = create_mock_server(vec![mock_response.to_vec()]).await;
+        let addr = format!("127.0.0.1:{}", port);
+
+        let response = send_command_and_receive(&addr, ttl_command).await;
+        assert_eq!(response, mock_response);
+    }
+
+    #[tokio::test]
+    async fn test_expire_on_nonexistent_key() {
+        let expire_command = b"*3\r\n$6\r\nEXPIRE\r\n$11\r\nnonexistent\r\n$2\r\n60\r\n";
+        let mock_response = b":0\r\n"; // Key doesn't exist
+
+        let port = create_mock_server(vec![mock_response.to_vec()]).await;
+        let addr = format!("127.0.0.1:{}", port);
+
+        let response = send_command_and_receive(&addr, expire_command).await;
+        assert_eq!(response, mock_response);
+    }
+
     #[tokio::test]
     async fn test_case_insensitive_commands() {
         let test_cases = vec![