From 172af25e8eb1ca1dc3bf5432f0b70576604a6487 Mon Sep 17 00:00:00 2001 From: unikdahal Date: Tue, 10 Feb 2026 23:02:51 +0530 Subject: [PATCH 1/4] Replication Added --- .github/copilot-instructions.md | 103 ++- README.md | 480 +++++++++++-- .../java/com/redis/commands/ICommand.java | 90 +++ .../redis/commands/generic/DelCommand.java | 5 + .../redis/commands/generic/ExpireCommand.java | 63 ++ .../commands/generic/PExpireAtCommand.java | 60 ++ .../com/redis/commands/list/BLPopCommand.java | 5 + .../com/redis/commands/list/LPopCommand.java | 5 + .../com/redis/commands/list/LPushCommand.java | 5 + .../com/redis/commands/list/RPushCommand.java | 5 + .../commands/replication/InfoCommand.java | 116 ++++ .../commands/replication/PsyncCommand.java | 156 +++++ .../commands/replication/ReplconfCommand.java | 154 +++++ .../commands/replication/WaitCommand.java | 84 +++ .../redis/commands/stream/XAddCommand.java | 82 +++ .../redis/commands/string/IncrCommand.java | 5 + .../com/redis/commands/string/SetCommand.java | 107 +++ .../java/com/redis/config/RedisConfig.java | 128 +++- .../redis/replication/CommandPropagator.java | 189 ++++++ .../redis/replication/MasterConnection.java | 637 +++++++++++++++++ .../com/redis/replication/RdbGenerator.java | 203 ++++++ .../redis/replication/ReplicaConnection.java | 337 +++++++++ .../com/redis/replication/ReplicationLog.java | 368 ++++++++++ .../redis/replication/ReplicationManager.java | 638 ++++++++++++++++++ .../com/redis/replication/ServerRole.java | 58 ++ .../redis/replication/SnapshotProducer.java | 341 ++++++++++ .../com/redis/server/NettyRedisServer.java | 88 ++- .../com/redis/server/RedisCommandHandler.java | 63 ++ .../java/com/redis/storage/RedisDatabase.java | 42 ++ .../java/com/redis/storage/RedisValue.java | 106 ++- .../services/com.redis.commands.ICommand | 5 + .../generic/PExpireAtCommandTest.java | 99 +++ .../integration/ReplicationCommandsIT.java | 254 +++++++ .../replication/CommandPropagatorTest.java | 291 ++++++++ .../MasterReplicaIntegrationTest.java | 638 ++++++++++++++++++ .../redis/replication/ReplicationLogTest.java | 339 ++++++++++ .../replication/ReplicationScaleTest.java | 546 +++++++++++++++ .../replication/SnapshotProducerTest.java | 219 ++++++ test_replication.sh | 393 +++++++++++ 39 files changed, 7411 insertions(+), 96 deletions(-) create mode 100644 src/main/java/com/redis/commands/generic/PExpireAtCommand.java create mode 100644 src/main/java/com/redis/commands/replication/InfoCommand.java create mode 100644 src/main/java/com/redis/commands/replication/PsyncCommand.java create mode 100644 src/main/java/com/redis/commands/replication/ReplconfCommand.java create mode 100644 src/main/java/com/redis/commands/replication/WaitCommand.java create mode 100644 src/main/java/com/redis/replication/CommandPropagator.java create mode 100644 src/main/java/com/redis/replication/MasterConnection.java create mode 100644 src/main/java/com/redis/replication/RdbGenerator.java create mode 100644 src/main/java/com/redis/replication/ReplicaConnection.java create mode 100644 src/main/java/com/redis/replication/ReplicationLog.java create mode 100644 src/main/java/com/redis/replication/ReplicationManager.java create mode 100644 src/main/java/com/redis/replication/ServerRole.java create mode 100644 src/main/java/com/redis/replication/SnapshotProducer.java create mode 100644 src/test/java/com/redis/commands/generic/PExpireAtCommandTest.java create mode 100644 src/test/java/com/redis/integration/ReplicationCommandsIT.java create mode 100644 src/test/java/com/redis/replication/CommandPropagatorTest.java create mode 100644 src/test/java/com/redis/replication/MasterReplicaIntegrationTest.java create mode 100644 src/test/java/com/redis/replication/ReplicationLogTest.java create mode 100644 src/test/java/com/redis/replication/ReplicationScaleTest.java create mode 100644 src/test/java/com/redis/replication/SnapshotProducerTest.java create mode 100755 test_replication.sh diff --git a/.github/copilot-instructions.md b/.github/copilot-instructions.md index e1b88b9..9437f01 100644 --- a/.github/copilot-instructions.md +++ b/.github/copilot-instructions.md @@ -2,7 +2,7 @@ ## High-Level Overview -This repository contains a lightweight, high-performance, in-memory Redis-compatible server built with **Java 25** and **Netty**. The project implements a subset of Redis commands (SET, GET, DEL) using the Redis Serialization Protocol (RESP) for communication. It's designed for simplicity, performance, and educational purposes. +This repository contains a lightweight, high-performance, in-memory Redis-compatible server built with **Java 25** and **Netty**. The project implements a comprehensive subset of Redis commands using the Redis Serialization Protocol (RESP) for communication. It features full master-replica replication with deterministic command propagation. **Key Technologies:** - Java 25 with preview features enabled @@ -11,6 +11,14 @@ This repository contains a lightweight, high-performance, in-memory Redis-compat - JUnit 5 and Mockito for testing - RESP (Redis Serialization Protocol) for client-server communication +**Supported Command Categories:** +- String operations: SET, GET, INCR +- List operations: LPUSH, RPUSH, LPOP, LLEN, LRANGE, BLPOP +- Stream operations: XADD, XRANGE, XREAD +- Transactions: MULTI, EXEC, DISCARD +- Key management: DEL, EXPIRE, TTL, TYPE +- Replication: REPLCONF, PSYNC, INFO + ## Build Instructions ### Prerequisites @@ -79,28 +87,43 @@ redis-java/ │ │ ├── commands/ # Command implementations │ │ │ ├── ICommand.java # Command interface │ │ │ ├── CommandRegistry.java # Command lookup registry -│ │ │ ├── SetCommand.java # SET command implementation -│ │ │ ├── GetCommand.java # GET command implementation -│ │ │ └── DelCommand.java # DEL command implementation +│ │ │ ├── string/ # String commands (SET, GET, INCR) +│ │ │ ├── list/ # List commands (LPUSH, RPUSH, LPOP, etc.) +│ │ │ ├── stream/ # Stream commands (XADD, XRANGE, XREAD) +│ │ │ ├── generic/ # Generic commands (DEL, EXPIRE, TYPE, etc.) +│ │ │ ├── transaction/ # Transaction commands (MULTI, EXEC, DISCARD) +│ │ │ └── replication/ # Replication commands (REPLCONF, PSYNC, INFO) │ │ ├── config/ │ │ │ └── RedisConfig.java # Server configuration │ │ ├── server/ │ │ │ ├── NettyRedisServer.java # Main server class (entry point) -│ │ │ └── RedisCommandHandler.java # RESP protocol handler +│ │ │ └── RedisCommandHandler.java # RESP protocol handler + replica write protection │ │ ├── storage/ │ │ │ ├── RedisDatabase.java # In-memory data store (ConcurrentHashMap) +│ │ │ ├── RedisValue.java # Type-safe value wrapper (sealed interface) │ │ │ └── ExpiryManager.java # Key expiration manager (DelayQueue) +│ │ ├── replication/ +│ │ │ ├── ReplicationManager.java # Central replication state and operations +│ │ │ ├── ReplicationLog.java # Lock-free ring buffer for partial resync +│ │ │ ├── SnapshotProducer.java # RDB snapshot generation for full resync +│ │ │ ├── CommandPropagator.java # Command canonicalization and propagation +│ │ │ ├── ReplicaConnection.java # Per-replica connection state +│ │ │ ├── MasterConnection.java # Replica-to-master connection handler +│ │ │ ├── RdbGenerator.java # RDB file format generation +│ │ │ └── ServerRole.java # MASTER/SLAVE enum +│ │ ├── transaction/ +│ │ │ └── TransactionContext.java # Per-connection transaction state │ │ └── util/ -│ │ └── ExpiryTask.java # Expiry task implementation +│ │ ├── ExpiryTask.java # Expiry task implementation +│ │ └── StreamId.java # Stream entry ID handling │ ├── main/test/test.sh # Manual integration test script │ └── test/java/com/redis/ │ ├── commands/ # Unit tests for commands -│ │ ├── SetCommandTest.java -│ │ ├── GetCommandTest.java -│ │ ├── DelCommandTest.java -│ │ └── GetDelCommandTest.java +│ ├── replication/ # Replication unit tests +│ │ ├── ReplicationLogTest.java +│ │ └── SnapshotProducerTest.java +│ ├── integration/ # Integration tests │ └── storage/ # Unit tests for storage -│ └── RedisDatabaseTest.java └── target/ └── redis-server.jar # Executable JAR (generated after build) ``` @@ -111,9 +134,61 @@ The server follows a layered architecture: 1. **Network Layer (Netty):** `NettyRedisServer` accepts connections using Netty's boss and worker thread pools 2. **Protocol Layer:** `RedisCommandHandler` parses RESP protocol and delegates to command registry -3. **Command Layer:** Individual command implementations (`SetCommand`, `GetCommand`, `DelCommand`) in `commands/` +3. **Command Layer:** Individual command implementations in `commands/` subdirectories (string/, list/, stream/, generic/, transaction/, replication/) 4. **Storage Layer:** `RedisDatabase` manages in-memory key-value storage with `ConcurrentHashMap` 5. **Expiration Layer:** `ExpiryManager` handles TTL-based key expiration using `DelayQueue` +6. **Replication Layer:** `ReplicationManager` + `CommandPropagator` + `ReplicationLog` + `SnapshotProducer` handle master-replica replication + +### Replication Architecture (Production-Grade) + +The replication system implements a **single-writer, log-based state machine** with snapshot checkpoints: + +``` +parse → validate → canonicalize → execute → append to log → send to replicas +``` + +**Core Roles:** +| Role | Responsibilities | +|:-----|:-----------------| +| **Master** | Accepts writes, executes commands, owns replication log, propagates to replicas | +| **Replica** | Rejects writes (READONLY), replays commands, tracks offset, requests PSYNC | +| **SnapshotProducer** | Creates point-in-time state snapshots independent of live mutations | + +**Key Components:** +- `ReplicationManager` - Manages master/replica roles, replica connections, offset tracking, circuit breakers +- `ReplicationLog` - Lock-free ring buffer (64KB-512MB) for partial resync support +- `SnapshotProducer` - Generates RDB snapshots for full resync +- `CommandPropagator` - Determines which commands to propagate and handles RESP encoding +- `ReplicaConnection` - Per-replica state tracking with backpressure handling +- `ICommand.getReplicationArgs()` - Commands can override to provide canonical arguments +- `ICommand.getReplicationCommandName()` - Commands can override to use different command for replication + +**Optimizations Over Redis:** +| Feature | Redis | This Implementation | +|:--------|:------|:--------------------| +| Offset Tracking | Mutex-protected long | Lock-free `AtomicLong` | +| Statistics | Atomic increments | `LongAdder` (10x faster) | +| Backlog | Single producer lock | Lock-free ring buffer | +| Propagation | Synchronous per-replica | Async with circuit breaker | +| WAIT Polling | Fixed interval | Adaptive exponential backoff | +| Failure Handling | Simple disconnect | Circuit breaker + recovery | + +**Command Canonicalization:** +Non-deterministic commands are rewritten before replication: +- `XADD stream * field value` → `XADD stream field value` +- `SET key value EX 60` → `SET key value PXAT ` +- `EXPIRE key 60` → `PEXPIREAT key ` + +**Full Sync vs Partial Sync Decision:** +``` +ID mismatch → FULL SYNC +offset not in backlog → FULL SYNC +else → PARTIAL SYNC +``` +No heuristics. No guessing. The decision is deterministic. + +**Write Protection on Replicas:** +Replicas automatically reject write commands with `-READONLY` error. This is enforced in `RedisCommandHandler`. ## Key Development Guidelines @@ -135,6 +210,10 @@ The server follows a layered architecture: - Implement `ICommand` interface - Register in `CommandRegistry` - Add unit tests in `src/test/java/com/redis/commands/` + - For write commands: + - Override `isWriteCommand()` to return `true` + - If command has non-deterministic behavior, override `getReplicationArgs()` to return canonical arguments + - If command should replicate as different command, override `getReplicationCommandName()` 5. **Testing:** - Always run `mvn test` before committing diff --git a/README.md b/README.md index f77e001..4d47600 100644 --- a/README.md +++ b/README.md @@ -3,20 +3,51 @@ [![Java Version](https://img.shields.io/badge/Java-25-orange.svg)](https://openjdk.java.net/) [![License](https://img.shields.io/badge/License-MIT-blue.svg)](LICENSE) [![Java CI](https://github.com/unikdahal/redis-java/actions/workflows/ci.yml/badge.svg)](https://github.com/unikdahal/redis-java/actions/workflows/ci.yml) +[![Tests](https://img.shields.io/badge/Tests-599%2B-brightgreen.svg)](#-testing) -A high-performance, lightweight, in-memory Redis-compatible server built from the ground up using **Java 25** and **Netty**. This project implements core Redis functionality with a focus on low latency, efficient memory usage, and clean architecture. +A high-performance, lightweight, in-memory Redis-compatible server built from the ground up using **Java 25** and **Netty**. This project implements core Redis functionality with a focus on low latency, efficient memory usage, clean architecture, and production-grade replication. --- ## ✨ Features -- **⚡ High Performance**: Built on Netty's asynchronous event-driven framework for massive concurrency. -- **💾 In-Memory Storage**: Optimized data structures using `ConcurrentHashMap` for thread-safe, lock-free reads. -- **🔌 Redis Protocol (RESP)**: Implements the Redis Serialization Protocol, compatible with any standard Redis client (`redis-cli`, `jedis`, `redis-py`, etc.). -- **⏳ Advanced Expiration**: Dual-strategy expiration (Lazy + Active background cleanup via `DelayQueue`). -- **🔄 Transaction Support**: Full MULTI/EXEC/DISCARD support with optimized batch execution and zero-contention state management. -- **🎯 Single-Threaded Execution**: Primarily single-threaded command execution for predictable behavior; blocking commands (e.g., `BLPOP`) are handled asynchronously using Netty's event loop to avoid blocking I/O. -- **🏗️ Extensible Command Registry**: Easy to add new commands via a simple interface. +### Core Capabilities +- **⚡ High Performance**: Built on Netty's asynchronous event-driven framework for massive concurrency +- **💾 In-Memory Storage**: Optimized data structures using `ConcurrentHashMap` for thread-safe, lock-free reads +- **🔌 Redis Protocol (RESP)**: Full RESP implementation, compatible with any standard Redis client (`redis-cli`, `jedis`, `lettuce`, `redis-py`, etc.) +- **⏳ Advanced Expiration**: Dual-strategy expiration (Lazy + Active background cleanup via `DelayQueue`) +- **🔄 Transaction Support**: Full MULTI/EXEC/DISCARD support with optimized batch execution +- **📊 Pipelining**: Full support for command pipelining with efficient batch processing + +### Replication (Master-Replica) +- **🔗 Master-Replica Architecture**: Production-grade single-writer, log-based state machine +- **📡 Command Propagation**: Automatic propagation with backpressure handling +- **🔄 Deterministic Replication**: Commands are canonicalized before replication to ensure consistent state: + - `XADD stream *` → `XADD stream ` (auto-generated IDs resolved) + - `SET key value EX 60` → `SET key value PXAT ` (relative → absolute time) + - `EXPIRE key 60` → `PEXPIREAT key ` (relative → absolute time) +- **📦 Partial Resync (PSYNC2)**: Efficient reconnection with backlog-based partial resync +- **💾 RDB Snapshots**: Point-in-time snapshots for full resync +- **🚫 Write Protection**: Replicas automatically reject write commands (READONLY error) +- **📊 Replication Log**: Lock-free ring buffer with configurable size (64KB - 512MB) + +### Advanced Features +- **🌊 Redis Streams**: Full XADD/XRANGE/XREAD support with blocking reads +- **🚫 Blocking Operations**: BLPOP with efficient async handling (no thread blocking) +- **🏗️ Extensible Command Registry**: Easy to add new commands via simple interface + +### 🚀 Key Differentiators from Redis + +| Feature | Redis | This Implementation | +|:--------|:------|:--------------------| +| **Offset Tracking** | Mutex-protected long | Lock-free `AtomicLong` | +| **Statistics** | Atomic increments | `LongAdder` (10x faster under contention) | +| **Backlog** | Single producer lock | Lock-free ring buffer with minimal contention | +| **Propagation** | Synchronous per-replica | Async with circuit breaker pattern | +| **WAIT Polling** | Fixed interval | Adaptive exponential backoff | +| **Memory** | Unbounded backlog growth | Bounded with automatic eviction | +| **Failure Handling** | Simple disconnect | Circuit breaker + self-healing recovery | +| **Health Monitoring** | Basic lag tracking | Per-replica health scores & backpressure detection | --- @@ -25,45 +56,52 @@ A high-performance, lightweight, in-memory Redis-compatible server built from th Detailed documentation for each command can be found in the [docs/commands](./docs/commands) directory. ### 🔑 Connection & Utility -| Command | Usage | Documentation | -|:---|:---|:---| -| `PING` | `PING [message]` | [PING.md](./docs/commands/PING.md) | -| `ECHO` | `ECHO message` | [ECHO.md](./docs/commands/ECHO.md) | -| `EXPIRE` | `EXPIRE key seconds` | [EXPIRE.md](./docs/commands/EXPIRE.md) | -| `TTL` | `TTL key` | [TTL.md](./docs/commands/TTL.md) | -| `TYPE` | `TYPE key` | [TYPE.md](./docs/commands/TYPE.md) | - -### 📝 Key-Value Operations -| Command | Usage | Documentation | -|:---|:---|:---| -| `SET` | `SET key value [EX s] [PX ms] [NX\|XX]` | [SET.md](./docs/commands/SET.md) | -| `GET` | `GET key` | [GET.md](./docs/commands/GET.md) | -| `DEL` | `DEL key [key ...]` | [DEL.md](./docs/commands/DEL.md) | -| `INCR` | `INCR key` | [INCR.md](./docs/commands/INCR.md) | +| Command | Usage | Description | +|:--------|:------|:------------| +| `PING` | `PING [message]` | Test connection, returns PONG or echoes message | +| `ECHO` | `ECHO message` | Echo the given message | +| `EXPIRE` | `EXPIRE key seconds` | Set key expiration in seconds | +| `TTL` | `TTL key` | Get remaining time to live in seconds | +| `TYPE` | `TYPE key` | Get the type of value stored at key | +| `DEL` | `DEL key [key ...]` | Delete one or more keys | + +### 📝 String Operations +| Command | Usage | Description | +|:--------|:------|:------------| +| `SET` | `SET key value [EX s] [PX ms] [PXAT ms] [NX\|XX]` | Set string value with optional expiry | +| `GET` | `GET key` | Get the value of a key | +| `INCR` | `INCR key` | Increment integer value by 1 | ### 🔄 Transactions -| Command | Usage | Documentation | -|:---|:---|:---| -| `MULTI` | `MULTI` | [MULTI.md](./docs/commands/MULTI.md) | -| `EXEC` | `EXEC` | [EXEC.md](./docs/commands/EXEC.md) | -| `DISCARD` | `DISCARD` | [DISCARD.md](./docs/commands/DISCARD.md) | +| Command | Usage | Description | +|:--------|:------|:------------| +| `MULTI` | `MULTI` | Start a transaction block | +| `EXEC` | `EXEC` | Execute all commands in transaction | +| `DISCARD` | `DISCARD` | Discard all commands in transaction | ### 📋 List Operations -| Command | Usage | Documentation | -|:---|:---|:---| -| `LPUSH` | `LPUSH key element [element ...]` | [LPUSH.md](./docs/commands/LPUSH.md) | -| `RPUSH` | `RPUSH key element [element ...]` | [RPUSH.md](./docs/commands/RPUSH.md) | -| `LPOP` | `LPOP key [count]` | [LPOP.md](./docs/commands/LPOP.md) | -| `LLEN` | `LLEN key` | [LLEN.md](./docs/commands/LLEN.md) | -| `LRANGE` | `LRANGE key start stop` | [LRANGE.md](./docs/commands/LRANGE.md) | -| `BLPOP` | `BLPOP key [key ...] timeout` | [BLPOP.md](./docs/commands/BLPOP.md) | +| Command | Usage | Description | +|:--------|:------|:------------| +| `LPUSH` | `LPUSH key element [element ...]` | Push elements to head of list | +| `RPUSH` | `RPUSH key element [element ...]` | Push elements to tail of list | +| `LPOP` | `LPOP key [count]` | Pop elements from head of list | +| `LLEN` | `LLEN key` | Get list length | +| `LRANGE` | `LRANGE key start stop` | Get range of elements | +| `BLPOP` | `BLPOP key [key ...] timeout` | Blocking pop from head | ### 🌊 Stream Operations -| Command | Usage | Documentation | -|:---|:---|:---| -| `XADD` | `XADD key ID field value [field v ...]` | [XADD.md](./docs/commands/XADD.md) | -| `XRANGE` | `XRANGE key start end [COUNT c]` | [XRANGE.md](./docs/commands/XRANGE.md) | -| `XREAD` | `XREAD [COUNT c] [BLOCK ms] STREAMS k [k ...] id [id ...]` | [XREAD.md](./docs/commands/XREAD.md) | +| Command | Usage | Description | +|:--------|:------|:------------| +| `XADD` | `XADD key ID field value [field value ...]` | Append entry to stream | +| `XRANGE` | `XRANGE key start end [COUNT count]` | Get range of entries | +| `XREAD` | `XREAD [COUNT c] [BLOCK ms] STREAMS key [key ...] id [id ...]` | Read from streams | + +### 🔗 Replication Commands +| Command | Usage | Description | +|:--------|:------|:------------| +| `REPLCONF` | `REPLCONF